标签: python-asyncio

在之前的问题中,其中一位作者aiohttp善意地建议使用以下新语法从aiohttp获取多个URL:async withPython 3.5

import aiohttp
import asyncio

async def fetch(session, url):
    with aiohttp.Timeout(10):
        async with session.get(url) as response:
            return await response.text()

async def fetch_all(session, urls, loop):
    results = await asyncio.wait([loop.create_task(fetch(session, url))
                                  for url in urls])
    return results

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    # breaks because of the first url
    urls = ['http://SDFKHSKHGKLHSKLJHGSDFKSJH.com',
            'http://google.com',
            'http://twitter.com']
    with aiohttp.ClientSession(loop=loop) as session:
        the_results = loop.run_until_complete(
            fetch_all(session, urls, loop))
        # do something with the the_results

但是,当其中一个session.get(url) …

假设我们有一堆下载链接,每个链接可能需要不同的下载时间.我只允许使用最多3个连接下载.现在,我想确保使用asyncio有效地执行此操作.

这就是我想要实现的目标:在任何时候,尽量确保我至少运行3次下载.

Connection 1: 1---------7---9---
Connection 2: 2---4----6-----
Connection 3: 3-----5---8-----

数字代表下载链接,而连字符代表等待下载.

这是我正在使用的代码

from random import randint
import asyncio

count = 0


async def download(code, permit_download, no_concurrent, downloading_event):
    global count
    downloading_event.set()
    wait_time = randint(1, 3)
    print('downloading {} will take {} second(s)'.format(code, wait_time))
    await asyncio.sleep(wait_time)  # I/O, context will switch to main function
    print('downloaded {}'.format(code))
    count -= 1
    if count < no_concurrent and not permit_download.is_set():
        permit_download.set()


async def main(loop):
    global count
    permit_download = asyncio.Event()
    permit_download.set()
    downloading_event = asyncio.Event()
    no_concurrent = 3 …

在处理 FastAPI 请求时，我有一个受 CPU 限制的任务要对列表的每个元素执行。我想在多个 CPU 内核上进行此处理。

在 FastAPI 中执行此操作的正确方法是什么？我可以使用标准multiprocessing模块吗？到目前为止，我发现的所有教程/问题仅涵盖 I/O 绑定任务，如 Web 请求。

import asyncio

from sqlalchemy import Column
from sqlalchemy import DateTime
from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import relationship
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import sessionmaker

engine = create_async_engine(
        "postgresql+asyncpg://user:pass@localhost/db",
        echo=True,
    )


# expire_on_commit=False will prevent attributes from being expired
# after commit.
async_session = sessionmaker(
    engine, expire_on_commit=False, class_=AsyncSession
)


Base = …

是因为协程可能会在未来被抢占吗？或者它允许人们使用关键部分的收益率(不应该鼓励IMO)？

我想启用Asyncio的非屈服协程检测,但没有成功.

这个简单的代码实现了以下建议:https:
//docs.python.org/3/library/asyncio-dev.html#asyncio-logger

但实际上并没有捕捉到未屈服的"虚拟"协程.

import sys, os
import asyncio
import logging
import warnings

os.environ['PYTHONASYNCIODEBUG'] = '1'
logging.basicConfig(level=logging.DEBUG)
warnings.resetwarnings()

@asyncio.coroutine
def dummy():
    print('yeah, dummy ran!!')

@asyncio.coroutine
def startdummy():
    print('creating dummy')
    dummy()

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

我预计程序会以关于协同程序'dummy'的警告而结束,但是没有产生.

实际上,结果是:

DEBUG:asyncio:Using selector: SelectSelector
creating dummy
sys:1: ResourceWarning: unclosed <socket object at 0x02DCB6F0>
c:\python34\lib\importlib\_bootstrap.py:2150: ImportWarning: sys.meta_path is empty
sys:1: ResourceWarning: unclosed <socket object at 0x02DE10C0>

没有一丝废弃的鲜花.我错过了什么？

我在我的应用程序中使用asyncio,我对将事件循环作为参数传递感到困惑.

使用事件循环编写函数/方法时有三种可能性:

• 将asyncio 事件循环作为参数传递
• 不要为事件循环使用参数并使用asyncio.get_event_loop()
• 将它作为参数传递给事件循环是可选的.如果没有通过,请使用asyncio.get_event_loop()

似乎最后一种情况大部分时间都在使用,但即使在asyncio api中,使用也是不一致的.因为我没有缩进使用两个分离的事件循环,所以只是asyncio.get_event_loop()在需要的地方使用？

什么是最好的方式？

我正在尝试制作一个装饰器来包装协程或函数。

我尝试的第一件事是包装器中的简单重复代码：

def duration(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_ts = time.time()
        result = func(*args, **kwargs)
        dur = time.time() - start_ts
        print('{} took {:.2} seconds'.format(func.__name__, dur))
        return result

    @functools.wraps(func)
    async def async_wrapper(*args, **kwargs):
        start_ts = time.time()
        result = await func(*args, **kwargs)
        dur = time.time() - start_ts
        print('{} took {:.2} seconds'.format(func.__name__, dur))
        return result

    if asyncio.iscoroutinefunction(func):
        return async_wrapper
    else:
        return wrapper

这有效，但我想避免代码重复，因为这并不比编写两个单独的装饰器好多少。

然后我尝试使用类制作一个装饰器：

class SyncAsyncDuration:
    def __init__(self):
        self.start_ts = None

    def __call__(self, func):
        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs):
            self.setup(func, args, kwargs)
            result = …

我正在尝试创建一个可以传递给 asyncio.gather() 的协程列表

但是，当我将它们附加到列表中时，我想将参数附加到这些协程。

下面显示的我当前的方法使用 functools.partial。不幸的是 asyncio.gather 不接受部分函数，​​这是有道理的。

对我来说没有意义的是如何找到解决方案。

示例代码：

async def test(arg1):
    print(arg1)

statements = []

function = functools.partial(test, "hello world")
statements.append(function)

results = await asyncio.gather(*statements)

那么如何将参数附加到函数，以便它仍然可以传递给 asyncio.gather？

*编辑

看来我是比较傻了。

我的解决方案相当简单，不要使用 functools.partial，只需将协程直接附加到列表中即可。

代码：

async def test(arg1):
    print(arg1)

async def main():
    statements = []
    statements.append(test("hello_world"))
    results = await asyncio.gather(*statements)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

我asyncio以一种非常基本的方式用于应用程序。检查互联网上的大多数教程（甚至官方文档），我看到他们使用get_event_loop()和loop.run_until_complete()：

import asyncio

async def say(what, when):
    await asyncio.sleep(when)
    print(what)

loop = asyncio.get_event_loop()
loop.run_until_complete(say('hello world', 1))
loop.close()

但是在Python 3.7 docs 中，我们可以阅读：

应用程序开发人员通常应该使用高级 asyncio 函数，例如asyncio.run()，并且很少需要引用循环对象或调用其方法。本节主要面向需要更好地控制事件循环行为的低级代码、库和框架的作者。

我发现它更简洁、更易于使用，但它仅适用于 Python 3.7+。所以在这里我必须做出选择，是使用 Python 3.7+run()还是使其与 Python 3.6 兼容并使用事件循环。你会如何处理这个？有没有一种简单的方法可以使它与 Python 3.6 向后兼容？在 Python 3.7 成为通用版本之前，我是否应该首先检查 Python 版本并基于此使用一种或另一种方式？