Python Asyncio run_forever()和任务 [英] Python Asyncio run_forever() and Tasks

问题描述

我修改了此代码，以在异步Python中使用Google Cloud PubSub: https://github. com/cloudfind/google-pubsub-asyncio

I adapted this code for using Google Cloud PubSub in Async Python: https://github.com/cloudfind/google-pubsub-asyncio

import asyncio
import datetime
import functools
import os

from google.cloud import pubsub
from google.gax.errors import RetryError
from grpc import StatusCode

async def message_producer():
    """ Publish messages which consist of the current datetime """
    while True:
        await asyncio.sleep(0.1)


async def proc_message(message):
    await asyncio.sleep(0.1)
    print(message)
    message.ack()


def main():
    """ Main program """
    loop = asyncio.get_event_loop()

    topic = "projects/{project_id}/topics/{topic}".format(
        project_id=PROJECT, topic=TOPIC)
    subscription_name = "projects/{project_id}/subscriptions/{subscription}".format(
        project_id=PROJECT, subscription=SUBSCRIPTION)

    subscription = make_subscription(
        topic, subscription_name)

    def create_proc_message_task(message):
        """ Callback handler for the subscription; schedule a task on the event loop """
        print("Task created!")
        task = loop.create_task(proc_message(message))

    subscription.open(create_proc_message_task)
    # Produce some messages to consume

    loop.create_task(message_producer())

    print("Subscribed, let's do this!")
    loop.run_forever()


def make_subscription(topic, subscription_name):
    """ Make a publisher and subscriber client, and create the necessary resources """
    subscriber = pubsub.SubscriberClient()
    try:
        subscriber.create_subscription(subscription_name, topic)
    except:
        pass
    subscription = subscriber.subscribe(subscription_name)

    return subscription


if __name__ == "__main__":
    main()

我基本上删除了发布代码，仅使用订阅代码. 但是，最初我没有包含loop.create_task(message_producer())行.我认为任务是按照预期的方式创建的，但实际上它们从未真正运行过.仅当我添加上述行后，代码才能正确执行并且所有创建的任务都将运行.是什么原因导致这种行为?

I basically removed the publishing code and only use the subscription code. However, initially I did not include the loop.create_task(message_producer()) line. I figured that tasks were created as they were supposed to however they never actually run themselves. Only if I add said line the code properly executes and all created Tasks run. What causes this behaviour?

推荐答案

PubSub从另一个线程调用create_proc_message_task回调.由于create_task是不是线程安全，它只能从运行事件循环的线程(通常是主线程)中调用.要解决此问题，请将loop.create_task(proc_message(message))替换为asyncio.run_coroutine_threadsafe(proc_message(message), loop)，并且不再需要message_producer.

PubSub is calling the create_proc_message_task callback from a different thread. Since create_task is not thread-safe, it must only be called from the thread that runs the event loop (typically the main thread). To correct the issue, replace loop.create_task(proc_message(message)) with asyncio.run_coroutine_threadsafe(proc_message(message), loop) and message_producer will no longer be needed.

至于为什么message_producer似乎可以修复代码，请考虑

As for why message_producer appeared to fix the code, consider that run_coroutine_threadsafe does two additional things compared to create_task:

• 它以线程安全的方式运行，因此并发执行时事件循环数据结构不会损坏.
• 它确保事件循环尽可能快地唤醒，以便它可以处理新任务.

在您的情况下，create_task将任务添加到了循环的可运行队列中(没有任何锁定)，但是未能确保唤醒，因为在事件循环线程中运行时不需要这样做.然后message_producer迫使循环定期唤醒，这也是它同时检查并执行可运行任务的时间.

In your case create_task added the task to the loop's runnable queue (without any locking), but failed to ensure the wakeup, because that is not needed when running in the event loop thread. The message_producer then served to force the loop to wake up in regular intervals, which is when it also checks and executes the runnable tasks.

这篇关于Python Asyncio run_forever()和任务的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！