这是使用 Python `asyncio` 协程的预期模式吗？

Question

我正在尝试使用 Slack RTM API 编写一个小的并发流处理程序，我想知道这是否是最有效地使用 Python 协程。 asyncio 包有很多选项，但很难确定项目的正确方法是什么，我认为文档并不能很好地解释每个项目的优点/缺点。

我认为我这里不需要多线程的开销，我需要异步循环之间的相互通信。我应该为我的每个函数创建一个单独的BaseEventLoop 吗？

作为 Python，我认为对于这个问题 (There should be one-- and preferably only one --obvious way to do it) 有一个接近的确定性答案，但我担心添加所有这些异步杂乱可能只会使我的代码性能低于完全顺序的幼稚实现。

# Is this the best way to communicate between coroutines? 
incoming_message_q = asyncio.Queue()

async def print_event():
    logging.info("Entering log loop")

    # Should this operate within it's own BaseEventLoop? 
    while True:
        event = await incoming_message_q.get()
        logging.info(event)

async def log_queue_status():
    while True:
        logging.info(incoming_message_q.qsize())
        await asyncio.sleep(5)

async def read_rtm_connection(client, q):
    if client.rtm_connect():
        logging.info("Successful Slack RTM connection")
        while True:

            # How do I make this part non-blocking?
            events = client.rtm_read()

            for event in events:
                logging.info("Putting onto the queue", event)
                if event["type"] == "presence_change":
                    await q.put(event)
                elif event["type"] == "message":
                    await q.put(event)
                else:
                    logging.info("Not sure what to do")

            await asyncio.sleep(0.1)
    else:
        logging.info("RTM connection failed.")

loop = asyncio.get_event_loop()
loop.create_task(print_event())
loop.create_task(log_queue_status())
loop.create_task(read_rtm_connection(client, incoming_message_q))
loop.run_forever()

Answer 1

如果您想以一种异步友好的方式与 slack 交互，您将需要使用非阻塞 API。我不确定您当前使用的是什么，但如果它不包含任何 asyncio 协程，它可能不会轻易集成到 asyncio 中，除非您通过 @ 在后台线程中运行所有阻塞调用987654324@。另一种选择是将库中的所有底层阻塞 I/O 调用实际转换为非阻塞，这通常是一大堆工作。

好消息是至少有一个图书馆已经为您完成了这项工作； slacker-asyncio，它是slacker 的一个分支。您应该能够使用它通过协程与 RTM API 进行交互。