Pub\Sub Python 客户端 - 正常关闭订阅者

Question

我在 python3.6 中使用 Google Pub/Sub 客户端 v2.2.0 作为订阅者。

我希望我的应用程序在确认已收到的所有消息后正常关闭。

来自 Google 指南的订阅者示例代码，稍作改动即可显示我的问题：

from concurrent.futures import TimeoutError
from google.cloud import pubsub_v1
from time import sleep

# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"
# Number of seconds the subscriber should listen for messages
# timeout = 5.0

subscriber = pubsub_v1.SubscriberClient()
# The `subscription_path` method creates a fully qualified identifier
# in the form `projects/{project_id}/subscriptions/{subscription_id}`
subscription_path = subscriber.subscription_path(project_id, subscription_id)

def callback(message):
    print(f"Received {message}.")
    sleep(30)
    message.ack()
    print("Acked")

streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
print(f"Listening for messages on {subscription_path}..\n")

# Wrap subscriber in a 'with' block to automatically call close() when done.
with subscriber:
    sleep(10)
    streaming_pull_future.cancel()
    streaming_pull_future.result()

来自https://cloud.google.com/pubsub/docs/pull

我希望此代码停止提取消息并完成运行ning 消息然后退出。

实际上此代码会停止提取消息并完成执行运行ning 消息，但不会确认消息。 .ack() 发生但服务器未收到确认，因此下一个运行相同的消息再次 return。

1.为什么服务器收不到ack？

2。如何优雅地关闭订阅者？

3。 .cancel() 的预期行为是什么？

Answer 1

更新 (v2.4.0+)

客户端版本 2.4.0 向流式拉取 future 的 cancel() 方法添加了一个新的可选参数 await_msg_callbacks。如果设置为True，该方法将阻塞，直到所有当前正在执行的消息回调完成并且后台消息流已关闭（默认为False）。

try:
    streaming_pull_future.result()
except KeyboardInterrupt:
    streaming_pull_future.cancel(await_msg_callbacks=True)  # blocks until done

一些发行说明：

等待回调意味着其中生成的任何消息 ACK 仍将被处理（读取：发送到后端）。
如果 await_msg_callbacks 为 False 或未给出，将不等待而继续关闭。 cancel() return 秒后，回调可能仍在后台运行运行ning，但它们生成的任何 ACK 都将无效，因为不会有任何线程运行ning 将 ACK 请求发送到后端。
位于客户端内部队列中的消息现在在关闭时自动 NACK-ed。无论 await_msg_callbacks 值如何，都会发生这种情况。

原回答（v2.3.0及以下版本）

流式拉取由流式拉取管理器在后台管理。当流式拉取 future 是 canceled, it invokes the manager's close() 优雅关闭后台辅助线程的方法时。

其中一个被关闭的东西是 调度程序 - 它是一个线程池，用于将接收到的消息异步分派给用户回调。需要注意的关键是 scheduler.shutdown() 不会 而不是 等待用户回调完成，因为它可能会“永远”阻塞，而是清空执行程序的工作队列并关闭后者向下：

def shutdown(self):
    """Shuts down the scheduler and immediately end all pending callbacks.
    """
    # Drop all pending item from the executor. Without this, the executor
    # will block until all pending items are complete, which is
    # undesirable.
    try:
        while True:
            self._executor._work_queue.get(block=False)
    except queue.Empty:
        pass
    self._executor.shutdown()

这解释了为什么在提供的代码示例中未发送 ACK - 回调休眠 30 秒，而流式拉取未来仅在大约 10 秒后取消。 ACK 未发送到服务器。

杂项。备注

由于流式拉取是一个很长的运行ning 操作，我们希望在主线程中阻塞，以免过早退出。这是通过阻止流式拉取未来结果来完成的：

try:
    streaming_pull_future.result()
except KeyboardInterrupt:
    streaming_pull_future.cancel()

或在预设超时后：

try:
    streaming_pull_future.result(timeout=123)
except concurrent.futures.TimeoutError:
    streaming_pull_future.cancel()

ACK 请求是尽力而为的。即使关闭被阻止并等待用户回调完成，仍然无法保证消息实际上会得到确认（例如，请求可能会在网络中丢失）。
回复：关于重新发送消息的问题（“所以接下来运行同样的消息 return 再次发送”）-这实际上是设计使然。后端将努力传递每条消息 at least once，因为请求可能会丢失。这包括来自订阅者的 ACK 请求，因此订阅者应用程序必须在设计时考虑到幂等性。

Pub\Sub Python 客户端 - 正常关闭订阅者

Pub\Sub Python Client - Gracefully shutdown subscriber

python

google-cloud-platform

publish-subscribe

google-cloud-pubsub

pypubsub

更新 (v2.4.0+)

原回答（v2.3.0及以下版本）

杂项。备注