java - Kafka - 如何检查消费者是否还活着，如果不是，如何将消费者恢复到运行状态？

Question

我目前正在开发一个 kafka java 项目。我是新手，我发现很难理解与 Kafka 生产者/消费者设计相关的一些基本概念。

1. 假设，我有一个具有单个分区的主题，并且有一个生产者写入该主题，还有一个消费者从该主题消费。如果我部署同一应用程序的多个实例，每个实例都将运行它自己的使用者。这样的话，由于所有的消费者都属于同一个groupId，那么消息会均匀地分布在运行在多个实例上的消费者之间吗？

2. 如何从应用程序中定期检查消费者是否存活？

3. 如果一个/所有消费者出现故障，我该如何重新启动/使其恢复到运行状态？
4. 在 java 中实现长时间运行的消费者的最佳方式是什么？

请对上述疑问进行澄清。如果我的任何/所有假设/理解是错误的，请纠正我。我知道我没有分享任何代码示例，因为这些是概念性问题。如果需要，我可以分享代码片段。

Answer 1

1. 您说具有单个分区的主题意味着它无法将消息分发到多个分区。你将失去 Kafka 的一大优势。您必须将分区增加一个以上。如果您部署同一应用程序的多个实例，则它无助于分发，因为正如您提到的，消息将发布到一个分区，并且只有一个实例只会分配给该分区，其他实例将处于空闲状态。

   <

2. 您可以使用 AdminClient Kafka API 检查您的消费者是否有任何滞后

   Properties props = new Properties();
   props.setProperty("bootstrap.servers", "localhost:9091");
   KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
   AdminClient client = org.apache.kafka.clients.admin.AdminClient.create(props);
   ListConsumerGroupOffsetsResult offsets = client.listConsumerGroupOffsets("consumerId");
   Map<TopicPartition, OffsetAndMetadata> tt = offsets.partitionsToOffsetAndMetadata().get();
   ListConsumerGroupOffsetsResult offsets = client.listConsumerGroupOffsets(consumerId);
   Map<TopicPartition, OffsetAndMetadata> tt = offsets.partitionsToOffsetAndMetadata().get();
   for (Entry<TopicPartition, OffsetAndMetadata> entry : tt.entrySet()) {TopicPartition tp = entry.getKey();
     OffsetAndMetadata op = entry.getValue();
     Collections.singletonList(tp);
     consumer.assign(Collections.singletonList(tp));
     consumer.seekToEnd(Collections.singletonList(tp));
     System.out.println(consumerId + "," + tp.partition() + "," + consumer.position(tp) + ","
                               + op.offset() + "," + (consumer.position(tp) - op.offset()));
   }
   

3. 您没有说明您的部署位置，但如果您在 mesos 中使用 marathon 进行部署，它将自动重新启动。您可以手动重新启动，如果您使用与之前相同的组 ID，您的应用程序将开始使用它留下的位置。