保证 Google Cloud Pub/Sub 消息顺序性

在某些业务场景下，消息的顺序至关重要。例如，银行交易必须按照发生的顺序处理，否则可能会导致账户余额错误。Google Cloud Pub/Sub 提供了一些机制来保证消息的顺序性，以满足这些特定需求。

1. Keyed Ordering 🔑

Keyed Ordering 是 Pub/Sub 保证顺序性的主要机制。它允许你为消息指定一个排序键 (ordering key)。具有相同排序键的所有消息将按照它们发布的顺序传递给订阅者。

工作原理:

• 发布者: 在发布消息时，设置 ordering_key 属性。这个键可以是任何字符串，用于标识消息所属的顺序组。
• Pub/Sub: Pub/Sub 保证具有相同 ordering_key 的消息按照发布的顺序存储和传递。
• 订阅者: 订阅者接收到的消息将按照 ordering_key 进行分组，并且每个组内的消息顺序与发布顺序一致。

示例:

假设你正在处理用户活动日志，并且需要确保特定用户的活动按照时间顺序处理。你可以使用用户 ID 作为 ordering_key。

  // 发布消息
  PubPublisher publisher = topic.newPublisher(settings);

  ByteString data = ByteString.copyFromUtf8("用户 A 的活动 1");
  PubsubMessage message = PubsubMessage.newBuilder()
    .setData(data)
    .setOrderingKey("user-A")
    .build();

  publisher.publish(message);

  ByteString data2 = ByteString.copyFromUtf8("用户 A 的活动 2");
  PubsubMessage message2 = PubsubMessage.newBuilder()
    .setData(data2)
    .setOrderingKey("user-A")
    .build();

  publisher.publish(message2);
  

在这个例子中，所有 ordering_key 为 "user-A" 的消息将按照发布的顺序传递给订阅者。

2. 注意事项 🤔

虽然 Keyed Ordering 提供了强大的顺序性保证，但也需要注意以下几点：

• 单个 Key 的吞吐量限制: 具有相同 ordering_key 的消息的处理吞吐量受到限制。如果你的应用需要处理大量具有相同 ordering_key 的消息，可能会遇到性能瓶颈。
• 错误处理: 如果订阅者在处理某个 ordering_key 的消息时发生错误，可能会导致后续消息的延迟。你需要设计适当的错误处理机制，例如重试或死信队列。
• 分区: Pub/Sub 会根据 ordering_key 将消息分配到不同的分区。因此，具有不同 ordering_key 的消息可以并行处理。

3. 替代方案 💡

除了 Keyed Ordering 之外，还有一些其他的替代方案可以考虑：

• 使用时间戳: 在消息中包含时间戳，并在订阅者端根据时间戳对消息进行排序。这种方法不依赖于 Pub/Sub 的顺序性保证，但需要你自行处理排序逻辑。
• 使用外部排序服务: 使用外部的排序服务（例如 Apache Kafka 或 Apache ZooKeeper）来保证消息的顺序。这种方法需要额外的基础设施，但可以提供更高的灵活性和可扩展性。

4. 选择合适的方案 ✅

选择哪种方案取决于你的具体业务需求。如果你的应用需要严格的顺序性保证，并且可以接受单个 ordering_key 的吞吐量限制，那么 Keyed Ordering 是一个不错的选择。如果你的应用对顺序性要求不高，或者需要更高的吞吐量，那么可以考虑使用其他的替代方案。

5. 代码示例 (Java) 💻

  import com.google.cloud.pubsub.v1.Publisher;
  import com.google.protobuf.ByteString;
  import com.google.pubsub.v1.TopicName;
  import com.google.pubsub.v1.PubsubMessage;
  import java.io.IOException;
  import com.google.api.gax.core.CredentialsProvider;
  import com.google.api.gax.grpc.GrpcTransportChannel;
  import com.google.api.gax.rpc.FixedTransportChannelProvider;
  import io.grpc.ManagedChannel;
  import io.grpc.ManagedChannelBuilder;
  import com.google.cloud.pubsub.v1.TopicAdminClient;
  import com.google.cloud.pubsub.v1.TopicAdminSettings;

  public class PubSubOrderingExample {

      public static void main(String[] args) throws IOException, InterruptedException {

          String projectId = "your-project-id"; // 替换为你的项目 ID
          String topicId = "your-topic-id";     // 替换为你的 Topic ID

          TopicName topicName = TopicName.of(projectId, topicId);

          // 创建 Publisher
          Publisher publisher = Publisher.newBuilder(topicName).build();

          try {
              // 发布消息 1
              ByteString data1 = ByteString.copyFromUtf8("Message 1 for user-A");
              PubsubMessage message1 = PubsubMessage.newBuilder()
                      .setData(data1)
                      .setOrderingKey("user-A")
                      .build();
              publisher.publish(message1);
              System.out.println("Published: " + message1);

              // 发布消息 2
              ByteString data2 = ByteString.copyFromUtf8("Message 2 for user-A");
              PubsubMessage message2 = PubsubMessage.newBuilder()
                      .setData(data2)
                      .setOrderingKey("user-A")
                      .build();
              publisher.publish(message2);
              System.out.println("Published: " + message2);

              // 发布消息 3 (不同的 ordering key)
              ByteString data3 = ByteString.copyFromUtf8("Message for user-B");
              PubsubMessage message3 = PubsubMessage.newBuilder()
                      .setData(data3)
                      .setOrderingKey("user-B")
                      .build();
              publisher.publish(message3);
              System.out.println("Published: " + message3);

          } finally {
              // 关闭 Publisher
              publisher.shutdown();
              publisher.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS);
          }
      }
  }
  

注意: 确保替换 your-project-id 和 your-topic-id 为你的实际项目 ID 和 Topic ID。

总结 📝

Google Cloud Pub/Sub 提供了 Keyed Ordering 机制来保证消息的顺序性。通过为消息指定 ordering_key，你可以确保具有相同 ordering_key 的消息按照发布的顺序传递给订阅者。在选择合适的方案时，请考虑你的具体业务需求和性能要求。 祝你一切顺利! 🎉