标签: python-asyncio

我目前正在尝试使用websockets 库。如果另一个图书馆更适合此目的，请告诉我。

鉴于这些功能：

def handle_message(msg):
    # do something

async def consumer_handler(websocket, path):
    async for message in websocket:
        handle_message(message)

我如何（无限期地）连接到多个 websockets？下面的代码会起作用吗？

import asyncio
import websockets


connections = set()
connections.add(websockets.connect(consumer_handler, 'wss://api.foo.com', 8765))
connections.add(websockets.connect(consumer_handler, 'wss://api.bar.com', 8765))
connections.add(websockets.connect(consumer_handler, 'wss://api.foobar.com', 8765))

async def handler():
    await asyncio.wait([ws for ws in connections])

asyncio.get_event_loop().run_until_complete(handler())

我为我的 Discord 服务器创建了一个机器人，它转到给定 subreddit 的 Reddit API，并根据您输入的 subreddit 在 Discord 聊天中发布当天的前 10 个结果。它无视自己的帖子，实际上只发布图片和 GIF。Discord 消息命令看起来像这样：=get funny awww news programming发布每个 subreddit 的结果，因为它从 Reddit API (PRAW) 获取它们。这没有问题。我知道 bot 能够访问 API 并发布到 discord 中。

我添加了另一个命令=getshuffled，它将来自 subreddits 的所有结果放在一个大列表中，然后在发布之前将它们打乱。这对于最多约 50 个 subreddits 的请求非常有效。

这是我需要帮助的：

因为它可以是如此大的结果列表，来自 100 多个 subreddit 的 1000 多个结果，机器人在非常大的请求上崩溃。根据我昨天从我的问题中得到的帮助，我明白出了什么问题。机器人正在启动，它正在与我的 Discord 服务器通信，当我向它传递一个很长的请求时，它在 Reddit API 调用完成后停止与服务器通信的时间太长，并且 Discord 连接失败。

因此，我认为我需要做的是为转到 Reddit API 并提取结果的代码创建一个子进程（我认为这将使不和谐连接保持运行），然后将这些结果传回给机器人当它完成时......

或者...这是 Asyncio 可以自行处理的事情...

正如我所知道的那样，我在子流程调用方面遇到了困难。

基本上，我要么需要关于这个子流程技巧的帮助，要么需要知道我是否是个白痴，而 Asyncio 可以为我处理所有这些。我认为这只是“我不知道我不知道什么”的例子之一。

总结一下：该机器人运行良好，改组了少量 subreddit。它通过发送的参数（它们是 subreddits），获取每个帖子的信息，然后在将链接发布到不和谐之前进行洗牌。问题是当它是一个更大的大约 50+ 的 subreddit 集时。为了让它与更大的数量一起工作，我需要让 Reddit …

我正在尝试使用 asyncio 将基于回调的 Python 代码中的同步流转换为 A 同步流。基本上，代码与 TCP/UNIX 套接字交互很多。它从套接字读取数据，对其进行操作以做出决定并将内容写回另一端。这是一次在多个套接字上进行的，并且有时在上下文之间共享数据以做出决定。

编辑::目前的代码主要基于为特定套接字注册一个中央实体的回调，并在相关套接字可读时让该实体运行回调（类似于“当该套接字有数据要读取时调用此函数” ”）。一旦回调被调用 - 会发生一堆事情，最终当新数据可用时会注册一个新的回调。中央实体对所有注册的套接字运行一个选择，以确定应该调用哪些回调。

我试图在不重构我的整个代码并使程序员尽可能无缝的情况下做到这一点 - 所以我试图这样思考 - 所有代码都应该像今天一样运行 - 但只要当前代码执行 socket.recv() 以获取新数据 - 该过程将执行其他任务。当读取返回时，它应该返回使用它获得的新数据处理来自同一点的数据。

为此，我编写了一个名为 AsyncSocket 的新类——它与 asyncIO 的 IO 流交互，并将 Async/await 语句几乎完全放在那里——认为我将在我的类中实现 recv 方法以使其看起来像一个“常规 IO 套接字”到我的其余代码。到目前为止 - 这是我对异步编程应该允许的理解。

现在问题来了：

我的代码等待客户端连接 - 当它连接时，允许每个客户端的上下文从它自己的连接读取和写入。我已经简化到以下流程以澄清问题：

class AsyncSocket():
    def __init__(self,reader,writer):
        self.reader = reader
        self.writer = writer
    def recv(self,numBytes):
        print("called recv!")
        data = self.read_mitigator(numBytes)
        return data
    async def read_mitigator(self,numBytes):
        print("Awaiting of AsyncSocket.reader.read")
        data = await self.reader.read(numBytes)
        print("Done Awaiting of AsyncSocket.reader.read data is %s " …

我正在关注asyncio.Queue. 在由worker_task()我调用的函数中，我想记录当前的工作人员名称，而不必将其传递到任何地方。

我经历的某种组合怀疑LoggerAdapter，并ContextVar可以做我想做的。

以下内容...

import logging
logging.basicConfig(format="%(worker_name)s %(message)s")
logger = logging.getLogger(__name__)

async def sub_function():
    logger.info("a message") # worker-1: a message

async def worker_task(queue, worker_name):
    logger.do_something(worker_name) # HELP HERE PLEASE
    await sub_function()

queue = asyncio.Queue()

workers = [
    loop.create_task(worker_task(queue, "worker-1"),
    loop.create_task(worker_task(queue, "worker-2"),
]

当使用编写异步爬虫asyncio和aiohttpPython中，我一直有一个疑问：为什么你必须使用async with，并可以很容易地报告错误，如果你不使用它们。

虽然aiohttp也有方法request，但是可以支持调用更简单的api。我想知道有什么区别。还是requests挺喜欢模块的，不知道能不能像requests模块一样简单使用。

此代码失败：

import asyncio
from motor import motor_asyncio


_client = motor_asyncio.AsyncIOMotorClient()
_db = _client.db

users = _db.users


async def main():
    await users.create_index(
        'login',
        unique=True
    )


if __name__ == '__main__':
    #loop = asyncio.get_event_loop()
    #loop.run_until_complete(main())
    asyncio.run(main())

出现此错误：

Traceback (most recent call last):
  File "/home/sanyash/myrepos/TKP/db.py", line 21, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/usr/local/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "/home/sanyash/myrepos/TKP/db.py", line 14, in main
    unique=True
RuntimeError: Task <Task pending coro=<main() running at /home/sanyash/myrepos/TKP/db.py:14> cb=[_run_until_complete_cb() at /usr/local/lib/python3.7/asyncio/base_events.py:158]> …

我正在尝试实现 Python aiokafka 异步库，但由于某种原因我无法异步处理消息。

我创建了异步消费者、生产者并使用了 asyncio python 库。

环境：

 python 3.7.2
aiokafka==0.5.1
kafka-python==1.4.3

消费者：

from aiokafka import AIOKafkaConsumer
import asyncio
import json
import ast

loop = asyncio.get_event_loop()

async def consume():
    consumer = AIOKafkaConsumer(
        "test_topic", loop=loop, bootstrap_servers='localhost:9092')
    # Get cluster layout and topic/partition allocation
    await consumer.start()
    try:
        async for msg in consumer:
            sleep_time = ast.literal_eval(json.loads(msg.value))
            print('before sleep %s' % sleep_time)
            await asyncio.sleep(sleep_time)
            print('after sleep %s' % sleep_time)
    finally:
        await consumer.stop()

loop.run_until_complete(consume())

制作人：

import json
import uuid
from kafka import KafkaProducer, KafkaConsumer …

我有这个程序

import time
import asyncio


def is_prime(x):

    return not any(x//i == x/i for i in range(x-1, 1, -1))


async def highest_prime_below(x):

    print('Highest prime below %d' % x)
    for y in range(x-1, 0, -1):
        if is_prime(y):
            print('? Highest prime below %d is %d' % (x, y))
            return y
        await asyncio.sleep(0.01)
    return None


async def main():

    t0 = time.time()
    await asyncio.wait( [
        highest_prime_below(100000),
        highest_prime_below(10000),
        highest_prime_below(1000)
        ] )
    t1 = time.time()
    print('Took %.2f ms' % (1000*(t1-t0)))


loop = asyncio.get_event_loop()
loop.run_until_complete(main())
#loop.close()

发现素数。你也可以在这里看到它https://osf.io/w8u26/ …

下面的代码失败了，我无法得到实际的错误，我只是得到了许多 CancelledError 消息

import aiobotocore, asyncio

async def replicate_to_region(chunks, region):
    session = aiobotocore.get_session()
    client = session.create_client('dynamodb', region_name=region)
    start = 0
    while True:
        chunk = chunks[start]
        item = {'my_table': chunk}
        response = await client.batch_write_item(RequestItems=item)

async def main():
    asyncio.gather(*(replicate_to_region(payloads, region) for region in regions))

asyncio.run(main())

我收到以下错误；

client_session: <aiohttp.client.ClientSession object at 0x7f6fb65a34a8>
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f6fb64c82b0>
_GatheringFuture exception was never retrieved
future: <_GatheringFuture finished exception=CancelledError()>
concurrent.futures._base.CancelledError
_GatheringFuture exception was never retrieved
future: <_GatheringFuture finished exception=CancelledError()>

我已经尝试了相当多的replicate_to_region函数变体，但它们都因上述相同的错误而失败。能够看到实际错误是什么会很有用。

如何使用所有 cpu 内核进行 asyncio - 除了 ProcessPoolExecutor 之外的任何其他选项？

我认为 asyncio 无法打破 GIL 限制（也许我错了），因此程序的执行速度将比踩踏版本快，但会在一个核心上执行。

我研究了一些例子，我发现一种方法是多处理和 ProcessPoolExecutor。

https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor

# 3. Run in a custom process pool:
with concurrent.futures.ProcessPoolExecutor() as pool:
    result = await loop.run_in_executor(
        pool, cpu_bound)
    print('custom process pool', result)

这很好，但需要在进程之间进行“pickle”，因此需要一些开销并对传递的参数进行一些优化以减少“pickle”序列化。

使用上面这个简单的模式，我写了这样的测试代码（如果你不喜欢它，你可以跳过这段代码阅读，因为它和以前一样）。顺便说一句，这是我解析文件问题的最快解决方案。这部分代码不是整个程序。

def _match_general_and_specific_file_chunk(file_name):
    with codecs.open(file_name, encoding='utf8') as f:
        while True:
            lines = f.readlines(sizehint=10000)
            if not lines:
                break
            for line in lines:
                general_match = RE_RULES.match(line)
                if general_match:
                    specific_match = RULES[general_match.lastindex].match(line)
                    groups = list(specific_match.groups())
                    continue


async def _async_process_executor_match_general_and_specific_read_lines_with_limit_file_chunk():
    loop = asyncio.get_event_loop()
    with ProcessPoolExecutor() …