标签: python-asyncio

我正在尝试使用Asyncio和aiohttp库发出一堆请求(~1000),但我遇到了一个我找不到太多信息的问题.

当我用10个网址运行这个代码时,它运行得很好.当我用100多个网址运行它时,它会中断并给我RuntimeError: Event loop is closed错误.

import asyncio
import aiohttp


@asyncio.coroutine
def get_status(url):
    code = '000'
    try:
        res = yield from asyncio.wait_for(aiohttp.request('GET', url), 4)
        code = res.status
        res.close()
    except Exception as e:
        print(e)
    print(code)


if __name__ == "__main__":
    urls = ['https://google.com/'] * 100
    coros = [asyncio.Task(get_status(url)) for url in urls]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(coros))
    loop.close()

堆栈跟踪可以在这里找到.

任何帮助或洞察都会非常感激,因为我现在已经敲了几个小时.显然,这表明事件循环已经关闭,应该仍然是开放的,但我不知道这是怎么可能的.

我正在努力写出一些尽可能惯用的东西来收集存储在字典中的期货的结果.

我们假设我有以下代码:

import asyncio

async def sleep(seconds):
    print(f'sleeping for {seconds} seconds')
    await asyncio.sleep(seconds)
    print(f'finished sleeping {seconds} seconds')


async def run():
    tasks = {
        '4': sleep(4),
        '3': sleep(3),
        '2': sleep(2),
        '1': sleep(1),
    }
    print(await gather_from_dict(tasks))


if __name__ == '__main__':
     asyncio.get_event_loop().run_until_complete(run())

我期待的输出是:

sleeping for 2 seconds
sleeping for 1 seconds
sleeping for 4 seconds
sleeping for 3 seconds
finished sleeping 1 seconds
finished sleeping 2 seconds
finished sleeping 3 seconds
finished sleeping 4 seconds
{'4': None, '3': None, '2': None, '1': None} …

我无法理解asyncio.create_task()Python 3.7 中引入的函数应该如何工作。如果我做：

import asyncio

async def helloworld():
    print("Hello world from a coroutine!")
    asyncio.create_task(helloworld())

def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(helloworld())

if __name__ == "__main__":
    main()

我得到：

Hello world from a coroutine! Hello world from a coroutine!

作为输出（即协程运行两次）。这怎么不是无限递归呢？当我使用await关键字时，我希望看到我看到的内容：

import asyncio


async def helloworld():
    print("Hello world from a coroutine!")
    await helloworld()


def main():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(helloworld())


if __name__ == "__main__":
    main()

有了这个，我得到：

Hello world from a coroutine! Hello world from a coroutine! Hello world from a coroutine! …

我正在使用 api，有时它会给出一些奇怪的状态代码，只需重试相同的请求就可以修复这些代码。我正在使用 aiohttp 异步地向这个 api 发出请求

我也在使用退避库来重试请求，但是似乎仍然没有重试 401 请求。

   @backoff.on_exception(backoff.expo, aiohttp.ClientError, max_tries=11, max_time=60)
    async def get_user_timeline(self, session, user_id, count, max_id, trim_user, include_rts, tweet_mode):

        params = {
            'user_id': user_id,
            'trim_user': trim_user,
            'include_rts': include_rts,
            'tweet_mode': tweet_mode,
            'count': count
        }


        if (max_id and max_id != -1):
            params.update({'max_id': max_id})

        headers = {
            'Authorization': 'Bearer {}'.format(self.access_token)    
        }

        users_lookup_url = "/1.1/statuses/user_timeline.json"

        url = self.base_url + users_lookup_url

        async with session.get(url, params=params, headers=headers) as response:
            result = await response.json()
            response = {
                'result': result,
                'status': response.status,
                'headers': response.headers …

我最近偶然发现了混淆gevent和基于asyncio的代码的问题，因为当我用猴子修补它们时，某些同步库可以很好地工作gevent.monkey.patch_all()。我找到了该aiogevent库，该库似乎对实现PEP 3156有所帮助，并将asyncio事件循环替换为您选择的其他实现（在本例中为gevent）。我发现的git仓库的最后一次重大提交是在4年前。修复setup.py之后，我设法成功安装了它，但是问题是它没有通过所有测试。

这些测试之一是test_soon，它产生应该执行操作并停止循环的greenlet。该测试将永远挂起，因为loop.stop()对循环没有任何影响，该循环预计将在所有任务完成后停止。我写了两个代码片段来检查它是否在传统的协程中发生，而另一个则通过来检查gevent.spawn。

import gevent
import aiogevent
import asyncio

asyncio.set_event_loop_policy(aiogevent.EventLoopPolicy())

loop = asyncio.get_event_loop()

async def func():
    print('bloop')
    loop.stop()

loop.create_task(func())
loop.run_forever() # works alright and stops as soon as func finish

和gevent.spawn：

import gevent
import aiogevent
import asyncio

asyncio.set_event_loop_policy(aiogevent.EventLoopPolicy())

loop = asyncio.get_event_loop()

def func():
    print('bloop')
    loop.stop()

g = gevent.spawn(func)
loop.run_forever() # func is executed as soon as loop runs, but loop.stop() is ignored

问题是：这里可能出什么问题？我清楚地看到，在启动循环之后，greenlet会运行，但是循环是否“未跟踪”？我在异步源中找不到与该机制相对应的确切行，对于gevent来说也是如此-我对这些模块的内部并不十分熟悉，并且通过它们进行搜索令人困惑，但是我想知道有什么区别和aiogevent为了通过测试，必须对事件循环进行哪些更改。

upd1：为了强调该问题，gevent.hub.Hub …

我正在使用 aiohttp 发出异步请求，我想测试我的代码。我想模拟 aiohttp.ClientSession 发送的请求。我正在寻找类似于响应处理requestslib模拟的方式。

我如何模拟aiohttp.ClientSession?

# sample method
async def get_resource(self, session):
    async with aiohttp.ClientSession() as session:
        response = await self.session.get("some-external-api.com/resource")
        if response.status == 200:
            result = await response.json()
            return result

        return {...}

# I want to do something like ...
aiohttp_responses.add(
    method='GET', 
    url="some-external-api.com/resource", 
    status=200, 
    json={"message": "this worked"}
)

async def test_get_resource(self):
    result = await get_resource()
    assert result == {"message": "this worked"}

• 我已经通读了aiohttp 测试文档。似乎他们涵盖了模拟对您的 Web 服务器的传入请求，但我不确定这是否有助于我模拟对传出请求的响应

编辑

我在几个项目中使用了https://github.com/pnuckowski/aioresponses，它非常适合我的需求。

阅读asyncio 文档，我意识到我不理解一个非常基本和基本的方面：直接等待协程与等待包含在任务中的同一个协程之间的区别。

在文档示例中，对say_after协程的两次调用在没有等待时顺序运行create_task，而在包装中时并发运行create_task。所以我明白这基本上就是区别，这是一个非常重要的区别。

然而让我困惑的是，在我到处阅读的示例代码中（例如展示如何使用aiohttp），有很多地方等待（用户定义的）协程（通常在其他一些用户定义的协程中间）而没有被包裹在一个任务中，我想知道为什么会这样。确定何时应将协程包装在任务中的标准是什么？

我正在使用 FastAPI 为分析系统构建异步后端。问题是它必须： a) 侦听 API 调用并始终可用；b) 定期执行数据收集任务（解析数据并将其保存到数据库中）。

我编写了这个函数来充当守护进程：

    async def start_metering_daemon(self) -> None:
        """sets a never ending task for metering"""
        while True:
            delay: int = self._get_delay()  # delay in seconds until next execution
            await asyncio.sleep(delay)
            await self.gather_meterings()  # perfom data gathering

我想要实现的是，当应用程序启动时，它还会将此守护程序函数添加到主事件循环中，并在有时间时执行它。然而，我一直无法找到适合任务规模的合适解决方案（添加芹菜和类似的东西是一种矫枉过正）。

我尝试过以下方法来实现这一目标，但没有一个有效：

@app.on_event("startup")
async def startup_event() -> None:
    """tasks to do at server startup"""
    await Gatherer().start_metering_daemon()

结果：由于线程被阻塞，服务器无法启动

@app.on_event("startup")
async def startup_event() -> None:
    """tasks to do at server startup"""
    fastapi.BackgroundTasks().add_task(Gatherer().start_metering_daemon)

结果：如日志中观察到的，任务从未执行

@app.on_event("startup")
async def startup_event() -> None:
    """tasks …

关于 Python 协程（我主要是指async/await）是无堆栈的还是有堆栈的，我看到了相互矛盾的观点。

一些消息来源称它们是堆栈的：

• http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2020/p2074r0.pdf

  “Python 协程是堆栈式的。”

• Python 中的协程与 Lua 中的协程相比如何？

  是的，Python 协程是堆栈式的、一流的且不对称的。

虽然其他人似乎暗示它们是无堆栈的，例如https://gamelisp.rs/reference/coroutines.html

GameLisp 的协程遵循 Rust、Python、C# 和 C++ 设置的模型。我们的协程是“无堆栈的”

总的来说，我的理解始终是，任何有意义的 async/await 实现都意味着无堆栈协程，而堆栈协程基本上是纤程（用户空间线程，通常或多或少地协作切换），例如 goroutine、Boost.Coroutine，显然是 Lua 中的协程等。

我的理解正确吗？或者 Python 协程在某种程度上与 C++ 中的协程有根本的不同，并且是堆栈式的？或者上述来源的作者是否有不同的意思？

我们正在使用 fastapi、uvicorn、sqlalchemy 和 PostgreSQL 构建 ASGI 应用程序。问题是：与具有多个工作线程的 WSGI 应用程序相比，我们应该如何设置pool_size才能create_async_engine不使其成为瓶颈？

据我了解，在 WSGI 应用程序中，如果我们使用每个（和）线程运行N进程，我们将获得最多连接。但是 ASGI 怎么样（如果我们有一个进程）——可以打开多少个连接？另外（因为我们每个进程只创建一个）？那么我们应该将其设置为？Mpool_size=MN * Mpool_sizeAsyncEnginepool_size=N * M

而且，如果我们简单地增加这个数字，那么我们就能够await向数据库发出更多的并发请求，对吧？

其背后的直觉是什么？

提前致谢！