标签: python-asyncio

我试图在异步函数运行完成后调用回调

这是我想做的一个例子：

import asyncio

async def asyncfunction():
    print('Hello')
    await asyncio.sleep(10)
    print('World')
    return 10

def callback(n):
    print(f'The async function returned: {n}')

loop = asyncio.get_event_loop()

# Will block the print until everything is done
callback(loop.run_until_complete(asyncfunction()))
print('Hey')

这是它的作用：

Hello
World
The async function returned: 10
Hey

这就是我想要它做的
编辑：“嘿”的位置并不重要，只要它不必等待异步函数完成即可

Hello
Hey
World
The async function returned: 10

编辑：经过一些测试，我找到了一种可以实现我想要的功能的方法，尽管我不知道这是否是最好的方法

Hello
World
The async function returned: 10
Hey

我可能需要帮助更好地表达这个问题。我正在通过 python3.7 和一个类（称为Worker()）编写一个异步 api 接口。Worker有一些我想使用运行的阻塞方法loop.run_in_executor()。

我想构建一个装饰器，我可以async在所有非方法之上添加Worker，但我不断遇到问题。

有人告诉我我需要await wraps()在下面的装饰器中：

def run_method_in_executor(func, *, loop=None):
    async def wraps(*args):
        _loop = loop if loop is not None else asyncio.get_event_loop()
        return await _loop.run_in_executor(executor=None, func=func, *args)
    return wraps

这会抛出： RuntimeWarning: coroutine 'run_method_in_executor.<locals>.wraps' was never awaited

我不知道如何正确地执行，await wraps()因为包含函数和修饰函数不是异步的。不确定这是由于误解asyncio还是装饰器的误解。

任何帮助（或帮助澄清）将不胜感激！

假设我有以下代码：

from types import coroutine

@coroutine
def stop():
    yield 1

async def test2():
    await stop()

async def test1():
    await test2()
    await test2() # Here
    await test2()

coro = test1()
coro.send(None)
coro.send(None)

如何获得当前协程状态的回溯（回溯对象），即标有注释的行，而不人为地抛出不需要的异常？

问题

我有一个库，目前没有异步支持，需要从异步代码调用。handler异步代码通过处理程序（下面代码中的函数）调用库。当处理程序执行时，库会定期调用回调 ( callback_wrapper) 来报告进度。

同步处理程序的执行顺序是为了ThreadPoolExecutor使主事件循环能够在处理程序运行时处理更多事件。

发生的情况是同步回调立即执行，但异步回调仅在主处理程序执行后执行。期望的结果是立即执行异步回调。

我猜事件循环在run_in_executor调用时被阻止，但我不知道如何解决这个问题。

代码

import asyncio
import time
from concurrent.futures.thread import ThreadPoolExecutor

loop = asyncio.get_event_loop()


def handler():
    print('handler started')
    callback_wrapper()
    time.sleep(1)
    print('handler stopped')


async def callback():
    print('callback')


def callback_wrapper():
    print('callback wrapper started')
    asyncio.ensure_future(callback(), loop=loop)
    print('callback wrapper stopped')


async def main():
    handler()


with ThreadPoolExecutor() as pool:
    async def thread_handler():
        await loop.run_in_executor(pool, handler)


    loop.run_until_complete(main())

输出

handler started
callback wrapper started
callback wrapper stopped
handler stopped
callback

所需输出

handler started
callback …

我知道在python37中我们有一个新的api asyncio.get_running_loop()，它很容易使用，让我们在调用协程时不需要显式地传递eventloop。

我想知道我们是否可以使用任何方法在 python36 中获得相同的效果？

# which allows us coding conveniently with this api:
import asyncio

async def test():
    print("hello world !")

async def main():
    loop = asyncio.get_running_loop()
    loop.create_task(test())

asyncio.run(main())

我正在尝试使用 aiobotocore 和 boto 的 s3 方法从 s3 中提取多个文件get_object。我想让请求并发并最后将结果收集到一个数组中。我将如何构建我的代码？我在网上找不到任何如何执行此操作的示例。任何帮助将非常感激。

在 python asyncio websockets 库中，该示例调用run_forever(). 为什么需要这样做？

不应该run_until_complete()阻止并运行 websockets 循环吗？

#!/usr/bin/env python

# WS server example

import asyncio
import websockets

async def hello(websocket, path):
    name = await websocket.recv()
    print(f"< {name}")

    greeting = f"Hello {name}!"

    await websocket.send(greeting)
    print(f"> {greeting}")

start_server = websockets.serve(hello, "localhost", 8765)

asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()
# if you comment out the above line, this doesn't work, i.e., the server
# doesn't actually block waiting for data...

如果我注释掉 run_forever()，程序会立即结束。

start_server是图书馆返回的等待对象。为什么不足以run_until_complete导致它阻塞/等待hello()？

作为我上一个关于从同步函数调用异步函数的问题的后续，我发现了asyncio.run_coroutine_threadsafe。

从纸面上看，这看起来很理想。根据StackOverflow 问题中的评论，这看起来很理想。我可以创建一个新线程，获取对原始事件循环的引用，并安排异步函数在原始事件循环内运行，同时仅阻塞新线程。

class _AsyncBridge:
    def call_async_method(self, function, *args, **kwargs):
        print(f"call_async_method {threading.get_ident()}")
        event_loop = asyncio.get_event_loop()
        thread_pool = ThreadPoolExecutor()
        return thread_pool.submit(asyncio.run, self._async_wrapper(event_loop, function, *args, **kwargs)).result()

    async def _async_wrapper(self, event_loop, function, *args, **kwargs):
        print(f"async_wrapper {threading.get_ident()}")
        future = asyncio.run_coroutine_threadsafe(function(*args, **kwargs), event_loop)
        return future.result()

这不会出错，但也不会返回。期货只是挂起，异步调用永远不会被命中。call_async_method我是否在、_async_wrapper或两者中使用 Future 似乎并不重要；无论我在哪里使用 Future，它都会挂起。

我尝试将调用run_coroutine_threadsafe直接放入主事件循环中：

event_loop = asyncio.get_event_loop()
future = asyncio.run_coroutine_threadsafe(cls._do_work_async(arg1, arg2, arg3), event_loop)
return_value = future.result()

未来也悬在这里。

我尝试使用此处LoopExecutor定义的类，这似乎完全满足了我的需求。

event_loop = asyncio.get_event_loop()
loop_executor …