标签: python-asyncio

你好，我对Python相当陌生，我正在尝试将Flask上的现有应用程序转换为Quart（https://gitlab.com/pgjones/quart），它应该构建在asyncio之上，所以我可以使用Goblin OGM 与 JanusGraph 或 TinkerPop 交互。根据我在 Goblin 上找到的示例，我需要获取一个事件循环来异步运行命令。

    >>> import asyncio
    >>> from goblin import Goblin

    >>> loop = asyncio.get_event_loop()
    >>> app = loop.run_until_complete(
    ...     Goblin.open(loop))
    >>> app.register(Person, Knows)

然而，我找不到从 Quart 获取事件循环的方法，即使它是构建在 asyncio 之上的。

有谁知道我怎样才能得到它？任何帮助将不胜感激。

我编写的代码似乎可以实现我想要的功能，但我不确定这是否是一个好主意，因为它混合了线程和事件循环以在主线程外运行无限循环。这是一个最小的代码片段，它体现了我正在做的事情的想法：

import asyncio
import threading

msg = ""

async def infinite_loop():
    global msg
    while True:
        msg += "x"
        await asyncio.sleep(0.3)

def worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    asyncio.get_event_loop().run_until_complete(infinite_loop())

t = threading.Thread(target=worker, daemon=True)
t.start()

主要思想是我有一个无限循环，每 0.3 秒操作一个全局变量。我希望这个无限循环在主线程之外运行，这样我仍然可以访问主线程中的共享变量。这在 jupyter 中特别有用，因为如果我run_until_complete在主线程中调用，我就无法再与 jupyter 交互。我希望主线程可用于交互式访问和修改msg. 在我的示例中，使用异步似乎没有必要，但我正在使用一个具有异步代码的库来运行服务器，因此这是必要的。我对 python 中的异步和线程很陌生，但我记得在某处读过/听到过，使用异步线程会带来麻烦......这是一个坏主意吗？我的方法是否存在潜在的并发问题？

我一直在使用 asyncio 进行并发，但是遇到了问题。我需要使用 asyncio 安排一个任务，但它是阻塞的，所以我想使用并发.futures 中的线程池在执行器中执行它。

我见过这样的例子，安排任务：

now = loop.time()
loop.call_at(now + 60, callback, arg, loop)

像这样在执行器中运行任务：

blocking_tasks = [
    loop.run_in_executor(executor, blocks)
    for i in range(6)
]
completed, pending = await asyncio.wait(blocking_tasks)

但是如何安排阻塞任务在执行器中运行呢？

我想使用一组类别执行网页抓取，每个类别还有一个 URL 列表。所以我决定在主函数中只根据每个类别来调用一个函数，并且在内部函数内有一个非阻塞调用。

所以这是代码：

def main():
    loop = asyncio.get_event_loop()
    b = loop.create_task(f("p", all_p_list))
    f = loop.create_task(f("f", all_f_list))

    loop.run_until_complete(asyncio.gather(p, f))

它应该同时执行该f函数。

但该f函数还必须运行循环，因为在该函数中它根据每个 URL 同时调用一个函数。

async def f(category, total): 
    urls = [urls_template[category].format(t) for t in t_list]
    soups_coro = map(parseURL_async, urls)

    loop = asyncio.get_event_loop()
    result = await loop.run_until_complete(asyncio.gather(*soups_coro))

但是运行脚本后，出现错误This event loop is already running，我发现这是因为我loop.run_until_complete()同时调用了内部函数和外部函数。

但是，当我剥离run_until_complete(), 并仅调用f()时main()，函数调用立即完成，并且无法等待内部函数完成。所以不可避免的要调用循环中的main(). 但后来我认为它与内部函数不兼容，内部函数也必须调用它。

我该如何处理这个问题并运行循环？原始代码全部相同main()并且有效，但如果可能的话我想使其更清晰。

我在一个项目中使用 django ORM（在 django 之外）。我的工作流程是

1. 通过 django ORM 选择对象并
2. 然后使用 asyncio lib 将其发送到消息队列

问题是您无法在异步环境中调用阻塞函数，并且无法在阻塞环境中使用 async/await 。

我想出了两个解决方案：

1. 整个程序应该是异步的。并loop.run_in_executor在需要时用于调用阻塞函数。

2. 整个程序应该是同步的。并使用asyncio.run()（Python 3.7）调用所需的异步函数。

我无法决定哪一种方法更好。

我知道以前也有人问过类似的问题。我的问题是尝试组合阻塞和非阻塞代码时是否有一般规则？

我试图将 pyppeteer 和 quart 结合起来，但由于启动浏览器需要花费很多时间，我宁愿全局处理它（使用异步锁），这似乎意味着我需要手动处理清理。这是我的最小代码示例：

#!/usr/bin/env python3\n\nimport asyncio\nimport atexit\nimport pyppeteer\n\nfrom quart import Quart, Response, request\n\napp = Quart(__name__)\nbrowser = asyncio.get_event_loop().run_until_complete(pyppeteer.launch())\n\nasync def cleanup_async():\n    await browser.quit()\n\n@atexit.register\ndef cleanup():\n    asyncio.get_event_loop().run_until_complete(cleanup_async())\n

问题如下：

[22:26:53] \xe2\x9e\x9c  strokes git:(async_browser) \xe2\x9c\x97  % QUART_APP=/tmp/quart_cleanup.py timeout 10s quart run -h 0.0.0.0\nRunning on http://0.0.0.0:5000 (CTRL + C to quit)\nTraceback (most recent call last):\n  File "/home/d33tah/virtualenv-py3/bin/quart", line 11, in <module>\n    sys.exit(main())\n  File "/home/d33tah/virtualenv-py3/lib/python3.6/site-packages/quart/cli.py", line 208, in main\n    cli.main(args=args, prog_name=name)\n  File "/home/d33tah/virtualenv-py3/lib/python3.6/site-packages/quart/cli.py", line 152, in main\n    return super().main(*args, **kwargs)\n  File "/home/d33tah/virtualenv-py3/lib/python3.6/site-packages/click/core.py", line …

这个问题不同于Is there a way to use asyncio.Queue in multiple thread?

我有 2 个异步事件循环在两个不同的线程中运行。Thread1 通过asyncio.Queue()向 Thread2 生成数据。

其中一个线程抛出异常：got Future <Future pending> attached to a different loop

现在这是真的，因为我有一个在不同循环中使用的队列。如何在两个不同线程中的两个循环之间共享队列？

示例代码：

q = asyncio.Queue()

async def producer(q):
    await asyncio.sleep(3)
    q.put(1)

def prod_work(q):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(producer(q))

async def consumer(q):
    await asyncio.sleep(3)
    res = await q.get()

def cons_work(q):
    loop2 = asyncio.new_event_loop()
    asyncio.set_event_loop(loop2)
    loop2.run_until_complete(consumer(q))

def worker(q):
    # main thread - uses this threads loop  
    prod = threading.Thread(target=prod_work, args=(q,))

    # separate …

我有一个用于在 Python (3.6) 中测试异步代码的方法：

@asyncio.coroutine
def coroutine_creater(value):
    return value

我这样使用它：

async def test_test():
    my_mock = Mock(return_value=coroutine_creater(5))

    # I call with "await"
    first_call = await my_mock()
    second_call = await my_mock()

    assert first_call == 5, 'first call failed'
    assert second_call == 5, 'second call failed'  # this call fails

这样我就可以为异步调用创建模拟。我发现如果我调用异步方法两次，这不起作用。在我的代码中，first_call如我所料，等于 5，但second_call等于 None。这里发生了什么？如何测试多次调用 Mock 异步方法的代码？

我实际上正在使用WebSockets 7，一切正常，多个用户可以连接到 websocket，发送请求并接收来自服务器的响应，但有一个问题。

如果用户发送到图像、pdf、歌曲、视频等。服务器必须解析该问题，保存到数据库、转换等并发送响应，用户可以向 websocket 发送更多数据，但该数据将排队直到收到上一条消息当用户迫切希望在尽可能短的时间内发送数据时，这并不好。

连接的用户有可能发送并行请求吗？

这是我正在使用的示例

import asyncio
import websockets

# Process the data
async def read(msg, websocket):
    # In this case, the message will get back to the user
    await websocket.send(msg)

async def counter(websocket, path):
    try:
        async for message in websocket:
            await read(message, websocket)
    except websockets.exceptions.ConnectionClosed:
        pass

try:
    asyncio.get_event_loop().run_until_complete(
        websockets.serve(counter, '0.0.0.0', 4444, max_size=10**8))
    asyncio.get_event_loop().run_forever()
except KeyboardInterrupt:
    quit()

更新

我正在使用循环，在尝试并行发出请求时，它不起作用。

import asyncio
import websockets

# Process the data
async def read(msg, websocket): …

"Cannot write to closing transport"我收到了从 引发的异常aiohttp.http_writer.StreamWriter#_write，但仅在一小部分情况下。

相关片段。

    session: aiohttp.ClientSession
    async with session.get(url, timeout=60) as response:
        txt = await response.text()
        response.close()
        return txt

到底是怎么回事？我不认为服务器大小正在关闭套接字。