发布对冲

发布消息时出现高尾延迟可能是由暂时性网络延迟或临时服务器端速度缓慢引起的。借助发布对冲，如果正在进行的请求未在配置的延迟时间内完成，Pub/Sub 发布者客户端可以并行发送备份（对冲）发布请求，并使用最先返回的响应。

本文档介绍了如何对发布到主题的消息使用发布对冲。

准备工作

在配置发布工作流之前，请确保您已完成以下任务：

所需的角色

如需获得向主题发送对冲发布请求所需的权限，请让您的管理员为您授予主题的 Pub/Sub Publisher (roles/pubsub.publisher) IAM 角色。 如需详细了解如何授予角色，请参阅管理对项目、文件夹和组织的访问权限。

您也可以通过自定义角色或其他预定义角色来获取所需的权限。

您需要额外权限才能创建或更新主题和订阅。

关于发布对冲

对冲设置可控制 Pub/Sub 客户端库发送对冲发布请求的方式和时间。发布对冲默认处于停用状态。 启用后，客户端库具有以下对冲设置：

  • 对冲延迟时间：如果上一个请求尚未完成，客户端库在并行发送其他发布请求之前等待的时间。默认值为 1 秒，配置的值必须介于 100 毫秒到 10 秒之间。
  • 令牌数上限：客户端库可以累积用于发送对冲发布请求的令牌数上限。每个对冲发布请求都会消耗 1 个令牌。默认值为 50，配置的值必须介于 1 到 250 之间。
  • 补充比率：每次成功发布请求后添加的令牌数量。此值控制一段时间内可对冲的发布请求的最大比例。例如，如果补充比率为 0.1，则最多可以对 10% 的发布请求进行对冲。默认值为 0.1，配置的值必须介于 0.001 和 0.2 之间。

如需对冲发布请求，对冲延迟时间必须短于发布商重试设置中的初始请求超时和总超时。否则，发布者客户端将无法初始化。启用发布对冲后，客户端库会按如下方式发送发布请求：

  • 如果传输中的发布请求未在对冲延迟时间内完成，并且至少有一个令牌可用，则客户端库会消耗一个令牌，并为同一批消息并行发送一个额外的发布请求，同时保持初始请求未完成。
  • 如果所有正在处理的请求在另一个对冲延迟时间后仍未完成，则客户端库会继续以对冲延迟时间为间隔发送其他对冲请求，直到令牌用尽或总超时时间到期。
  • 一旦该批次的任何正在进行的发布请求成功，客户端库就会完成发布操作，取消该批次的任何剩余正在进行的尝试，并按补充比率增加可用令牌，最多可达到最大令牌数。

对冲发布请求

与重试请求（等待正在进行的发布请求失败或超时）不同，发布对冲会在发送并发对冲请求的同时，保持初始请求处于未完成状态。这样可以减少尾延迟时间，而不会过早取消即将成功的请求。

由于针对同一批消息的多个请求可以同时进行，因此在取消其余尝试之前，可能会有多次尝试成功，因为发布请求取消是尽力而为。因此，您可能会收到具有不同消息 ID 的重复消息，这可能会增加客户端网络流量以及发布和传送这些重复消息的 Pub/Sub 费用。如需了解有关处理重复消息的提示，请参阅“排查重复、重新递送或缺失的消息”。

如果服务降级导致延迟时间或错误持续增加，则最大令牌数和补充比率设置可防止对冲发布速率过高。由于仅在发布请求成功时才会添加令牌，因此持续的缓慢或错误会快速耗尽可用令牌并自动限制对冲请求。

此示例代码演示了如何创建具有自定义对冲设置的发布商。 并非所有客户端库都支持发布对冲；请参阅您所选语言的 API 参考文档。

Java

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


import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.HedgingSettings;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import com.google.pubsub.v1.TopicName;
import java.io.IOException;
import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class PublishWithHedgingSettingsExample {
  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";

    publishWithHedgingSettingsExample(projectId, topicId);
  }

  public static void publishWithHedgingSettingsExample(String projectId, String topicId)
      throws IOException, ExecutionException, InterruptedException {
    TopicName topicName = TopicName.of(projectId, topicId);
    Publisher publisher = null;

    try {
      // Hedging settings configures how and when the publisher sends multiple concurrent publish
      // requests to reduce tail latency.
      Duration hedgeDelay = Duration.ofMillis(500); // default: 1000 ms (1 second)
      int maxTokens = 100; // default: 50
      float refillRatio = 0.05f; // default: 0.1

      HedgingSettings hedgingSettings =
          HedgingSettings.newBuilder()
              .setHedgeDelay(hedgeDelay)
              .setMaxTokens(maxTokens)
              .setRefillRatio(refillRatio)
              .build();

      // Create a publisher instance with hedging settings bound to the topic
      publisher = Publisher.newBuilder(topicName).setHedgingSettings(hedgingSettings).build();

      String message = "Hello world!";
      ByteString data = ByteString.copyFromUtf8(message);
      PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

      // Once published, returns a server-assigned message id (unique within the topic)
      ApiFuture<String> messageIdFuture = publisher.publish(pubsubMessage);
      String messageId = messageIdFuture.get();
      System.out.println("Published a message with hedging settings: " + messageId);

    } finally {
      if (publisher != null) {
        // When finished with the publisher, shutdown to free up resources.
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }
}

使用排序键发布对冲

发布对冲不能与消息排序一起使用。由于发布对冲会针对同一批消息发送并发发布请求，因此 Pub/Sub 服务可能会不按顺序接收和处理这些请求。如果您在发布者客户端上同时启用消息排序和发布对冲，则创建发布者会失败并报错。

技巧和最佳实践

配置发布对冲时，请考虑以下最佳实践：

  • 使用对冲来应对瞬时尾延迟：发布对冲旨在消除因临时网络延迟或服务器端短暂缓慢而导致的短期延迟峰值。不要依赖对冲来缓解持续中断或持续错误。
  • 使用重试政策进行对冲：在重试设置中设置较长的初始请求超时，同时设置较短的对冲延迟。如果超时时间过短，系统会过早取消请求并触发完整重试，从而丢弃服务器上已在进行的工作。如果原始请求只是略有延迟，则较长的超时时间可让原始请求完成。
  • 客户端流控制和对冲：发布对冲无法解决客户端延迟问题，例如 CPU 争用或线程饥饿。如果发布延迟时间过长是由客户端内存中的消息队列引起的，则使用流控制来管理队列大小可能会有所帮助，但宿主资源限制也应得到解决。
  • 监控发布延迟时间和吞吐量指标：使用 Monitoring 指标图表（例如 topic/send_request_latencies）来识别尾部延迟时间峰值，并确定对冲是否有帮助。CANCELLED 错误率较高可能表示您的请求截止期限配置得过低。如果调整得当，发布对冲可使消息吞吐量和主题字节费用大致保持不变，即使在发布尾延迟期间也是如此。

后续步骤

如需了解如何配置高级发布选项，请参阅以下内容：