如何在 Kubernetes multipod 部署中使用 spring kafka 处理 Kafka 容器生命周期

Question

我正在使用 Spring kafka 实现，我需要通过 REST API 启动和停止我的 kafka 消费者。为此，我正在使用 KafkaListenerEndpointRegistry endpointRegistry

endpointRegistry.getListenerContainer("consumer1").stop();

endpointRegistry.getListenerContainer("consumer1").start();

我们正在 kubernetes pod 上部署微服务，因此同一个微服务可能有多个部署。我怎样才能设法启动和停止所有容器上的消费者。

Answer 1

Kubernetes 没有提供任何自动向所有 pod 广播 http 请求以获取服务的功能；所以你必须自己做。

通过 Kafka 广播

您可以将接收http请求的单个实例的启动/停止命令发布到某个主题，专用于在所有实例之间广播命令。

当然，您必须确保每个实例都可以读取该主题的所有消息，因此您需要防止这些实例之间的分区平衡。您可以通过在该主题的 Consumer 上设置唯一的 group-id（例如，在正常的 groupId 后添加 UUID）来实现这一点。

通过 Http 广播

Kubernetes 知道哪些 Pod 正在侦听哪些端点，您可以在服务中获取该信息。 Spring Cloud Kubernetes (https://cloud.spring.io/spring-cloud-static/spring-cloud-kubernetes/2.0.0.M1/reference/html/#ribbon-discovery-in-kubernetes) 让一切变得简单获取该信息；可能有很多不同的方法可以做到这一点，使用 Spring Cloud Kubernetes 会是这样的：

在随机选择的 pod 上接收命令，从 Ribbon 获取服务的 ServerList（它包含所有实例以及可以访问它们的 IP 地址/端口），并向每个实例发送新的 http 请求。

我更喜欢 Kafka 方法，因为它的稳健性，如果您已经在使用 Spring Cloud，http 方法可能更容易实现。

Answer 2

您是否获得了该问题的实施？请分享。我也有相同的用例，即停止 kubernetes 上所有正在运行的 pod 上的侦听器。