在 Kubernetes 部署中的任意一个 Pod 上运行进程

Question

我有一个在多个 pod 上运行并在流量增加时横向扩展的应用程序。该应用程序的功能之一是“它从 Kafka 中挑选消息并触发电子邮件”。

当多个 pod 运行时，它们都会触发电子邮件，因为所有 pod 都应该按照设计接收消息。

如何限制电子邮件功能一次在任何一个 pod 上工作？

集群 - EKS ，编程语言 - Scala AKKA

Answer 1

如何限制电子邮件功能在任何一个 一个豆荚？

简而言之：对所有触发电子邮件的 pod 使用相同的消费者组。 通常，工作负载会根据它们所做的工作分为几组。同一组的成员彼此分担工作量。

您当然可以将 bootstrap.servers 等 Kafka 消费者配置提供给您的 pod。在该配置中，将名称为 group.id 的属性赋予某个值，例如 email-trigger-group，然后将按照您的预期共享工作负载。

您可以为触发电子邮件的 pod 使用 标签。您可以为您的消费者 group.id 为您的所有 pod 使用相同的标签值。

---

我们可以把问题分成两个子问题：

1.触发电子邮件

此工作负载可以由组中的多个使用者共享。

2。向前端应答请求

对整个主题（所有分区）使用手动consumer.assign()。

前端将指定 时间戳，它希望从中获取新消息，即时间戳为 > 的消息将从主题的所有分区中检索此时间戳。为此，请使用consumer.offsetsForTimes() 获取时间戳、轮询并将消息作为响应发送。

List<TopicPartition> topicPartitions = consumer.partitionsFor("your_topic").stream().map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()).toList();
consumer.assign(topicPartitions);

// Populate the map
Map<TopicPartition, Long> partitionTimestamp = new LinkedHashMap<>();

// Add the same timestamp received from frontend for all partitions
topicPartitions.forEach(topicPartition -> partitionTimestamp.put(topicPartition, timestampFromFrontend));

// Get the offsets and seek
Map<TopicPartition,OffsetAndTimestamp> offsetsForTimes = consumer.offsetsForTimes(offsetsForTimestamp);

// Seek to the offsets
offsetsForTimes.forEach( (tp, oft) -> consumer.seek(tp, oft.offset()) );

// Poll and return
consumer.poll();

Answer 2

如果您使用的是 Kafka，则可以使用分区。

每个主题可以有多个分区。这些分区在消费者之间共享。

例如：

Email Topic: Partitions[0,1,2,3,4,5]

Email Consumer Group:
   Consumer 1: Partitions[0,3]
   Consumer 2: Partitions[1,4]
   Consumer 3: Partitions[2,5]

On Scale Up Event:
   Consumer 1: Partitions[0]
   Consumer 2: Partitions[1]
   Consumer 3: Partitions[2,5]
   Consumer 4: Partitions[3]
   Consumer 5: Partitions[4]

On Scale Down Event:
   Consumer 1: Partitions[0,2,4]
   Consumer 2: Partitions[1,3,5]

这样只有特定组中的一个消费者会消费消息。