标签: 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。

下面是一个收集 URL 长度的简单程序。

import aiohttp
import asyncio
from time import perf_counter


URLS = ['http://www.cnn.com', 'http://www.huffpost.com', 'http://europe.wsj.com',
        'http://www.bbc.co.uk', 'http://failfailfail.com']

async def async_load_url(url, session):
    try:
        async with session.get(url) as resp:
            content = await resp.read()
        print(f"{url!r} is {len(content)} bytes")
    except IOError:
        print(f"failed to load {url}")        

async def main():

    async with aiohttp.ClientSession() as session:
        tasks = [async_load_url(url, session) for url in URLS]
        await asyncio.wait(tasks)

if __name__ == "__main__":
    start = perf_counter()
    asyncio.run(main())
    elapsed = perf_counter() - start
    print(f"\nTook {elapsed} seconds")

为什么以下代码在 python 3.9 中失败并出现运行时错误并忽略异常？如何修复它？ …

Binance API 和python-binance提供非阻塞执行的异步功能，如Binance 异步基础知识中所述。

我正在通过 websocket 使用BinanceSocketManager监听（异步非阻塞）实时数据。

在网络间歇性连接丢失等情况下，我希望为我的项目添加自动重新连接功能。但我似乎找不到任何信息BinanceSocketManager。我只能找到一个使用 的指南ThreadedWebsocketManager，但它不是异步实现。

有谁知道如何实现 Binance websocket 断开连接检测和自动重新连接机制？

这是我到目前为止所拥有的一些代码：

import asyncio
from binance import AsyncClient, BinanceSocketManager


async def main():
    client = await AsyncClient.create()
    await kline_listener(client)

async def kline_listener(client):
    bm = BinanceSocketManager(client)
    async with bm.kline_socket(symbol='BTCUSDT') as stream:
        while True:
            res = await stream.recv()
            print(res)
    # a way detect websocket error/disconnect, callback 'disconnect_callback'

async def disconnect_callback():
    await client.close_connection()
    await main()  # restart client and kline …

我试图使用fastapi users包快速向使用 PostgreSQL 数据库的 FastAPI 项目添加注册和身份验证系统。我用来asyncio创建异步函数。

一开始，我只使用了 sqlAlchemy，并且我在这里尝试了他们的示例。我将这些代码行添加到我的 app/app.py 中，以在服务器启动时创建数据库。一切都很顺利。表 users 是在我的数据库上创建的。

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

由于我使用的是 SQLModel，因此我将FastAPI Users - SQLModel 的数据库适配器添加到我的虚拟 en 包中。我添加了这些行以便fastapi_users/db/__init__.py能够使用 SQL 模型数据库。

try:
    from fastapi_users_db_sqlmodel import (  # noqa: F401
        SQLModelBaseOAuthAccount,
        SQLModelBaseUserDB,
        SQLModelUserDatabase,
    )
except ImportError:  # pragma: no cover
    pass

我还修改了app/users.py, 以SQLModelUserDatabase代替 sqlAchemy 之一。

async def get_user_manager(user_db: SQLModelUserDatabase = Depends(get_user_db)):
    yield UserManager(user_db)

以及app/dp.py使用SQLModelUserDatabase，，SQLModelBaseUserDB这里是完整的代码 …

我正在运行“workers”协程，其工作是等待 5 秒，从 asyncio.Queue() 获取值并不断打印出来。

q = asyncio.Queue()

def worker():
    while True:
        await asyncio.sleep(5)
        i = await q.get()
        print(i)
        q.task_done()

async def main(q):
    workers = [asyncio.create_task(worker()) for n in range(10)]
    await asyncio.gather(*workers)


if __name__ == "__main__":
    asyncio.run(main())

我希望能够使用 FastAPI 通过 http 请求与队列进行交互。例如，POST 请求会将项目“放入”队列中以供工作人员打印。

我不确定如何与 FastAPI 同时运行工作线程的协程来实现这种效果。我相信 Uvicorn 有自己的事件循环，并且我尝试使用 asyncio 方法没有成功。

我认为路由器看起来像这样。

@app.post("/")
async def put_queue(data:str):
    return q.put(data)

我希望有一些东西会产生这样的效果：

await asyncio.gather(main(),{FastApi() app run})

我已经使用异步实现了所有路线。并遵循 FastAPI 文档中的所有准则。

每个路由都有多个数据库调用，没有异步支持，所以它们是这样的正常功能

def db_fetch(query):
    # I take a few seconds to respond
    return 

为了避免阻塞我的事件循环，我使用fastapi.concurrancy.run_in_threadpool

现在的问题是，当大量请求到来时，我的新请求就会被阻止。即使我关闭浏览器选项卡（取消请求），整个应用程序也会卡住，直到处理旧的请求为止。

我在这里做错了什么？

我用它uvicorn作为我的 ASGI 服务器。我在具有 2 个副本的 kubernetes 集群中运行。

很少有人怀疑：我是否产生了太多线程？这是 uvicron 中的一些错误吗？不太确定！

（以下代码可以在Jupyter中运行。）我有一个B类，它使用A类，需要测试。

class A:
    async def f(self):
        pass

class B:
    async def f(self):
        a = A()
        x = await a.f()  # need to be patched/mocked

我有以下测试代码。看起来它模拟了 A 的类方法而不是实例方法。

from asyncio import Future
from unittest.mock import MagicMock, Mock, patch

async def test():
    sut = B()
    with patch('__main__.A') as a:  # it's __main__ in Jupyter
        future = Future()
        future.set_result('result')
        a.f = MagicMock(return_value=future)
        await sut.f()

await test()

但是，代码出现以下错误：

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
C:\Users\X~1\AppData\Local\Temp\1/ipykernel_36576/3227724090.py in <module>
     20         await sut.f()
     21 
---> 22 …