Pub\Sub Python 客户端 - 优雅地关闭订阅者

Posted 2023-02-16

【中文标题】Pub\\Sub Python 客户端 - 优雅地关闭订阅者

【英文标题】：Pub\Sub Python Client - Gracefully shutdown subscriberPub\Sub Python 客户端 - 优雅地关闭订阅者

【发布时间】：2021-04-21 23:13:23

【问题描述】：

我在 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

我希望这段代码停止拉取消息并完成正在运行的消息然后退出。

实际上，这段代码停止拉取消息并完成执行正在运行的消息，但它不确认消息。 .ack() 发生但服务器没有收到 ack，所以下次运行相同的消息再次返回。

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

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

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

【问题讨论】：

我看了一下库，停止进程（取消）等待所有线程结束。我想到了别的东西：您的订阅确认截止日期是什么时候？

@guillaumeblaquiere 我的确认截止日期是默认的 600 秒

@JohnHanley 即使睡了 60 秒，确认仍然没有发生。

SIGTERM 发生在一个更复杂的代码中，所以我做了一个没有它的简单示例。

在我的实际应用程序中，我使用 sigterm 处理程序来调用 .cancel()。在这里，使用没有 sigterm（处理或调用）的更简单的代码，我观察到取消后未确认的消息的相同行为。在问题中写 sigterm 令人困惑，我将其删除。

【参考方案1】：

更新 (v2.4.0+)

客户端版本2.4.0在流拉未来的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() 返回后，回调可能仍在后台运行，但它们生成的任何 ACK 都将无效，因为不会再运行任何线程来将 ACK 请求分派到后端。 位于客户端内部队列中的消息现在在关闭时会自动进行 NACK。无论await_msg_callbacks 值如何，都会发生这种情况。

原始答案（v2.3.0 及以下）

流式拉取由流式拉取管理器在后台管理。当流式拉取未来为canceled 时，它会调用管理器的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 不会发送到服务器。

杂项。备注

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

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 请求是尽力而为。即使关闭被阻塞并等待用户回调完成，仍然无法保证消息会真正得到确认（例如，请求可能会在网络中丢失）。

Re：关于重新传递消息的担忧（“所以下次运行相同的消息会再次返回”）——这实际上是设计使然。后端将努力传递每条消息at least once，因为请求可能会丢失。这包括来自订阅者的 ACK 请求，因此订阅者应用程序的设计必须考虑幂等性。

【讨论】：

如果我确实想在关闭连接之前等待挂起的任务完成，我可以这样做吗？似乎 wait=False/True 标志在这里是合适的。因为在我的场景中，我希望停止接收消息并处理已经收到的消息。 @plamut

也许通过覆盖传入的调度程序，但开箱即用的答案可能是否定的，恐怕。另一方面，过去也有类似的feature requests，如果有更多需求，它实际上可能会被添加到路线图中。

我可以通过等待开始处理消息的线程在我的代码中解决这个问题。如果已开始关闭，则不会启动新线程。在我确定没有正在运行的线程执行消息之后，我将所有尚未启动的消息都取消并调用 .cancel()。如果这将在库级别实现，那肯定很好。

我同意，Montoya，修复此行为会很有用。就目前而言，如果订阅者在 ack 消息有机会发送出去之前关闭，那么即使是 ack() 也可能在它到达服务器之前被丢弃。我们也可以为用户提供 ack() 的未来，但它可能太复杂了。通常，该库针对持续吞吐量进行了优化，并且在处理一些消息和关闭方面表现不佳。对于数以百万计的消息，一些丢失的 ack 没什么大不了的，但在较小的数量下它是一个更大的问题。

@Montoya 昨天发布的最新版本v2.4.0 正好添加了这一点 - 自动 NACK 和可选的阻塞，直到回调完成执行。有关详细信息，请参阅更新的答案。