标签: python-asyncio

我正在寻找一种方法来生成不同的线程（在我的实际程序中，线程数在执行过程中可以改变）来执行无限运行的操作，这会在运行期间阻塞我的整个应用程序（最坏的情况是）几秒钟。
因此，我使用标准线程类asyncio（因为我的程序的其他部分正在使用它）。

这似乎工作得很好，根据这个线程，它似乎没问题，但是当搜索异步线程和asyncio时，我经常会偶然发现使用的建议ProcessPoolExecutor（例如在这个stackoverflow 帖子中）。 现在我想知道，以下方法是否真的是好的做法（甚至是危险的）？

class Scanner:
  def __init__(self):
    # Start a new Scanning Thread
    self.scan_thread = Thread(target=self.doScan, args=())
    self.scan_thread.start()

  def doScan(self):
    print("Started scanning")
    loop = asyncio.new_event_loop()
    loop.run_until_complete(self.connection())
    print("Stopped scanning")

list_of_scanner = []
list_of_scanner.append(Scanner())
list_of_scanner.append(Scanner())

背景：我自己开始质疑这一点，因为我的程序在生成线程时开始崩溃，主要是出现错误消息 RuntimeError: Task <Task pending ...> attached to a different loop。我知道这与我给您的示例没有直接联系，但我想我开始通过使用这些线程搞乱我的异步协程。

编辑

为了澄清起见，我想补充一下，为什么我使用asyncioand的这种奇怪的构造threads。

1. 我正在使用项目hbldh/bleak
   的这一部分作为线程运行的部分基本上是这样的：

   async def connection():
     x = await client.is_connected()
     async with BleakClient(address, …

   Run Code Online (Sandbox Code Playgroud)

我使用 dbus 进行 IPC。为了在程序的整个生命周期中只有一辆总线，我在这里使用单例。为了演示，我连接到 NetworkManager，但可以交换。此外，我正在asyncio整个项目中使用。这是该模块的一个简约工作示例，它将突出显示下面描述的问题：

import asyncio  # noqa

from dbus_next.aio import MessageBus
from dbus_next import BusType


BUS = None


async def get_bus():
    # Returns a BUS singleton
    global BUS
    if not BUS:
        BUS = await MessageBus(bus_type=BusType(2)).connect()

    return BUS


async def introspect():
    # Get the dbus singleton and call a method on that singleton
    bus = await get_bus()
    return await bus.introspect(
        'org.freedesktop.NetworkManager',
        '/org/freedesktop/NetworkManager',
    )

我正在使用pytest该pytest-asyncio插件进行测试，除了这种情况之外，它的工作原理就像魅力一样。这是一个简约的工作测试模块：

import pytest

from example import introspect


@pytest.mark.asyncio
async def …

我正在使用 pytest 编写 python 测试，并且有一些我想测试的异步代码，因此我安装了 pytest-asyncio 插件。异步代码使用 aiohttp 并运行测试，在测试成功运行后，我收到以下警告/错误/提示。

========================================================================================================================================================================= test session starts =========================================================================================================================================================================
platform win32 -- Python 3.8.7, pytest-6.2.1, py-1.10.0, pluggy-0.13.1
rootdir: C:\Users\user\workspace\example\app
plugins: asyncio-0.14.0
collected 1 item                                                                                                                                                                                                                                                                                                                                                       

tests\test_something.py .

========================================================================================================================================================================== 1 passed in 1.22s ==========================================================================================================================================================================
Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x000001D09DE4C310>
Traceback (most recent call last):
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 116, in __del__
    self.close()
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\proactor_events.py", line 108, in close
    self._loop.call_soon(self._call_connection_lost, None)
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\base_events.py", line 719, in call_soon
    self._check_closed()
  File "C:\Users\user\AppData\Local\Programs\Python\Python38\lib\asyncio\base_events.py", line 508, in _check_closed
    raise RuntimeError('Event loop …

我是 asyncio 的新手，我有一个测试函数，我正在尝试使用 pytest 进行测试：这是我的测试结构：

MyClass()
    
    async def myFunction(payload, headers):
        session = aiohttp.ClientSession()
        with session.post(url, json=payload, headers=headers) as resp:
        
            response = await resp.json()
    
        return response

然后我的测试如下：

def my_test():
    loop = asyncio.get_event_loop()
    reponse = loop.run_until_complete(MyClass().myFunction(url, headers))
    loop.close()

当我运行测试时我得到：

pytest运行时错误线程“主线程”中没有当前事件循环

我该如何解决它，我觉得我需要以不同的方式处理线程。

我看到了一些相关的问题，但没有人使用 pytests 最后我使用的是 Python 3.8.6。

我有个问题。因此，每次用户在我的不和谐服务器上写入聊天消息时，我都会运行一个任务 - 它称为on_message. 所以我的机器人在这个事件中有很多事情要做，我经常遇到这种错误：

Task was destroyed but it is pending!\ntask: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f68a7bdfc10>()]>>\n

所以我想如果我想解决这个问题，我需要加速我的代码。但遗憾的是，我不知道如何修复此错误。

编辑：我集成了计时，这就是我打印的内容：

Task was destroyed but it is pending!\ntask: <Task pending name='pycord: on_message' coro=<Client._run_event() done, defined at /Bots/gift-bot/discord/client.py:374> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f01063f98e0>()]>>\n2 if checks done - 7.867813110351562e-06\n5 if checks done - 0.0061550140380859375\nmysql checks done - 0.010785341262817383\ntask done - 0.13075661659240723\n2 if checks done - 8.344650268554688e-06\n5 if …

我正在尝试使用异步客户端库（该库也是由 Google 提供）调用外部 API（由 Google 提供的 API）。

我尝试调用的异步方法是async list_featurestores()（文档）。它提供了以下示例代码：

from google.cloud import aiplatform_v1

async def sample_list_featurestores():
    # Create a client
    client = aiplatform_v1.FeaturestoreServiceAsyncClient()

    # Initialize request argument(s)
    request = aiplatform_v1.ListFeaturestoresRequest(
        parent="parent_value",
    )

    # Make the request
    page_result = client.list_featurestores(request=request)

    # Handle the response
    async for response in page_result:
        print(response)

（我已经从上面的链接页面复制/粘贴了该代码）。

为了运行该代码，我对其进行了稍微调整：

import asyncio
from google.cloud import aiplatform_v1

async def sample_list_featurestores():
    client = aiplatform_v1.FeaturestoreServiceAsyncClient()
    request = aiplatform_v1.ListFeaturestoresRequest(parent="projects/MY_GCP_PROJECT/locations/europe-west2",)
    page_result  = client.list_featurestores(request=request)
    async for response in page_result:
        print(response)


if …

我最近发现并重现了由于使用asyncio.wait导致的内存泄漏。具体来说，我的程序定期执行某些函数，直到stop_event设置为止。我将程序简化为下面的代码片段（减少超时以更好地演示问题）：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    await asyncio.wait([stop_event.wait()], timeout=0.0001)

asyncio.run(main())

虽然这对我来说看起来无害，但事实证明这里存在内存泄漏。如果执行上面的代码，您将看到内存使用量在几分钟内增长到数百 MB。这让我很惊讶，并花了很长时间才找到。我期望在超时后，我正在等待的任何内容都会被清理（因为我自己没有保留任何对它的引用）。然而，事实证明并非如此。

使用gc.get_referrers，我能够推断每次调用时asyncio.wait(...)，都会创建一个新任务，该任务保存对返回的对象的引用stop_event.wait()，并且该任务将永远保留。具体来说，len(asyncio.all_tasks())随着时间的推移不断增加。即使超时了，任务仍然存在。只有在调用时，stop_event.set()这些任务才会立即完成，并且内存使用量才会急剧减少。

发现这一点后，文档中的这条注释让我尝试使用asyncio.wait_for代替：

与 wait_for() 不同，wait() 在发生超时时不会取消 future。

事实证明，它的表现确实如我所料。超时后不会保留任何引用，内存使用量和任务数量保持不变。这是没有内存泄漏的代码：

async def main():
  stop_event = asyncio.Event()

  while True:
    # Do stuff here
    try:
      await asyncio.wait_for(event.stop_event(), timeout=0.0001)
    except asyncio.TimeoutError:
      pass

asyncio.run(main())

虽然我很高兴这个问题现在已经解决了，但我不太理解这种行为。如果超过了超时时间，为什么要让这个任务保留一个引用呢？这似乎是造成内存泄漏的一个秘诀。关于不取消期货的说明我也不清楚。如果我们不明确取消 future，但我们只是不让任务在超时后保留引用，该怎么办？那不是也可以吗？

如果有人能对此有所启发，我们将不胜感激。多谢！

目标是在后台提取实时数据（例如每 5 秒）并在需要时提取到仪表板。这是我的代码。它有点工作，但我看到两个问题： 1.如果我将 st.write("TESTING!") 移到最后，由于 while 循环，它永远不会被执行。有办法改进吗？我可以想象，随着仪表板的增长，将会有多个页面/表格等。这不会提供太多的灵活性。2. async 函数中的 return px 行，我对此不太满意，因为我通过反复试验得到了正确的结果。抱歉，我是个新手，但如果有更好的方法，我将非常感激。

谢谢你！

import asyncio
import streamlit as st
import numpy as np

st.set_page_config(layout="wide")

async def data_generator(test):
    while True:
        with test:
            px = np.random.randn(5, 1)
        await asyncio.sleep(1)
        return px

test = st.empty()
st.write("TESTING!")

with test:
    while True:
        px = asyncio.run(data_generator(test))
        st.write(px[0])

我刚刚发现了 Python 3.11 的新功能，如 ExceptionGroup 和 TaskGroup，我对以下 TaskGroup 行为感到困惑：如果组内的一个或多个任务失败，则所有其他正常任务都将被取消，我没有机会更改该行为 示例：

async def f_error():
    raise ValueError()

async def f_normal(arg):
    print('starting', arg)
    await asyncio.sleep(1)
    print('ending', arg)


async with asyncio.TaskGroup() as tg:
    tg.create_task(f_normal(1))
    tg.create_task(f_normal(2))
    tg.create_task(f_error())

# starting 1
# starting 2
#----------
#< traceback of the error here >

在上面的例子中，我无法打印“ending 1”和“ending 2”。asyncio.gather(return_exceptions=True)同时，在发生错误时提供类似选项不取消剩余任务的选项将非常有用。

你可以说“如果你不想要这种取消行为，就不要使用TaskGroup”，但答案是我想使用新的异常组功能，并且它严格绑定到TaskGroup

所以问题是：

1. 我可以以某种方式在 asyncio 中使用异常组，而无需在 TaskGroup 中使用这种全有或全无的取消策略吗？
2. 如果对于前面的答案是“否”：为什么 python 开发人员消除了在 TaskGroup API 中禁用取消的可能性？

我正在使用 Python 3.10，我对asyncio.create_task.

在下面的示例代码中，无论我是否使用asyncio.create_task. 似乎没有什么区别。

如何确定何时使用以及使用与不使用相比asyncio.create_task有何优势？asyncio.create_task

import asyncio
from asyncio import sleep

async def process(index: int):
    await sleep(1)
    print('ok:', index)

async def main1():
    tasks = []
    for item in range(10):
        tasks.append(asyncio.create_task(process(item)))
    await asyncio.gather(*tasks)

async def main2():
    tasks = []
    for item in range(10):
        tasks.append(process(item)) # Without asyncio.create_task
    await asyncio.gather(*tasks)

asyncio.run(main1())
asyncio.run(main2())