标签: python-asyncio

我正在努力熟悉asyncio,所以我决定编写一个数据库客户端.但是,性能与同步代码完全匹配.我确信这是我对一个概念的误解.有人可以解释我在做什么吗？

请参阅以下代码示例:

class Connection:
    def __init__(self, reader, writer, loop):
        self.futures = deque()

        # ...

        self.reader_task = asyncio.async(self.recv_data(), loop=self.loop)

    @asyncio.coroutine
    def recv_data(self):
        while 1:
            try:
                response = yield from self.reader.readexactly(4)
                size, = struct.unpack('I', response)
                response = yield from self.reader.readexactly(size)

                # ...                

                future = self.futures.popleft()

                if not future.cancelled():
                    future.set_result(response)

            except Exception:
                break

    def send_data(self, data):
        future = asyncio.Future(loop=self.loop)
        self.futures.append(future)

        self.writer.write(data)

        return future


loop = asyncio.get_event_loop()


@asyncio.coroutine
def benchmark():
    connection = yield from create_connection(loop=loop, ...)

    for i in range(10000):
        yield from connection.send_data(...)


s …

我正在尝试从python asyncio任务和协同程序文档中运行此示例

import asyncio

@asyncio.coroutine
def slow_operation(future):
    yield from asyncio.sleep(1)
    future.set_result('Future is done!')

def got_result(future):
    print(future.result())
    loop.stop()

loop = asyncio.get_event_loop()
future = asyncio.Future()
asyncio.ensure_future(slow_operation(future))
future.add_done_callback(got_result)
try:
    loop.run_forever()
finally:
    loop.close()

但是,我收到此错误:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
AttributeError: 'module' object has no attribute 'ensure_future'

这条线似乎让我感到悲痛:

asyncio.ensure_future(slow_operation(future))

我的python解释器在OSX Yosemite上是3.4.3,我上面链接的文档版本,我从中复制了这个例子,所以我不应该得到这个错误.这是我的python解释器的终端抓取:

Python 3.4.3 (default, Feb 25 2015, 21:28:45) 
[GCC 4.2.1 Compatible Apple LLVM 6.0 (clang-600.0.56)] on darwin
Type "help", "copyright", "credits" or "license" for more …

我习惯了asyncio并且发现任务处理相当不错,但是将异步库与传统的io库混合起来可能很困难.我目前面临的问题是如何正确解码异步StreamReader.

最简单的解决方案是read()块字节字符串,然后解码每个块 - 请参阅下面的代码.(在我的程序中,我不打印每个块,但将其解码为字符串并将其发送到另一个方法进行处理):

import asyncio
import aiohttp

async def get_data(port):
    url = 'http://localhost:{}/'.format(port)
    r = await aiohttp.get(url)
    stream = r.content
    while not stream.at_eof():
        data = await stream.read(4)
        print(data.decode('utf-8'))

这样可以正常工作,直到有一个utf-8字符在太多块之间分开.例如,如果响应是b'M\xc3\xa4dchen mit Bi\xc3\x9f\n',则读取3的块将起作用,但是4的块将不起作用(因为\xc3并且\x9f处于不同的块中并且解码以结尾的块\xc3将引发以下错误:

UnicodeDecodeError: 'utf-8' codec can't decode byte 0xc3 in position 3: unexpected end of data

我查看了这个问题的正确解决方案,至少在阻塞世界中,似乎是io.TextIOWrapper或codecs.StreamReaderWriter(其差异在PEP 0400中讨论).但是,这两者都依赖于典型的阻塞流.

我花了30分钟搜索asyncio的例子并继续寻找我的decode()解决方案.有没有人知道更好的解决方案,或者这是python的asyncio中缺少的功能？

作为参考,以下是使用具有异步流的两个"标准"解码器的结果.

使用编解码器流阅读器:

r = yield from aiohttp.get(url)
decoder = codecs.getreader('utf-8')
stream = decoder(r.content)

例外:

File "echo_client.py", line 13, in …

想象一下以下非常常见的情况:您编写了一个冗长而复杂的函数,并意识到应该将一些代码提取到一个单独的函数中以便重用和/或可读.通常,这个额外的函数调用不会改变程序的语义.

但是,现在假设您的函数是一个协程,并且您要提取的代码包含至少一个异步调用.将它提取到一个单独的函数现在突然改变你的程序语义,方法是插入一个新的协同程序产生的点,事件循环控制,并且可以在其间安排任何其他协程.

以前的例子:

async def complicated_func():
    foo()
    bar()
    await baz()

示例之后:

async def complicated_func():
    foo()
    await extracted_func()

async def extracted_func():
    bar()
    await baz()

在之前的示例中,complicated_func保证不会在呼叫foo()和呼叫之间暂停bar().重构后,此保证将丢失.

我的问题是:是否有可能调用extracted_func()它立即执行,好像它的代码是内联的？或者是否有其他方法可以在不更改程序语义的情况下执行此类常见的重构任务？

我有一个简单的测试,我使用该run_forever方法运行Python asyncio事件循环,然后立即在另一个线程中停止它.但是,事件循环似乎没有终止.我有以下测试用例:

import asyncio
from threading import Thread

loop = asyncio.get_event_loop()
thread = Thread(target=loop.run_forever)
thread.start()
print('Started!')
loop.stop()
print('Requested stop!')
thread.join()
print('Finished!')

此测试用例打印:

Started!
Requested stop!

因此,测试似乎阻塞thread.join(),等待事件循环终止.

如果我转储我的线程,我得到以下内容:

Thread 0x00007000087ec000 (most recent call first):
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/selectors.py", line 577 in select
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 1388 in _run_once
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 421 in run_forever
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/threading.py", line 862 in run
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/threading.py", line 914 in _bootstrap_inner
  File "/Library/Frameworks/Python.framework/Versions/3.5/lib/python3.5/threading.py", line 882 in _bootstrap

Current thread 0x00007fffc6b273c0 (most recent call first): …

描述：（简化）

• 我有 2 个任务。
• 在每个任务中我有 3 个协程。
• 第一个任务的 2 个协程失败。（模拟）
• 在处理任务结果时，我收到一条“从未检索到任务异常”消息。
• 我相信这是因为该任务中仅处理了两个失败的协程之一的异常。
• 如何处理任务中两个协程的异常和/或避免“从未检索到任务异常”消息？

代码：（简体）

import asyncio
async def download(data):
    filename = "*" if data in ["b", "c"] else data  # simulated failure
    with open(filename, "w") as f:
        f.write(data)
async def coro(data_list):
    coroutines = [download(data) for data in data_list]
    for coroutine in asyncio.as_completed(coroutines):
        await coroutine
async def main():
    task1 = asyncio.create_task(coro(["a", "b", "c"]))
    task2 = asyncio.create_task(coro(["d", "e", "f"]))
    results = await asyncio.gather(task1, task2, return_exceptions=True)
    for _ in results:
        pass
asyncio.run(main())

输出：（简化） …

我正在 FastAPI 端点中工作，该端点进行 I/O 绑定操作，该操作是异步的以提高效率。但是，这需要时间，所以我想缓存结果以在一段时间内重复使用。

我目前有这个：

from fastapi import FastAPI
import asyncio

app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

@app.get('/')
async def get(key):
    return await _get_expensive_resource(key)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("test:app")

我正在尝试使用该cachetools包来缓存结果，并且我尝试了如下操作：

import asyncio
from cachetools import TTLCache
from fastapi import FastAPI
  
app = FastAPI()

async def _get_expensive_resource(key) -> None:
    await asyncio.sleep(2)
    return True

class ResourceCache(TTLCache):
    def __missing__(self, key):
        loop = asyncio.get_event_loop()
        resource = loop.run_until_complete(_get_expensive_resource(key))
        self[key] = resource
        return resource

resource_cache …

Flask 2.0 中介绍了 async/await 的用法。（https://flask.palletsprojects.com/en/2.0.x/async-await/）

我正在使用 Flask-RestX，那么是否可以在 RestX 请求处理程序中使用async/ ？await

就像是：

@api.route('/try-async')
class MyResource(Resource):
    @api.expect(some_schema)
    async def get(self):
        result = await async_function()
        return result

不起作用，当我尝试到达此端点时，出现错误：

TypeError: Object of type coroutine is not JSON serializable

有这方面的信息吗？

封装版本：

flask==2.0.1
flask-restx==0.4.0

我也flask[async]按照文档的建议安装了。

我正在尝试以下操作：

class Payload_Session_Generator:
    def __init__(self):
        pass

    async def __anext__(self):
        async for payload in generate_fb_payload():
            if type(payload) != str:
                yield payload
            else:
                StopAsyncIteration

    def __aiter__(self):
        return self

然后将其作为实例传递给不同的函数，并__aiter__显式调用该方法，并且该方法_iter是上述类的对象：

chunk = await self._iter.__anext__()

这会产生以下错误：

类型错误：对象 async_generator 不能在“await”表达式中使用

想象一下一个异步aiohttpWeb 应用程序，它由通过连接的 Postgresql 数据库支持asyncpg，并且不执行其他 I/O。我怎样才能有一个托管应用程序逻辑的中间层，而不是异步的？（我知道我可以简单地使所有内容异步 - 但想象我的应用程序具有大量应用程序逻辑，仅受数据库 I/O 约束，并且我无法触及其中的所有内容）。

伪代码：

async def handler(request):\n    # call into layers over layers of application code, that simply emits SQL\n    ...\n\ndef application_logic():\n    ...\n    # This doesn\'t work, obviously, as await is a syntax\n    # error inside synchronous code.\n    data = await asyncpg_conn.execute("SQL")\n    ...\n    # What I want is this:\n    data = asyncpg_facade.execute("SQL")\n    ...\n

如何asyncpg构建同步 fa\xc3\xa7ade 来允许应用程序逻辑进行数据库调用？async.run()在这种情况下，诸如 using或等浮动的配方asyncio.run_coroutine_threadsafe()不起作用，因为我们来自已经异步的上下文。我认为这不可能是不可能的，因为已经有一个事件循环原则上可以运行协 …