Google Cloud PubSub 消息未通过回调处理



我正在尝试使用Google PubSub在两个服务之间传递和接收消息。但是,发送的某些消息似乎是随机丢弃的,并且不会由订阅者的回调方法处理。

发送消息时,大约一半的消息由回调方法处理。对于另一半,似乎根本没有调用回调方法(不记录任何信息(。但是,消息仍然从主题中消失,并且不会重新发送。

用于启动订阅服务器的代码:

logger = logging.getLogger(LOGGER_NAME)
logger.info('Starting the pubsub subscriber')
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(GOOGLE_CLOUD_PROJECT, SUBSCRIPTION_NAME)
subscriber.subscribe(subscription_path, callback=callback)
while True:
try:
sleep(60)
except Exception as e:
// Log exception

回调方法:

def callback(message):
logger = logging.getLogger(LOGGER_NAME)
logger.info(f'Recieved callback with message: {message}', extra = {'callback_message': message}  )
// Process message

错误似乎出在订户端。消息是从发布者发送的,如果订阅者未连接到主题,则消息不会消失。

我尝试使用流控制来控制订阅者检索的消息数,但它似乎没有任何效果。

是否可以在不调用回调方法的情况下处理消息?消息可能从主题中消失还有其他原因吗?

编辑:原来另一个服务正在从同一订阅读取,处理丢失的消息。

我知道你找到了问题的答案,但我认为列出一些调试此类问题的有用步骤是值得的:

  1. 检查以确保消息确实已发布。当发布成功时,响应应包含消息的 ID,例如,作为 Java Publish 方法中的 APIFuture 生成的字符串。
  2. 检查邮件是否积压。您可以通过堆栈驱动程序查看subscription/oldest_unacked_message_agesubscription/num_undelivered_messages
  3. 检查您的订阅者是否设置了流控制,阻止您及时接收所有消息。如果您设置了流控制,并且它阻止了所有消息的传递,您可能会看到 Stackdriver 中未传递的消息数量正在增加。
  4. 确保没有任何其他客户端订阅同一订阅的消息。例如,也许您正在使用gcloud工具来提取和查看消息。在这种情况下,您可能不会看到 Stackdriver 中未传递的消息数增加。

如果在检查完所有这些内容后,您不确定消息发生了什么,最好联系支持人员,并提供项目和订阅的名称以及您认为未传递的任何消息的 ID。

最新更新