Kafka-如何检查消费者是否活着，如果不是如何将消费者带回运行状态？

Question

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

>

比方说，我有一个带有单个分区的主题，我有一个生产者正在写这个主题，一个消费者正在从这个主题中消费。如果我部署同一个应用程序的多个实例，每个实例将运行自己的消费者。在这种情况下，因为所有消费者都属于同一个group pId，所以消息是否会在多个实例上运行的消费者之间平均分配？

如何从应用程序中定期检查消费者是否还活着？。

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

Answer 1

>

Kafka只允许每个分区和用户组有一个用户。因此，多个消费者将等待主消费者倒下，然后在该点跳入，但对于给定的分区/消费者组，一次只有一个消费者在消费。

您可以检查消费者组偏移是否与最大偏移保持一致，以查看是否存在滞后。如果一个人开始有问题，Kafka在循环多个消费者方面做得很好

重启应用程序，它会从中断的地方开始，只是不要从一开始就重启消息。Kafka为你处理。

我们用Spring靴搭配SpringKafka。没有“最佳”方法，但我们发现使用spring进行设置和维护很容易。

Answer 2

>

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

您可以使用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()));
}

您还没有说明在何处部署，但如果您在mesos中使用marathon进行部署，它将自动重新启动。您可以手动重新启动，如果您使用与上一个相同的组id，您的应用程序将开始使用它离开的位置。