python - 在另一个文件中使用 Asyncio 循环引用

Question

您好，我有以下脚本，它监听 Google pubsub 消息，在回调中我将消息传递给另一个带有循环参数的脚本。

在第二个脚本中，我有一个具有依赖 future 功能的任务。但是 future 函数永远不会被调用。

import os, time
import base64, json
import asyncio
from google.cloud import pubsub_v1
from app.libs.pubsub.pubsub_connect import PubsubConnect
from app.config import config
from evergage.scripts.evga_process import initEvga
from app.utils import log

LOG = log.get_logger()

PROJECT_NAME = config.get('pubsub', 'project_name')
SUBSCRIPTION = config.get('pubsub', 'insights_subscription')


class PubsubConsumer(PubsubConnect):

    _subscriber = None

    def __init__(self, loop):
        self._subscriber = self.getClientService()
        #loop.create_task( self.getMessages(loop) )
        self.getMessages(loop)


    def getMessages(self, loop):

        def consumeMessageTask(message):
            #loop.create_task( self.callback(message, loop) )
            loop.call_soon( self.callback(message, loop) )


        subscription_path = self._subscriber.subscription_path( PROJECT_NAME, SUBSCRIPTION )
        flow_control = pubsub_v1.types.FlowControl(max_messages=10)
        self._subscriber.subscribe(subscription_path, callback=consumeMessageTask, flow_control=flow_control)

        #loop.run_forever()


    def callback(self, message, loop):

        pubSubMsg = message.data.decode("utf-8")
        pubSubMsg = json.loads( base64.urlsafe_b64decode( pubSubMsg) )

        if pubSubMsg['verb']['display'] == 'evga_upsert':
            loop.create_task( initEvga(pubSubMsg, loop) )

        message.ack()


asyncLoop = asyncio.get_event_loop()
test = PubsubConsumer(asyncLoop)
asyncLoop.run_forever()
asyncLoop.close()

第二个文件evga_process.py

import os, time
import base64, json
import asyncio
import functools
from app.config import config
from app.utils import log

LOG = log.get_logger()
MKTO_BATCH = {
    'count': 0,
    'total': 5,
    'data': []
}


async def getMarketoData(cookies):
    #await asyncio.sleep(0.1)
    tmpData = []

    for cookie in cookies:
        tmpData.append("test-" + cook)

    return tmpData


def initSalesforce(future, param1):
    result = future.result()
    LOG.info( param1 )
    LOG.info( str(result) )
    pass


def initEvga(pubSubMsg, loop):
    pubSubID = pubSubMsg['id']
    upsertInfo = pubSubMsg['object']['definition']['description']

    MKTO_BATCH['data'].append(pubSubID)

    MKTO_BATCH['count'] += 1

    if MKTO_BATCH['count'] == MKTO_BATCH['total']:
        mktoTask = loop.create_task( getMarketoData(MKTO_BATCH['data']) )
        mktoTask.add_done_callback( functools.partial( initSalesforce, "myparam1") )

        MKTO_BATCH['count'] = 0
        MKTO_BATCH['data'] = []

我什至尝试使用简单的代码，但它不起作用。它只工作，如果我这样打电话

loop = asyncio.get_event_loop()
task = loop.create_task( test_task(loop) )
initEvga(loop)
task.add_done_callback(got_result)

我在这里遗漏了什么吗？

---

单个文件中的更新脚本<​​/p>

async def heartbeat1():
    while True:
        await asyncio.sleep(1)
        print("heartbeat 1")

async def heartbeat2():
    while True:
        await asyncio.sleep(1)
        print("heartbeat 2")


class PubsubConsumer(PubsubConnect):

    _subscriber = None

    def __init__(self, loop):
        self._subscriber = self.getClientService()

    def getMessages(self, loop):

        def consumeMessageTask(message):
            #loop.create_task( self.callback(message, loop) )
            #loop.call_soon( self.callback(message, loop) )
            PubsubConsumer.callback(message, loop)

        subscription_path = self._subscriber.subscription_path( PROJECT_NAME, SUBSCRIPTION )

        #subClient.subscribe(subscription_path, callback=PubsubConnect.callback)
        # Limit the subscriber to only have ten outstanding messages at a time.
        flow_control = pubsub_v1.types.FlowControl(max_messages=10)
        self._subscriber.subscribe(subscription_path, callback=consumeMessageTask, flow_control=flow_control)

        #loop.run_forever()

    @staticmethod
    def callback(message, loop):
        time.sleep(1)
        pubSubMsg = message.data.decode("utf-8")
        pubSubMsg = json.loads( base64.urlsafe_b64decode( pubSubMsg) )

        loop.create_task( heartbeat2() )

        message.ack()

loop = asyncio.get_event_loop()

loop.create_task( heartbeat1() )

psClient = PubsubConsumer(loop)
psClient.getMessages(loop)

loop.run_forever()
loop.close()

heartbeat1() 和 heartbeat2() 都正常调用。但是，如果我删除 heartbeat1() 调用，则 heartbeat2() 也不会触发。

这是什么原因？如果我让 heartbeat1() 一直运行，它会产生任何问题吗？

谢谢， 巴拉

Answer 1

这一行绝对是不正确的:

mktoTask.add_done_callback( functools.partial( initSalesforce, "myparam1")

您不能将异步函数发送到 add_done_callback，它需要一个常规函数。调用时，异步/协程函数返回一个协程对象，而不执行其中的任何代码。要实际执行代码，协程对象必须提交给事件循环。在您的情况下，它将被删除而不执行函数内部的代码。 (这很像调用生成器，但永远不会将返回的迭代器传递给将从中提取值的对象。)

你需要的是这样的:

async def initEvga(pubSubMsg):
    ...
    if MKTO_BATCH['count'] == MKTO_BATCH['total']:
        await getMarketoData(MKTO_BATCH['data'])
        await initSalesforce("myparam1")
    ...

现在 initEvga 是一个协程，使用以下方法从 callback 启动它:

loop.create_task(initEvga(pubSubMsg))

...或者，如果回调是从不同的线程调用的:

asyncio.run_coroutine_threadsafe(initEvga(pubSubMsg), loop)

(您不需要将 loop 传递给 asyncio 函数，它们总是可以通过调用 asyncio.get_event_loop() 来获取正在运行的事件循环。)

最后，loop.call_soon(self.callback(message, loop))没有意义，因为call_soon接受一个函数来调用，上面的代码是使用 self.callback() 的结果 调用它，它会立即被调用。您需要:

self.callback(message, loop)

或类似的东西:

loop.call_soon(self.callback, message, loop)
# or loop.call_soon_threadsafe if called from a different thread

由于在这两种情况下 callback 都必须简短且非阻塞，推迟它似乎没有任何好处，所以我会选择第一个变体。