Google uses AI technology to translate content into your preferred language. AI translations can contain errors.

仅传送一次

本页面介绍了如何使用 Pub/Sub 的“正好一次”功能接收和确认消息，该功能可让您跟踪和防止重复处理消息。启用该功能后，Pub/Sub 提供以下语义：

订阅者可以确定消息确认是否成功。
消息成功确认后，不会发生重新传送。
在消息未完成期间，不会发生重新传送。在确认截止时间到期或消息被确认之前，消息被视为未完成。
如果由于确认截止时间到期或客户端发起的否定确认而导致多次有效传送，则只能使用最新的确认 ID 来确认消息。使用之前的确认 ID 的任何请求都会失败。

启用“正好一次”后，订阅者可以按照以下准则确保消息被处理一次：

在确认截止时间之前确认消息。
维护有关消息处理进度的信息，直到消息成功确认为止。
使用有关消息处理进度的信息，以防止在确认失败时重复工作。

只有拉取订阅类型支持“正好一次”传送，包括使用 StreamingPull API的订阅者。推送和导出订阅不支持“正好一次”传送。

Pub/Sub 支持在一个云区域内基于 Pub/Sub 定义的唯一消息 ID进行“正好一次”传送。

重新传送与重复

务必了解预期重新传送与意外重新传送之间的区别。

重新传送可能是由于客户端发起了对消息的否定确认，也可能是由于客户端未在确认截止时间到期之前延长消息的确认截止时间。重新传送被认为是有效的，并且系统按预期运行。

如需排查重新传送问题，请参阅处理重复项。
重复是指在成功确认后或在确认截止时间到期之前重新发送消息。
重新传送的消息在重新传送尝试之间保留相同的消息 ID。

启用了“正好一次”传送的订阅不会收到重复传送。

客户端库中的“正好一次”传送支持

受支持的客户端库具有用于确认响应的接口（例如：Go）。您可以使用此接口检查确认请求是否成功。如果确认请求成功，则保证客户端不会收到重新传送。如果确认请求失败，则客户端可以预期会收到重新传送。
客户端也可以使用受支持的客户端库，而无需确认接口。但是，在这种情况下，确认失败可能会导致消息的静默重新传送。
受支持的客户端库具有用于设置最短租约延期时间的接口（例如： Go）。您必须将最短租约延期时间的值设置为较大的数字，以避免任何与网络相关的确认过期。最大值设置为 600 秒。
如果您使用的是 Java 客户端库，并且使用 setChannelProvider() 方法通过自定义 gRPC 通道初始化订阅者，建议您在构建 TransportChannelProvider 时，也将 maxInboundMetadataSize 设置为至少 1MB。对于此配置，您可以使用 InstantiatingGrpcChannelProvider.Builder.setMaxInboundMetadataSize() 或 ManagedChannelBuilder.maxInboundMetadataSize() 方法。

与“正好一次”传送相关的变量的默认值和范围以及变量的名称可能因客户端库而异。例如，在 Java 客户端库中，以下变量控制“正好一次”传送。

变量	说明	值
`setEnableExactlyOnceDelivery`	启用或停用“正好一次”传送。	true 或 false 默认值=false
`minDurationPerAckExtension`	用于延长修改确认截止时间的最短时间（以秒为单位）。	范围=0 到 600 默认值=无
`maxDurationPerAckExtension`	用于延长修改确认截止时间的最长时间（以秒为单位）。	范围=0 到 600 默认值=无

对于“正好一次”传送，当确认 ID 已过期时，对 Pub/Sub 的 modifyAckDeadline 或 acknowledgment 请求会失败。在这种情况下，服务会将过期的确认 ID 视为无效，因为较新的传送可能已在进行中。这是为“正好一次”传送而设计的。然后，您会看到 acknowledgment 和 ModifyAckDeadline 请求返回 INVALID_ARGUMENT 响应。停用“正好一次”传送后，如果确认 ID 过期，这些请求会返回 OK。

为确保 acknowledgment 和 ModifyAckDeadline 请求具有有效的确认 ID，请考虑将 minDurationPerAckExtension 的值设置为较大的数字。

区域注意事项

只有当订阅者在同一区域中连接到服务时，“正好一次”传送保证才适用。如果您的订阅者应用分布在多个区域，即使启用了“正好一次”传送，也可能会导致重复传送消息。发布者可以向任何区域发送消息，并且“正好一次”保证仍然有效。

当您在 Google Cloud中运行应用时，默认情况下，它会连接到同一区域中的 Pub/Sub 端点。因此，在 Google Cloud 中的单个区域运行应用通常可确保您与单个区域进行交互。

当您在或多个区域之外运行订阅者应用时，您可以在配置 Pub/Sub 客户端时使用位置端点，以保证连接到单个区域。 Google CloudPub/Sub 的所有位置端点都指向单个区域。如需详细了解位置端点，请参阅 Pub/Sub 端点。如需查看 Pub/Sub 的所有位置端点的列表，请参阅位置端点列表。

创建具有“正好一次”传送的订阅

您可以使用 Google Cloud 控制台、Google Cloud CLI、客户端库或 Pub/Sub API 创建具有“正好一次”传送的订阅。

拉取订阅

控制台

如需创建具有“正好一次”传送的拉取订阅，请按以下步骤操作：

在 Google Cloud 控制台中，进入订阅页面。

前往订阅页面
点击创建订阅 。
输入订阅 ID 。
从下拉菜单中选择或创建一个主题。

订阅将接收来自该主题的消息。
在正好一次传送 部分，选择启用“正好一次”传送 。
点击创建。

gcloud

如需创建具有“正好一次”传送的拉取订阅，请使用 gcloud pubsub subscriptions create 命令和 --enable-exactly-once-delivery 标志：

gcloud pubsub subscriptions create SUBSCRIPTION_ID \
  --topic=TOPIC_ID \
  --enable-exactly-once-delivery

替换以下内容：

SUBSCRIPTION_ID：要创建的订阅的 ID
TOPIC_ID：要附加到订阅的主题的 ID

REST

如需创建具有“正好一次”传送的订阅，请使用 projects.subscriptions.create 方法。

PUT https://pubsub.googleapis.com/v1/projects/PROJECT_ID/subscriptions/SUBSCRIPTION_ID
Authorization: Bearer $(gcloud auth print-access-token)

替换以下内容：

PROJECT_ID：要在其中创建订阅的项目的项目 ID
SUBSCRIPTION_ID：要创建的订阅的 ID

如需创建具有“正好一次”传送的拉取订阅，请在请求正文中指定此内容：

{
  "topic": "projects/PROJECT_ID/topics/TOPIC_ID",
  "enableExactlyOnceDelivery": true,
}

替换以下内容：

PROJECT_ID：包含主题的项目的项目 ID
TOPIC_ID：要附加到订阅的主题的 ID

C++

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 C++ 设置说明进行操作。如需了解详情，请参阅 Pub/Sub C++ API 参考文档。

namespace pubsub = ::google::cloud::pubsub;
namespace pubsub_admin = ::google::cloud::pubsub_admin;
[](pubsub_admin::SubscriptionAdminClient client,
   std::string const& project_id, std::string const& topic_id,
   std::string const& subscription_id) {
  google::pubsub::v1::Subscription request;
  request.set_name(
      pubsub::Subscription(project_id, subscription_id).FullName());
  request.set_topic(pubsub::Topic(project_id, topic_id).FullName());
  request.set_enable_exactly_once_delivery(true);
  auto sub = client.CreateSubscription(request);
  if (sub.status().code() == google::cloud::StatusCode::kAlreadyExists) {
    std::cout << "The subscription already exists\n";
    return;
  }
  if (!sub) throw std::move(sub).status();

  std::cout << "The subscription was successfully created: "
            << sub->DebugString() << "\n";
}

C#

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 C# 设置说明进行操作。如需了解详情，请参阅 Pub/Sub C# API 参考文档。


using Google.Cloud.PubSub.V1;
using Grpc.Core;

public class CreateSubscriptionWithExactlyOnceDeliverySample
{
    public Subscription CreateSubscriptionWithExactlyOnceDelivery(string projectId, string topicId, string subscriptionId)
    {
        SubscriberServiceApiClient subscriber = SubscriberServiceApiClient.Create();
        TopicName topicName = TopicName.FromProjectTopic(projectId, topicId);
        SubscriptionName subscriptionName = SubscriptionName.FromProjectSubscription(projectId, subscriptionId);

        var subscriptionRequest = new Subscription
        {
            SubscriptionName = subscriptionName,
            TopicAsTopicName = topicName,
            EnableExactlyOnceDelivery = true
        };

        Subscription subscription = null;

        try
        {
            subscription = subscriber.CreateSubscription(subscriptionRequest);
        }
        catch (RpcException e) when (e.Status.StatusCode == StatusCode.AlreadyExists)
        {
            // Already exists.  That's fine.
        }
        return subscription;
    }
}

Go

以下示例使用 Go Pub/Sub 客户端库的主要版本 (v2)。如果您仍在使用 v1 库，请参阅迁移到 v2 的指南。如需查看 v1 代码示例的列表，请参阅已废弃的代码示例。

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Go 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Go API 参考文档。

import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/pubsub/v2"
	"cloud.google.com/go/pubsub/v2/apiv1/pubsubpb"
)

func createSubscriptionWithExactlyOnceDelivery(w io.Writer, projectID, topic, subscription string) error {
	// projectID := "my-project-id"
	// topic := "projects/my-project-id/topics/my-topic"
	// subscription := "projects/my-project/subscriptions/my-sub"
	ctx := context.Background()
	client, err := pubsub.NewClient(ctx, projectID)
	if err != nil {
		return fmt.Errorf("pubsub.NewClient: %w", err)
	}
	defer client.Close()

	pbSub := &pubsubpb.Subscription{
		Name:                      subscription,
		Topic:                     topic,
		EnableExactlyOnceDelivery: true,
	}
	sub, err := client.SubscriptionAdminClient.CreateSubscription(ctx, pbSub)
	if err != nil {
		return fmt.Errorf("failed to create exactly once sub: %w", err)
	}
	fmt.Fprintf(w, "Created a subscription with exactly once delivery enabled: %v\n", sub)
	return nil
}

Java

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Java 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Java API 参考文档。

import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.ProjectTopicName;
import com.google.pubsub.v1.Subscription;
import java.io.IOException;

public class CreateSubscriptionWithExactlyOnceDelivery {
  public static void main(String... args) throws Exception {
    // TODO(developer): Replace these variables before running the sample.
    String projectId = "your-project-id";
    String topicId = "your-topic-id";
    String subscriptionId = "your-subscription-id";

    createSubscriptionWithExactlyOnceDeliveryExample(projectId, topicId, subscriptionId);
  }

  public static void createSubscriptionWithExactlyOnceDeliveryExample(
      String projectId, String topicId, String subscriptionId) throws IOException {
    try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create()) {

      ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
      ProjectSubscriptionName subscriptionName =
          ProjectSubscriptionName.of(projectId, subscriptionId);

      Subscription subscription =
          subscriptionAdminClient.createSubscription(
              Subscription.newBuilder()
                  .setName(subscriptionName.toString())
                  .setTopic(topicName.toString())
                  // Enable exactly once delivery in the subscription.
                  .setEnableExactlyOnceDelivery(true)
                  .build());

      System.out.println(
          "Created a subscription with exactly once delivery enabled: "
              + subscription.getAllFields());
    }
  }
}

Python

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Python 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Python API 参考文档。

from google.cloud import pubsub_v1

# TODO(developer): Choose an existing topic.
# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"

publisher = pubsub_v1.PublisherClient()
subscriber = pubsub_v1.SubscriberClient()
topic_path = publisher.topic_path(project_id, topic_id)
subscription_path = subscriber.subscription_path(project_id, subscription_id)

with subscriber:
    subscription = subscriber.create_subscription(
        request={
            "name": subscription_path,
            "topic": topic_path,
            "enable_exactly_once_delivery": True,
        }
    )
    print(
        f"Created subscription with exactly once delivery enabled: {subscription}"
    )

Node.js

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Node.js 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Node.js API 参考文档。

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// Imports the Google Cloud client library
const {PubSub} = require('@google-cloud/pubsub');

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createSubscriptionWithExactlyOnceDelivery(
  topicNameOrId,
  subscriptionNameOrId,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableExactlyOnceDelivery: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with exactly-once delivery.`,
  );
  console.log(
    'To process messages, remember to check the return value of ackWithResponse().',
  );
}

Node.js

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Node.js 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Node.js API 参考文档。

/**
 * TODO(developer): Uncomment these variables before running the sample.
 */
// const topicNameOrId = 'YOUR_TOPIC_NAME_OR_ID';
// const subscriptionNameOrId = 'YOUR_SUBSCRIPTION_NAME_OR_ID';

// Imports the Google Cloud client library
import {PubSub} from '@google-cloud/pubsub';

// Creates a client; cache this for further use
const pubSubClient = new PubSub();

async function createSubscriptionWithExactlyOnceDelivery(
  topicNameOrId: string,
  subscriptionNameOrId: string,
) {
  // Creates a new subscription
  await pubSubClient
    .topic(topicNameOrId)
    .createSubscription(subscriptionNameOrId, {
      enableExactlyOnceDelivery: true,
    });
  console.log(
    `Created subscription ${subscriptionNameOrId} with exactly-once delivery.`,
  );
  console.log(
    'To process messages, remember to check the return value of ackWithResponse().',
  );
}

Ruby

以下示例使用 Ruby Pub/Sub 客户端库 v3。如果您仍在使用 v2 库，请参阅迁移到 v3 的指南。如需查看 Ruby v2 代码示例的列表，请参阅已废弃的代码示例。

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 Ruby 设置说明进行操作。如需了解详情，请参阅 Pub/Sub Ruby API 参考文档。

# project_id = "your-project-id"
# topic_id = "your-topic-id"
# subscription_id = "your-subscription-id"

pubsub = Google::Cloud::PubSub.new project_id: project_id
subscription_admin = pubsub.subscription_admin

subscription = subscription_admin.create_subscription \
  name: pubsub.subscription_path(subscription_id),
  topic: pubsub.topic_path(topic_id),
  enable_exactly_once_delivery: true

puts "Created subscription with exactly once delivery enabled: " \
     "#{subscription_id}"

PHP

在尝试此示例之前，请按照《快速入门：使用客户端库》中的 PHP 设置说明进行操作。如需了解详情，请参阅 Pub/Sub PHP API 参考文档。

use Google\Cloud\PubSub\PubSubClient;

/**
 * Creates a Pub/Sub subscription with `Exactly Once Delivery` enabled.
 *
 * @param string $projectId  The Google project ID.
 * @param string $topicName  The Pub/Sub topic name.
 * @param string $subscriptionName  The Pub/Sub subscription name.
 */
function create_subscription_with_exactly_once_delivery(
    string $projectId,
    string $topicName,
    string $subscriptionName
): void {
    $pubsub = new PubSubClient([
        'projectId' => $projectId,
    ]);
    $topic = $pubsub->topic($topicName);
    $subscription = $topic->subscription($subscriptionName);
    $subscription->create([
        'enableExactlyOnceDelivery' => true
    ]);

    // Exactly Once Delivery status for the subscription
    $status = $subscription->info()['enableExactlyOnceDelivery'];

    printf('Subscription created with exactly once delivery status: %s' . PHP_EOL, $status ? 'true' : 'false');
}

监控“正好一次”传送订阅

The subscription/exactly_once_warning_count 指标记录可能导致重新传送（有效或重复）的事件数。此指标统计 Pub/Sub 无法处理与确认 ID（ModifyAckDeadline 或 acknowledgment 请求）关联的请求的次数。失败的原因可能是基于服务器或客户端的。例如，如果用于维护“正好一次”传送信息的持久层不可用，则属于基于服务器的事件。如果客户端尝试使用无效的确认 ID 确认消息，则属于基于客户端的事件。

了解指标

subscription/exactly_once_warning_count 捕获可能会或可能不会导致实际重新传送的事件，并且可能会根据客户端行为而产生干扰。例如：使用无效确认 ID 重复 acknowledgment 或 ModifyAckDeadline 请求会重复增加指标。

以下指标也有助于了解客户端行为：

subscription/expired_ack_deadlines_count 指标显示确认 ID 过期的次数。确认 ID 过期可能会导致 ModifyAckDeadline 和 acknowledgment 请求失败。
service.serviceruntime.googleapis.com/api/request_count 指标可用于捕获请求到达但未到达 Pub/Sub 的情况下 ModifyAckDeadline 或 acknowledgment 请求的失败。 Google Cloud 此指标不会捕获某些失败，例如，当客户端与断开连接时。 Google Cloud

在大多数可以重试的失败事件中，受支持的客户端库会自动重试请求。

配额

“正好一次”传送订阅需要满足额外的配额要求。这些配额适用于：

每个区域中从启用了“正好一次”传送的订阅中使用的消息数。
每个区域中使用启用了“正好一次”传送的订阅时确认或截止时间延长的消息数。

如需详细了解这些配额，请参阅配额主题中的表格。

“正好一次”传送和有序订阅

Pub/Sub 支持“正好一次”传送和有序传送。

将排序与“正好一次”传送结合使用时，Pub/Sub 希望确认按顺序进行。如果确认顺序不正确，服务会因临时错误而导致请求失败。如果确认截止时间在传送的有序确认之前到期，客户端将收到重新传送的消息。因此，当您将排序与“正好一次”传送结合使用时，客户端吞吐量将限制为每秒数千条消息。

“正好一次”传送和推送订阅

Pub/Sub 仅支持“正好一次”传送和拉取订阅。

使用推送订阅的消息的客户端通过使用成功响应来响应推送请求，从而确认消息。但是，客户端不知道 Pub/Sub 订阅是否收到了响应并对其进行了处理。这与拉取订阅不同，在拉取订阅中，确认请求由客户端发起，如果请求已成功处理，Pub/Sub 订阅会做出响应。因此，“正好一次”传送语义与推送订阅不太一致。

注意事项

如果在 CreateSubscription 时未指定确认截止时间，则启用了“正好一次”传送的订阅的默认确认截止时间为 60 秒。
较长的默认确认截止时间有助于避免因网络事件而导致的重新传送。受支持的客户端库不使用默认订阅确认截止时间。
与常规订阅相比，“正好一次”传送订阅的发布到订阅的延迟时间要长得多。
如果您需要高吞吐量，则“正好一次”传送客户端还必须使用流式拉取。
即使启用了“正好一次”传送，订阅也可能会收到同一消息的多个副本，这是因为发布端存在重复项。发布端重复项可能是由于发布客户端或 Pub/Sub 服务多次进行唯一发布重试。发布客户端在重试期间多次进行唯一发布会导致重新传送，且消息 ID 不同。Pub/Sub 服务多次进行唯一发布以响应客户端发布请求会导致重新传送，且消息 ID 相同。
您可以重试 subscription/exactly_once_warning_count 中的失败，并且受支持的客户端库会自动重试这些失败。但是，与无效确认 ID 相关的失败无法重试。

仅传送一次 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

推荐的客户端库版本

重新传送与重复

客户端库中的“正好一次”传送支持

区域注意事项

创建具有“正好一次”传送的订阅

拉取订阅

控制台

gcloud

REST

C++

C#

Go

Java

Python

Node.js

Node.js

Ruby

PHP

监控“正好一次”传送订阅

了解指标

配额

“正好一次”传送和有序订阅

“正好一次”传送和推送订阅

注意事项

仅传送一次