为什么当我开始测试服务器上的负载时，同步写入方式的工作速度比异步方式快数百倍。我只是尝试异步，我做错了什么？

ab -t 10 -n 10000 -c 100 http://127.0.0.1:8888/api/db/sync/ 每秒请求数：35

ab -t 10 -n 10000 -c 100 http://127.0.0.1:8888/api/db/ 每秒请求数：0.37

import asyncio
import json
import time
from asyncio import sleep
from typing import Optional, Union

import tornado.ioloop
import tornado.web

from jinja2 import Environment, select_autoescape, FileSystemLoader
from sqlalchemy import text, create_engine
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, Session
from tornado import gen
from tornado.httpserver import HTTPServer

env = Environment(
    loader=FileSystemLoader('templates'),
    autoescape=select_autoescape()
)

async_engine = create_async_engine(
    "postgresql+asyncpg://***:***@localhost:5431/***",
    echo=True,
)

async_session …

我有一个从async没有 的函数调用的函数await，并且我的函数需要调用async函数。我可以这样做asyncio.get_running_loop().create_task(sleep())，但run_until_complete在新任务完成之前，顶层不会运行。

如何让事件循环运行直到新任务完成？

我无法执行我的函数，async因为它不是用 调用的await。

我无法改变future或者sleep。我只能控制in_control。

import asyncio


def in_control(sleep):
    """
    How do I get this to run until complete?
    """
    return asyncio.get_running_loop().create_task(sleep())


async def future():
    async def sleep():
        await asyncio.sleep(10)
        print('ok')

    in_control(sleep)


asyncio.get_event_loop().run_until_complete(future())

好吧，这有点复杂，但我有一个带有很多异步代码的异步类。

我希望在该类中并行化一个任务，并且我想生成多个进程来运行阻塞任务，并且在每个进程中我想创建一个asyncio循环来处理各种子任务。

所以我无法使用 ThreadPollExecutor 来做到这一点，但是当我尝试使用 ProcessPoolExecutor 时，我收到一个 Can't pickle local object 错误。

这是我的代码的简化版本，与 ThreadPoolExecutor 一起运行。如何与 ProcessPoolExecutor 并行？

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


class MyClass:
    def __init__(self) -> None:
        self.event_loop = None
        # self.pool_executor = ProcessPoolExecutor(max_workers=8)
        self.pool_executor = ThreadPoolExecutor(max_workers=8)
        self.words = ["one", "two", "three", "four", "five"]
        self.multiplier = int(2)

async def subtask(self, letter: str):
    await asyncio.sleep(1)
    return letter * self.multiplier

async def task_gatherer(self, subtasks: list):
    return await asyncio.gather(*subtasks)

def blocking_task(self, word: str):
    time.sleep(1)
    subtasks …

我有一个 Django 应用程序，在其视图之一中，我使用 asyncio 来向外部组件发出一些并发请求。

这个想法是这样的：

import asyncio


async def do_request(project):
    result = ...

    return result

def aggregate_results(projects: list):
    loop = asyncio.new_event_loop()

    asyncio.set_event_loop(loop)

    results = loop.run_until_complete(
        asyncio.gather(*(do_request(project) for project in projects))
    )

    loop.close()

    return zip(projects, results)

好吧，当我运行测试时，我得到了DeprecationWarning: There is no current event loop这一行：

        asyncio.gather(*(do_request(project) for project in projects))

我应该如何解释此警告以及我需要进行哪些更改才能消除它？谢谢！

背景

我正在从 Python 调用可执行文件，并且需要将变量传递给可执行文件。然而，可执行文件需要一个文件，而不是从标准输入读取。

我之前在使用子进程模块时通过简单地调用可执行文件来读取/dev/stdin以下内容来规避这个问题：

# with executable 'foo'
cmd = ['foo', '/dev/stdin']
input_variable = 'bar'

with subprocess.Popen(
    cmd,
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    ) as process:
    stdout, stderr = process.communicate(input_variable)

    print(f"{process.returncode}, {stdout}, {stderr}")

到目前为止效果很好。为了添加并发性，我现在正在实现 asyncio，因此需要将 subprocess 模块替换为 asyncio subprocess 模块。

问题

使用程序调用 asyncio 子进程/dev/stdin失败。使用以下异步函数：

import asyncio

async def invoke_subprocess(cmd, args, input_variable):
    process = await asyncio.create_subprocess_exec(
        cmd,
        args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.PIPE,
    )

    stdout, stderr = await process.communicate(input=bytes(input_variable, 'utf-8'))

    print(f"{process.returncode}, {stdout.decode()}, {stderr.decode()}")

这通常适用于文件，但不适用于/dev/stdin：

# 'cat' can be …

非常简单的代码：

async def poll():
    await asyncio.sleep(1)

def caller(somefunction):
    somefunction() # we don't know if callable is async or not

def main():
    caller(poll) # how to pass an async function "poll" here?

我想通过将其作为可调用/lambda 传递给“调用者”函数来调用“poll”。通常，在设计良好的语言中，例如 C#，您会这样做poll().Result，因为 poll() 返回一个任务，该任务具有 Wait() 和 Result 等方法来阻止和等待，或者如果您想阻止和等待并返回，您可以使用“await”到您当前所在异步方法的调用者。

我正在使用 asyncio subprocess 来执行子命令。我想查看长时间运行的进程并同时将内容保存到缓冲区以供以后使用。此外，我发现了这个相关的问题（Getting live output from asyncio subprocess），但它主要围绕 ssh 的用例。

asyncio subprocess 文档有一个逐行读取输出的示例，这符合我想要实现的目标。（https://docs.python.org/3/library/asyncio-subprocess.html#examples）

import asyncio
import sys

async def get_date():
    code = 'import datetime; print(datetime.datetime.now())'

    # Create the subprocess; redirect the standard output
    # into a pipe.
    proc = await asyncio.create_subprocess_exec(
        sys.executable, '-c', code,
        stdout=asyncio.subprocess.PIPE)

    # Read one line of output.
    data = await proc.stdout.readline()
    line = data.decode('ascii').rstrip()

    # Wait for the subprocess exit.
    await proc.wait()
    return line

date = asyncio.run(get_date())
print(f"Current date: {date}")

我将此示例改编为以下内容： …

我是 Python asyncio 的新手，我正在做一些实验。我有以下代码：

async def say_after(n, s):
    await asyncio.sleep(n)
    print(s)

async def main():
    task1 = asyncio.create_task(say_after(2, 'a'))
    task2 = asyncio.create_task(say_after(1, 'b'))
    await task1
    print('x', flush=True)
    await task2
    print('y', flush=True)

asyncio.run(main())

和输出：

b
a
x
y

我不明白这里的顺序。有人可以帮忙解释一下吗？特别是为什么x出现在b和a之后？

我的项目包括使用构建在 aws lambda 服务之上的 api。从技术上讲，构建 api 的负责人告诉我，由于服务是弹性的，因此没有固定的请求限制，但重要的是要考虑 api 可以支持的每秒请求数。

为了控制每秒请求的限制（同时），我正在开发的 python 脚本使用 asyncio 和 httpx 来同时使用 api，并利用httpx.Limits的 max_connections 参数，我试图找到最佳值，以便API 不会冻结。

我的问题是，我不知道我是否误解了 max_connections 参数的使用，因为当使用值 1000 进行测试时，我的理解告诉我，每秒我会同时向 api 发出 1000 个请求，但即便如此， api 在一定时间后冻结。

我希望能够控制每秒请求的限制，而无需使用第三方库。

我怎样才能做到呢？

这是我的 MWE

async def consume(client, endpoint: str = '/create', reg):
  data = {"param1": reg[1]}

  response = await client.post(url=endpoint, data=json.dumps(data))

  return response.json()

async def run(self, regs):
  # Empty list to consolidate all responses
  results = []

  # httpx limits configuration
  limits = httpx.Limits(max_keepalive_connections=None, max_connections=1000)
  timeout = …

我正在尝试访问具有机器人防护功能的网站。

使用以下脚本使用请求我可以访问该网站。

request = requests.get(url,headers={**HEADERS,'Cookie': cookies})

我得到了所需的 HTML。但是当我使用 aiohttp 时

async def get_data(session: aiohttp.ClientSession,url,cookies):
    async with session.get(url,timeout = 5,headers={**HEADERS,'Cookie': cookies}) as response:
        text = await response.text()
        print(text)

我收到机器人预防页面作为响应。

这是我用于所有请求的标头。

HEADERS = {
    'User-Agent': 'PostmanRuntime/7.29.0',
    'Host': 'www.dnb.com',
    'Connection': 'keep-alive',
    'Accept': '/',
    'Accept-Encoding': 'gzip, deflate, br'
} 

我比较了 requests.get 和 aiohttp 的请求标头，它们是相同的。

结果不同有什么原因吗？如果是这样为什么？

编辑：我检查了 httpx 模块，问题也出现在httpx.Client()那里httpx.AsyncClient()。

response = httpx.request('GET',url,headers={**HEADERS,'Cookie':cookies})

效果不太好。（不是异步的）