假设我想使用 dnspython 为 Alexa 前 100 万站点发送各种类型的 DNS 查询（A、AAAA、NS、SOA、DNSKEY、NSEC3、DS 等）。
一项一项地执行此操作会花费一点时间，因为我要为每个站点发送多个查询。所以我希望在 python 3 中使用 asyncio 执行一些并行操作。
我经历了大卫的生成器/协程三部曲，http://www.dabeaz.com/talks.html，但我仍然不确定如何完成我的简单任务......
进一步来说，
results = dns.resolver.query('google.com','AAAA')
是一个阻塞函数调用，等待 DNS 回复返回。
如何在这段等待时间内发送其他查询而不使用线程？由于 DNS 查询通常是 udp 数据包，因此我认为 asyncio 可能会有所帮助。
pycares 似乎不支持我需要的所有记录类型，因此 pycares 和 aiodns 不适用于我的情况。
任何参考和想法都会有所帮助。
我的应用程序具有以下顶级代码（略有缩写）：
def main():
# Some setup code ...
try:
asyncio.run(my_coroutine())
except Exception as e:
print("Exiting due to exception {}: {}".format(type(e).__name__, e))
print("Coroutine finished")
# Some cleanup code ...
print("Shutdown complete")
在 Python 3.8 中asyncio.run永远不会完成，因此清理代码不会运行（并且不会打印关闭文本）。当应用程序应该退出时，它只是永远挂起。在 Python 3.7 中它运行良好。Python的3.6没有asyncio.run，但相当类似的代码loop.run_until_complete()和loop.close()工作过。
一些额外的上下文：设置和清理代码启动并优雅地退出使用threading.Thread. 正是这个线程实际上停止了主协程：它取消了协程（从技术上讲，它取消了在 内调用的另一个协程my_coroutine()）loop.call_soon_threadsafe
我正在编写一个具有以下功能的 CSV 文件：
import csv
import os
import aiofiles
async def write_extract_file(output_filename: str, csv_list: list):
"""
Write the extracted content into the file
"""
try:
async with aiofiles.open(output_filename, "w+") as csv_file:
writer = csv.DictWriter(csv_file, fieldnames=columns.keys())
writer.writeheader()
writer.writerows(csv_list)
except FileNotFoundError:
print("Output file not present", output_filename)
print("Current dir: ", os.getcwd())
raise FileNotFoundError
但是，由于没有 await 允许 overwriterows方法，因此没有将行写入 CSV 文件。
如何解决这个问题？有没有可用的解决方法？
谢谢你。
整个代码可以在这里找到。
注意事项：这是我第一个使用 asyncio 的方法，所以我可能做了一些非常愚蠢的事情。
场景如下：
我需要“http-ping”一个庞大的网址列表，以检查它们是否响应 200 或任何其他值。尽管像 gobuster 这样的工具报告 200,403 等，但我对每个请求都会超时。
我的代码与此类似：
import asyncio,aiohttp
import datetime
#-------------------------------------------------------------------------------------
async def get_data_coroutine(session,url,follow_redirects,timeout_seconds,retries):
#print('#DEBUG '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')+' '+url)
try:
async with session.get(url,allow_redirects=False,timeout=timeout_seconds) as response:
status = response.status
#res = await response.text()
if( status==404):
pass
elif(300<=status and status<400):
location = str(response).split("Location': \'")[1].split("\'")[0]
print('#HIT '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')+' '+str(status)+' '+url+' ---> '+location)
if(follow_redirects==True):
return await get_data_coroutine(session,location,follow_redirects,timeout_seconds,retries)
else:
print('#HIT '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')+' '+str(status)+' '+url)
return None
except asyncio.exceptions.TimeoutError as e:
print('#ERROR '+datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')+' '+' '+' '+url+' TIMEOUT '+str(e))
return None …我在我的 Python 应用程序中创建了一个字典，用于保存数据，并且我有两个任务同时运行并从外部 API 获取数据。一旦他们获得数据，他们就会更新字典——每个字典都有一个不同的键。
我想了解字典是否是异步安全的，还是在读取/更新字典时需要加锁？
任务还每次读取最后保存的值。
my_data = {}
asyncio.create_task(call_func_one_coroutine)
asyncio.create_task(call_func_two_coroutine)
async def call_func_one_coroutine():
data = await goto_api_get_data()
my_data['one'] = data + my_data['one']
async def call_func_two_coroutine():
data = await goto_api_another_get_data()
my_data['two'] = data + my_data['two']
我正在尝试制作一个在后台与多个 BLE 设备进行通信的 UI。为此，我实现了一个运行 asyncio.loop 的单独线程。这是必要的，因为我使用 bleak 0.9.1 连接到设备。
使用信号和槽从 UI 线程获取数据到工作线程工作正常。但是，它在另一个方向上不起作用。据我所知，这是因为线程忙于运行循环并且永远不会停止这样做。因此，它无法处理来自 UI 线程的输入。
下面是显示问题的示例代码。
有没有办法在运行异步循环时处理线程中的输入槽？
import sys
from PyQt5.QtWidgets import QWidget, QPushButton, QApplication, QVBoxLayout
from PyQt5.QtCore import QThread, QObject, pyqtSignal, pyqtSlot
import asyncio
class Test_Thread(QObject):
signal_back = pyqtSignal(int)
def __init__(self,
loop: asyncio.AbstractEventLoop,
parent=None):
super(Test_Thread, self).__init__(parent)
self.text = "Task1 not configured"
self.loop = loop
self.counter = 0
@pyqtSlot(str)
def set_text_slot(self, txt):
self.text = txt
async def do_stuff1(self):
while True:
print(self.text)
await asyncio.sleep(2.0)
async def do_stuff2(self):
while True:
self.counter += 1 …我的目标是使用音频流。从逻辑上讲，这是我的目标：
FastAPI端点）为了解决这个问题，我有很多想法，但最终我一直试图桥WebSocket A接到WebSocket B. 到目前为止，我的尝试涉及一个ConnectionManager类，其中包含一个Queue.queue. 音频流的块被添加到这个队列中，这样我们就不会直接从WebSocket A.
在ConnectionManager还包含一个发生器方法从队列中得到的所有值。
我的 FastAPI 实现websocket A是这样消耗的：
@app.websocket("/ws")
async def predict_feature(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
chunk = await websocket.receive_bytes()
manager.add_to_buffer(chunk)
except KeyboardInterrupt:
manager.disconnect()
在此摄取的同时，我希望有一个任务可以将我们的音频流桥接到WebSocket B，并将获得的值发送到WebSocket A。可以通过上述generator方法消耗音频流。
由于 WebSocket B 消费消息的方式，生成器方法是必要的，如 Rev-ai 的示例所示： …
鉴于main.py：
import asyncio
async def new_app():
# Await some things.
async def app(scope, receive, send):
...
return app
app = asyncio.run(new_app())
其次是：
uvicorn main.app
给出：
RuntimeError: asyncio.run() cannot be called from a running event loop
这是因为uvicorn在导入我的应用程序之前已经启动了一个事件循环。如何在 下异步构建应用程序uvicorn？
在 Python 中，有3 种主要类型的可等待对象：协程、任务和期货。
我可以await是协程，也可以是tasks.
等待协程
import asyncio
async def nested():
return 42
async def main():
print(await nested()) # will print "42".
asyncio.run(main())
等待任务
import asyncio
async def nested():
return 42
async def main():
task = asyncio.create_task(nested())
await task
asyncio.run(main())
首先将协程包装在任务中的价值是什么？看起来他们做同样的事情。
我什么时候需要使用任务与协程？
我有一个功能：
import time
def all_40k():
for _ in range(400000):
print('validate')
print('parsing')
print('inserting')
if __name__ == '__main__':
start_time = time.time()
all_40k()
print(f'used time:{time.time()-start_time}')
输出是：
used time:9.545064210891724
因为这个相同的函数重复了 40k 次，所以我希望有 4 个并行函数同时运行，每个函数运行 10k，理想情况下这会快 4 倍。
所以我首先尝试了多线程：
import threading
import time
def first_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def second_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def third_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
def forth_10k():
for _ in range(100000):
print('validate')
print('parsing')
print('inserting')
thread1 = threading.Thread(target=first_10k)
thread2 = threading.Thread(target=second_10k) …python multithreading python-multithreading python-asyncio python-multiprocessing