如何使用GCP Pub/Sub从订阅中提取消息？

Question

我对GCP Pub/Sub REST API感到困惑。

背景：我正在尝试使用GCP发布/订阅编写一个应用程序，其中该语言不会作为客户端库退出（我正在尝试使用R）。

因此，我需要依赖提供的REST API：https://cloud.google.com/pubsub/docs/reference/rest

根据我对REST API的理解，我们必须使用pull订阅：https://cloud.google.com/pubsub/docs/reference/rest/v1/projects.subscriptions/pull

问题：我的困惑是这是一个帖子请求

POST https://pubsub.googleapis.com/v1/{subscription}:pull

此POST请求有响应：

{
  "receivedMessages": [
    {
      object (ReceivedMessage)
    }
  ]
}

我如何收到POST请求的回复？这对我来说没有意义。

我的目标是订阅消息的发布/订阅，类似于此处的Python库：

要订阅Cloud Pub/Sub中的html" target="_blank">数据，您可以根据主题创建订阅，并通过传递回调函数订阅该订阅。

import os
from google.cloud import pubsub_v1

topic_name = 'projects/{project_id}/topics/{topic}'.format(
    project_id=os.getenv('GOOGLE_CLOUD_PROJECT'),
    topic='MY_TOPIC_NAME',  # Set this to something appropriate.
)

subscription_name = 'projects/{project_id}/subscriptions/{sub}'.format(
    project_id=os.getenv('GOOGLE_CLOUD_PROJECT'),
    sub='MY_SUBSCRIPTION_NAME',  # Set this to something appropriate.
)

def callback(message):
    print(message.data)
    message.ack()

with pubsub_v1.SubscriberClient() as subscriber:
    subscriber.create_subscription(
        name=subscription_name, topic=topic_name)
    future = subscriber.subscribe(subscription_name, callback)
    try:
        future.result()
    except KeyboardInterrupt:
        future.cancel()

Answer 1

云发布/订阅客户端库使用流式pull来接收消息，而不是pull。它还将用户从这些实际请求中抽象出来。客户端库本身接收消息列表，然后调用用户提供的回调。

如果您直接使用pull请求，那么您希望迭代响应中的每条消息，并对每条消息调用用户回调。在所有受支持的语言（如Python）的云发布/订阅文档中有一个例子：

from google.api_core import retry
from google.cloud import pubsub_v1

# TODO(developer)
# project_id = "your-project-id"
# subscription_id = "your-subscription-id"

subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)

NUM_MESSAGES = 3

# Wrap the subscriber in a 'with' block to automatically call close() to
# close the underlying gRPC channel when done.
with subscriber:
    # The subscriber pulls a specific number of messages. The actual
    # number of messages pulled may be smaller than max_messages.
    response = subscriber.pull(
        request={"subscription": subscription_path, "max_messages": NUM_MESSAGES},
        retry=retry.Retry(deadline=300),
    )

    ack_ids = []
    for received_message in response.received_messages:
        print(f"Received: {received_message.message.data}.")
        ack_ids.append(received_message.ack_id)

    # Acknowledges the received messages so they will not be sent again.
    subscriber.acknowledge(
        request={"subscription": subscription_path, "ack_ids": ack_ids}
    )

    print(
        f"Received and acknowledged {len(response.received_messages)} messages from {subscription_path}."
    )