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

2019年7月27日 174次阅读

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

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

作为Python,我认为这个问题有一个接近确定性的答案(应该有一个 – 最好只有一个 – 显而易见的方法)但是我担心添加所有这些异步错误可能只是让我代码性能低于完全顺序的天真实现.

# 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()

最佳答案 如果您想以asyncio友好的方式与slack进行交互,那么您将需要使用非阻塞API.我不确定你目前使用的是什么,但如果它不包含任何asyncio协同程序,它可能不会轻易集成到asyncio中,除非你在后台线程中运行所有阻塞调用,通过loop.run_in_executor .另一种选择是将库中的所有底层阻塞I / O调用实际转换为非阻塞,这通常是一大堆工作.

好消息是,至少有一个图书馆已经为您完成了这项工作; slacker-asyncio,这是一个懒散的分叉.您应该能够使用它通过协同程序与RTM API进行交互.

点赞