相关疑难解决方法(0)

如何使用aiohttp在客户端设置每秒的最大请求数(限制它们)？

中的许多函数asyncio都有不推荐使用的loop参数，计划在 Python 3.10 中删除。实例包括as_completed()，sleep()，和wait()。

我正在寻找有关这些参数及其删除的一些历史背景。

• loop解决了哪些问题？一开始为什么要使用它？
• 出了什么问题loop？为什么要大量删除？
• 什么取代了loop，现在它消失了？

当我在 Python 3.7 中运行此代码时：

import asyncio

sem = asyncio.Semaphore(2)

async def work():
    async with sem:
        print('working')
        await asyncio.sleep(1)

async def main():
    await asyncio.gather(work(), work(), work())

asyncio.run(main())

它因运行时错误而失败：

$ python3 demo.py
working
working
Traceback (most recent call last):
  File "demo.py", line 13, in <module>
    asyncio.run(main())
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/opt/local/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "demo.py", line 11, in main
    await asyncio.gather(work(), work(), work())
  File "demo.py", line 6, in work
    async with sem:
  File …

我已经使用asyncio编写了我的脚本,但发现同时运行的协同程序的数量太大而且经常会出现问题.

所以我想同时限制协同程序的数量,一旦达到限制,我想等待任何协同程序在另一个程序执行之前完成.

我目前的代码如下:

loop = asyncio.get_event_loop()
p = map(my_func, players)
result = loop.run_until_complete(asyncio.gather(*p))

async def my_func(player):
    # something done with `await`

该players类型是list与包含大量元素(比如,12000).它需要大量的计算资源来同时运行所有这些,asyncio.gather(*p)所以我宁愿同时运行的玩家数量为200.一旦达到199,我希望另一个协程开始执行.

在asyncio中这可能吗？

我正在使用 Python Asyncio 来执行大量 HTTP 请求。

我想知道 Asyncio 的默认并发级别是多少，或者在任何给定时间会并行发生多少个 HTTP 请求？

我用来执行 HTTP 请求的代码如下：

async def call_url(self, session, url):
    response = await session.request(method='GET', url=url)
    return response


async def main(self, url_list):
    async with aiohttp.ClientSession() as session:
        res = await asyncio.gather(*[self.call_url(session, url) for url in url_list])
        return res

根据asyncio 同步原语，有多种同步方法。

• 我很困惑为什么我们需要异步同步？
• 我的意思是，asyncio 是异步的。
• 在异步中添加一些同步有意义吗？

我使用 OpenAI API。我从 PowerPoint 演示文稿中提取了幻灯片文本，并为每张幻灯片编写了提示。现在，我想进行异步 API 调用，以便同时处理所有幻灯片。

这是异步主函数的代码：

for prompt in prompted_slides_text:
    task = asyncio.create_task(api_manager.generate_answer(prompt))
    tasks.append(task)
results = await asyncio.gather(*tasks)

这是generate_answer函数：

@staticmethod
    async def generate_answer(prompt):
        """
        Send a prompt to OpenAI API and get the answer.
        :param prompt: the prompt to send.
        :return: the answer.
        """
        completion = await openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=[{"role": "user", "content": prompt}]

        )
        return completion.choices[0].message.content

问题是：

对象 OpenAIOb​​ject 不能在“await”表达式中使用

我不知道如何等待generate_answer函数中的响应

将不胜感激任何帮助！

简单示例:我需要并行创建两个不相关的HTTP请求.最简单的方法是什么？我希望它是这样的:

async def do_the_job():
    with aiohttp.ClientSession() as session:
        coro_1 = session.get('http://httpbin.org/get')
        coro_2 = session.get('http://httpbin.org/ip')
        return combine_responses(await coro_1, await coro_2)

换句话说,我想启动IO操作并等待其结果,以便它们有效地并行运行.这可以通过以下方式实现asyncio.gather:

async def do_the_job():
    with aiohttp.ClientSession() as session:
        coro_1 = session.get('http://example.com/get')
        coro_2 = session.get('http://example.org/tp')
        return combine_responses(*(await asyncio.gather(coro_1, coro_2)))

接下来,我想要一些复杂的依赖结构.我希望在我拥有所有先决条件时开始操作,并在需要结果时获得结果.这里有助于asyncio.ensure_future分别从事件循环管理的协程中执行单独的任务:

async def do_the_job():
    with aiohttp.ClientSession() as session:
        fut_1 = asyncio.ensure_future(session.get('http://httpbin.org/ip'))
        coro_2 = session.get('http://httpbin.org/get')
        coro_3 = session.post('http://httpbin.org/post', data=(await coro_2)
        coro_3_result = await coro_3
        return combine_responses(await fut_1, coro_3_result)

是不是在我的逻辑流程中使用协同程序实现并行非阻塞IO,我必须使用asyncio.ensure_future或者asyncio.gather(实际使用asyncio.ensure_future)？是否有一种不那么"冗长"的方式？

通常开发人员必须考虑协同应该成为单独的任务并使用上述功能来获得最佳性能,这是真的吗？

在事件循环中使用没有多个任务的协同程序是否有意义？

现实生活中的事件循环任务有多"重"？当然,它们比OS线程或进程"更轻".我应该在多大程度上争取尽可能少的此类任务？

我正在使用aiohttp下载图像，并且想知道是否存在一种方法来限制尚未完成的打开请求的数量。这是我目前拥有的代码：

async def get_images(url, session):

    chunk_size = 100

    # Print statement to show when a request is being made. 
    print(f'Making request to {url}')

    async with session.get(url=url) as r:
        with open('path/name.png', 'wb') as file:
            while True:
                chunk = await r.content.read(chunk_size)
                if not chunk:
                    break
                file.write(chunk)

# List of urls to get images from
urls = [...]

conn = aiohttp.TCPConnector(limit=3)
loop = asyncio.get_event_loop()
session = aiohttp.ClientSession(connector=conn, loop=loop)
loop.run_until_complete(asyncio.gather(*(get_images(url, session=session) for url in urls)))

问题是，我输入了打印语句，以显示每个请求的发出时间，并且它一次发出将近21个请求，而不是我希望限制的3个请求（即，一旦图像下载完成） ，它可以移至列表中的下一个网址进行获取）。我只是想知道我在做什么错。

这是一个奇怪的错误，因为当我尝试/捕获它时，它什么也不打印。

我正在使用 sanic 服务器 asyncio.gather 同时收集一堆图像，超过 3000 张图像。

在处理较小的样本量时，我没有遇到此错误。

简化示例：

from sanic import Sanic
from sanic import response
from aiohttp import ClientSession

from asyncio import gather

app = Sanic()

@app.listener('before_server_start')
async def init(app, loop):
    app.session = ClientSession(loop=loop)

@app.route('/test')
async def test(request):
    data_tasks = []
    #The error only happened when a large amount of images were used
    for imageURL in request.json['images']:
        data_tasks.append(getRaw(imageURL))
    await gather(*data_tasks)
    return response.text('done')

async def getRaw(url):
    async with app.session.get(url) as resp:
        return await resp.read()

这个错误可能是什么？如果这是我的主机/互联网的某种限制，我该如何避免？

如果有帮助，我正在使用来自 DigitalOcean 的具有 …