Python Asyncio run_forever() 和任务

我改编了这段代码，以便在异步 Python 中使用 Google Cloud PubSub：https://github.com/cloudfind/google-pubsub-asyncio 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())线。我认为任务是按预期创建的，但它们从未真正运行过。仅当我添加上述行时，代码才能正确执行并且所有创建的任务都会运行。是什么导致了这种行为？

PubSub 正在调用create_proc_message_task来自不同线程的回调。自从create_task is 不是线程安全的 https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading，它只能从运行事件循环的线程（通常是主线程）调用。要纠正该问题，请更换loop.create_task(proc_message(message)) with asyncio.run_coroutine_threadsafe(proc_message(message), loop) and message_producer将不再需要。

至于为什么message_producer似乎修复了代码，请考虑run_coroutine_threadsafe https://docs.python.org/3/library/asyncio-task.html#asyncio.run_coroutine_threadsafe与create_task:

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

在你的情况下create_task将任务添加到循环的可运行队列中（没有任何锁定），但无法确保唤醒，因为在事件循环线程中运行时不需要唤醒。这message_producerthen 用于强制循环定期唤醒，此时它还检查并执行可运行任务。