问题描述
我在 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?
3. .cancel() 的预期行为是什么?
解决方法
更新 (v2.4.0+)
客户端版本 2.4.0 向流式拉取未来的 await_msg_callbacks
方法添加了一个新的可选参数 cancel()
。如果设置为 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 及以下)
流拉由流拉管理器在后台管理。当流式拉取 Future 为 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 请求,因此在设计订阅者应用程序时必须考虑到幂等性。