在谷歌云函数中使用 PubSub

Question

以官方文档为准

我尝试使用“PubSub Pull Subscription”触发器创建云函数

import base64

def hello_pubsub(event, context):

    """Triggered from a message on a Cloud Pub/Sub topic.

    Args:

         event (dict): Event payload.

         context (google.cloud.functions.Context): Metadata for the event.

    """

    print("This Function was triggered by messageId {} published at {}".format(context.event_id, context.timestamp))

    if 'data' in event:

        name = base64.b64decode(event['data']).decode('utf-8')

    print('"{}" received!'.format(name))

    

    if 'attributes' in event:

        print(event['attributes'])

    if '@type' in event:

        print(event['@type'])  

然后找到一篇文章说“cloud function will send ACK on its invocation”，和官方文档是一致的。

但是，当云函数完成对 PubSub 消息的处理后，“Unacked message count”并没有减少（如上图所示）

因此，我在本地尝试google-cloud-pubsub

subscription_path = subscriber.subscription_path(PROJECT, SUBSCRIPTION)

response = subscriber.pull(subscription_path, max_messages=5)

for msg in response.received_messages:

    print("Received message:", msg.message.data)

ack_ids = [msg.ack_id for msg in response.received_messages]

subscriber.acknowledge(subscription_path, ack_ids)

这样，消息计数成功减少。

我的问题是：

• 我的云函数脚本中是否缺少某些内容？

• 我怎样才能在我的云函数中实际“使用”PubSub 消息？

任何建议表示赞赏，谢谢。

Answer 1

使用 PubSub，您有发布者，将消息发布到主题中。该消息在每个现有订阅（在主题上创建）中重复。最后，订阅者可以收听订阅。因此，在这里，您有 1 个主题和 1 个请求订阅。您还有一个部署在主题上的 Cloud Functions （在 gcloud cli 中，param --trigger-topic=myTopic）。关于主题，而不是订阅。返回订阅页面，您应该看到 2 个订阅：您的请求订阅和对一个陌生端点的推送订阅因此，您的消息发布在 2 个订阅中。如果您查看您的 Pull 订阅，除了您在本地的代码外，没有任何内容会消耗其中的消息。云函数中的日志应该显示正确的消息处理。是不是更清楚了？编辑准确地说是你的情况：您的 Cloud Functions 无法确认请求订阅中的消息，因为它没有连接到它您的 Cloud Functions 处理并确认在其自己的订阅中发布的消息。