python - 访问两个协程之间的公共共享变量

Question

我正在使用 GCP PubSub 开发 Python3.6。

1. 我从 PubSub 收到订阅主题的回调。

2. 收到回调消息后，我将消息排队并ACK

3. 我在 init_apns_process() 中处理了收到的消息

未来的问题：

如果我立即确认回调（正在执行）并且processing(init_apns_process)发生异常时，我将丢失来自任一端的消息。

期待：

在完成之前不要 ACK PubSub init_apns_process()。

到目前为止我的解决方案：

apns_msg_queue = queue.Queue()
start_time = 0


async def init_apns_process(message):
    logging.info("Process Msg: {0}".format(message))
    await send_apns_notification(token)


async def send_apns_notification(device_token):
    global start_time
    apns = None
    apns = mi_apns_dev
    apns_payload = {"alert": "hello world 121", "serial": 21, "sound": "mov.mp3",
                    "mutable-content": "1", "category": "done", "id": -9999}

    if apns is not None:
        request = NotificationRequest(
            device_token=device_token,
            message={"aps": apns_payload}
        )

        ret = await apns.send_notification(request)
        if not ret.is_successful:
            logger.error(
                f"Failed to send APNS notification ID:")
        else:
            logging.info(f"Successfully sent APNS notification"
                         f" Time Taken in processing: {time.time() - start_time}")
             #update shared FLAG

def callback_value(message):
    apns_msg_queue.put(message)
    # return based on value of shared FLAG
    return True


if __name__ == '__main__':
    logger.setup()
    mi_apns_dev = APNs('apns.pem', use_sandbox=False)
    loop = asyncio.get_event_loop()

    subscriber = sub.PubSubSubscriber()
    subscriber.subscribe(subscription_name, callback_value, auto_ack_on_received=False)

    while True:
        if apns_msg_queue.qsize() > 0:
            msg = apns_msg_queue.get()
            if msg is None:
                break
            loop.run_until_complete(init_apns_process(msg))
        time.sleep(5)

可以在上述两个任务和基于其值的 ACK PubSub 之间有一些共享标志吗？

标签： pythonpython-3.xpython-asynciocoroutinegoogle-cloud-pubsub