标签: python-asyncio

使用 asyncio async/await 和 Python 3.7+ 提供 100k API 调用的最佳方法是什么？想法是始终并行使用 100 个任务？

应该避免的是：
1. 开始处理所有 100k 任务
2. 等待所有 100 个并行任务完成，以便安排新的 100 个任务。

这个例子说明了第一种方法，这不是所需要的。

import aiohttp
import asyncio

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
            'http://python.org',
            'https://google.com',
            'http://yifei.me'
        ]
    tasks = []
    async with aiohttp.ClientSession() as session:
        for url in urls:
            tasks.append(fetch(session, url))
        htmls = await asyncio.gather(*tasks)
        for html in htmls:
            print(html[:100])

if __name__ == '__main__':
    loop …

是否可以获取使用 aiohttp 发出的每个请求的响应时间和响应大小？

文档似乎在任何地方都没有这些属性。

谢谢

我有一个异步函数来从站点获取数据：

async def get_matches_info(url):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(url, proxy=proxy) as response:
                ...
                ...
                ...
                ...
        except:
            print('ERROR GET URL: ', url)
            print(traceback.print_exc())

我有一个大约 200 个链接的列表。几乎总是一切正常，但有时我会收到以下错误：

Traceback (most recent call last):
  File "C:\Python37\lib\site-packages\aiohttp\connector.py", line 924, in _wrap_create_connection
    await self._loop.create_connection(*args, **kwargs))
  File "C:\Python37\lib\asyncio\base_events.py", line 986, in create_connection
    ssl_handshake_timeout=ssl_handshake_timeout)
  File "C:\Python37\lib\asyncio\base_events.py", line 1014, in _create_connection_transport
    await waiter
ConnectionResetError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "parser.py", line 90, …

我想要一个类函数暂停执行，直到另一个线程根据函数的请求更改类变量的值。我是 python asyncio 模块的新手。

asyncio.Future 似乎提供了一种等待未来价值的机制，因此我尝试了一个玩具示例：

class RandomReader:
    def __init__(self):
        self.loop = asyncio.get_event_loop()
        self.service = 3
        self.thread = threading.Thread(target=self.reader)
        self.thread.start()
        self.futures: Dict[int, asyncio.Future] = {}

    def reader(self):
        asyncio.set_event_loop(self.loop)
        while self.service != 0:
            k, v = read()
            if k in self.futures:
                if self.futures[k].done():
                    continue

                self.futures[k].set_result(v)
                self.service -= 1

    async def wait(self, v: int):
        self.futures[v] = self.loop.create_future()
        a = await self.futures[v]
        logging.debug("value %d received %f", v, a)
        return v, a

read上面的函数读取可能与wait.

调用函数使调用如下 3 次 ( RandomReader.service)

    t1 = …

我正在尝试从我使用包装器的 api 获得一些响应。

我寻找了更多错误，但他们没有解决我的问题。我不太了解 await 函数或协程。我关闭了 is_asnyc，它起作用了，但我需要那个选项。所以我不能关闭它。

import clashroyale, asyncio
token = "my token"
cr = clashroyale.official_api.Client(token=token, is_async=True)

async def top():
    p = await cr.get_top_players()
    return p

topplayers = asyncio.run(top())

我排除了它只是从api获取信息但出现错误-

RuntimeError: Timeout context manager should be used inside a task
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x000001774361A3C8>

我是 Python 和这些库/模块的新手。我正在编写一个简单的 ping 测试网络扫描仪作为学习项目。

我首先使用 asyncio 开发了一个脚本来 ping 网络上的地址

#ip_test.py
import asyncio
import ipaddress

async def ping(addr):
    proc = await asyncio.create_subprocess_exec(
        'ping','-W','1','-c','3',addr,
        stdout=asyncio.subprocess.PIPE
    )
    await proc.wait()
    return proc.returncode

async def pingMain(net):
    #hosts() returns list of Ipv4Address objects
    result = await asyncio.gather(*(ping(str(addr)) for addr in net.hosts()))
    return result

def getHosts(net_): #net_ is an Ipv4Network object
    return asyncio.run(pingMain(net_))
    #Returns list of response codes which I then zip with the list of searched ips

当我打开 python 并运行以下命令时，它按预期工作：

import ip_test as iptest
import …

我希望能够从一组运行的任务中产生结果，gather因为它们进入进一步处理。

# Not real code, but an example

async for response in asyncio.gather(*[aiohttp.get(url) for url in ['https://www.google.com', 'https://www.amazon.com']]):
    await process_response(response)

目前，我可以使用该gather方法全部get并发运行，但必须等到它们全部完成才能处理它们。我还是 Python async/await 的新手，所以也许我缺少一些明显的方法来做到这一点。

# What I can do now
responses = await asyncio.gather(*[aiohttp.get(url) for url in ['https://www.google.com', 'https://www.amazon.com']])
await asyncio.gather(*[process_response(response) for response in responses])

谢谢！

import asyncio
from threading import Thread
from datetime import datetime
from aiogram import Bot, Dispatcher, executor, types

API_TOKEN = ''

bot = Bot(token=API_TOKEN)
dp = Dispatcher(bot)

chat_ids = {}

@dp.message_handler()
async def echo(message: types.Message):
    # old style:
    # await bot.send_message(message.chat.id, message.text)

    chat_ids[message.message_id] = message.from_user
    text = f'{message.message_id} {message.from_user} {message.text}'
    await message.reply(text, reply=False)


async def periodic(sleep_for, queue):
    while True:
        await asyncio.sleep(sleep_for)
        now = datetime.utcnow()
        print(f"{now}")
        for id in chat_ids:
            queue.put_nowait((id, f"{now}"))
            # await bot.send_message(id, f"{now}", disable_notification=True)


def run_tick(queue):
    newloop = asyncio.new_event_loop() …

我正在尝试使用 asyncio 从数千个 url 中获取一些数据。以下是该设计的简要概述：

1. Queue一次性填写一堆网址Producer
2. 产生一堆 Consumers
3. 每个都Consumer保持异步从请求中提取 urlQueue并发送GET请求
4. 对结果做一些后处理
5. 合并所有处理结果并返回

问题： asyncio几乎从不显示是否有任何问题，它只是默默地挂起而没有错误。我在print各处放置语句来自己检测问题，但这并没有多大帮助。

根据输入网址的数量和消费者的数量或限制，我可能会收到以下错误：

1. Task was destroyed but it is pending!
2. task exception was never retrieved future: <Task finished coro=<consumer()
3. aiohttp.client_exceptions.ServerDisconnectedError
4. aiohttp.client_exceptions.ClientOSError: [WinError 10053] An established connection was aborted by the software in your host machine

问题：如何检测和处理中的异常asyncio？如何在不中断的情况下重试Queue？

Bellow 是我在查看异步代码的各种示例时编译的代码。目前，def get_video_title函数末尾存在故意错误。运行时，什么都不显示。

import asyncio
import aiohttp
import json
import re
import nest_asyncio
nest_asyncio.apply() …

我的程序执行以下操作：

1. 获取 .txt 文件的文件夹

2. 对于每个文件：

   2.1. 读取文件

   2.2 将内容排序为列表并将列表推送到主列表

我在没有任何 async/await 的情况下这样做了，这些是时间统计信息

real    0m0.036s

user    0m0.018s

sys     0m0.009s

使用下面的异步/等待代码我得到

real    0m0.144s

user    0m0.116s

sys     0m0.029s

考虑到用例表明我错误地使用了 aysncio。

有人知道我做错了什么吗？

import asyncio
import aiofiles
import os

directory = "/tmp"
listOfLists = list()

async def sortingFiles(numbersInList):
    numbersInList.sort()

async def awaitProcessFiles(filename,numbersInList):
    await readFromFile(filename,numbersInList)
    await sortingFiles(numbersInList)
    await appendToList(numbersInList)


async def readFromFile(filename,numbersInList):
    async with aiofiles.open(directory+"/"+filename, 'r') as fin:
        async for line in fin:
            return numbersInList.append(int(line.strip("\n"),10))            
    fin.close()    

async def appendToList(numbersInList):
    listOfLists.append(numbersInList)

async def main():
    tasks=[]
    for filename …