当我暂停消费者时，来自同一消费者组的其他消费者应该接收消息答案

【问题标题】：When I Pause a Consumer other Consumer from the same Consumer group should pick up the messages当我暂停消费者时，来自同一消费者组的其他消费者应该接收消息
【发布时间】：2017-08-21 07:09:00
【问题描述】：

我正在尝试处理一个有两个或多个消费者的案例，他们正在收听一个只包含一个分区的常见主题。我正在暂停其中一个消费者，在此期间，其他未暂停的消费者应该能够从主题中获取消息。我观察到，只有当暂停的消费者恢复时，其他活动消费者之一才能接收消息。我怎样才能通过spring kakfka实现这一点。我正在使用 Spring kafka 2.0.0.M2。

下面是我的代码行

public class listener{
    @KafkaListener(id = "id2", topics = "abcd", group = "group1", containerFactory = "kafkaListenerContainerFactory")
        public void listenPartition1(String data, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions, @Header(KafkaHeaders.OFFSET) Long offsets, Acknowledgment acknowledgment, Consumer consumer) throws InterruptedException {
            count_consumer2 = count_consumer2 + 1;
            if(count_consumer2 == 10) {
                consumer.pause(consumer.assignment());
            }
            acknowledgment.acknowledge();

    }

    @KafkaListener(id = "id1", topics = "abcd", group = "group1", containerFactory = "kafkaListenerContainerFactory")
        public void listenPartition1(String data, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions, @Header(KafkaHeaders.OFFSET) Long offsets, Acknowledgment acknowledgment, Consumer consumer) throws InterruptedException {
            count_consumer1 = count_consumer1 + 1;
            if(count_consumer1 == 5) {
                consumer.pause(consumer.assignment());
                while(data.database <10){
                      consumer.resume(consumer.assignment());
                 }
            }

             acknowledgment.acknowledge();



        }
}

【问题讨论】：

使用 pause，实际上是在暂停多个分区。当一个分区暂停时，在它再次恢复之前，不会从中消耗任何记录。见kafka.apache.org/0110/javadoc/org/apache/kafka/clients/consumer/…
在你的情况下，你需要做的是关闭一个消费者，然后另一个消费者会自动处理分区。
@amethystic 你知道我如何实现消费者关闭。我正在使用 consumer.close() 但它抛出了一个错误。谢谢
什么错误？ Close() 应该可以工作。

标签： apache-kafka kafka-consumer-api spring-kafka

【解决方案1】：

查看 pause() 方法文档：

暂停从请求的分区中提取。未来对 poll(long) 的调用将不会从这些分区返回任何记录，直到使用 resume(Collection) 恢复它们。

请注意，此方法不影响分区订阅。特别是在使用自动分配时，它不会导致组重新平衡。

pause 方法不会触发分区重新平衡，因此该组中的其他消费者不会消费该分区，并且请注意，一个分区将仅被同一消费者组中的一个消费者消费。所以在你的情况下，只是一个消费者工作，其他人闲置。

【讨论】：

如果我的一个消费者仍在处理消息或暂停同一组的其他消费者可以检索消息，您是否知道如何实现我的上述用例？关闭消费者是一种选择，但如果我必须多次这样做，这不是一项昂贵的操作吗？
@Anand Bangad 你可以在不同的消费者组中使用消费者，如果这些消费者在同一个组中，则不能这样做。如果你使用两个属于不同组的消费者，这两个消费者可以同时从一个分区读取消息。
我也不希望不同的消费群体读取重复数据。那里可以同步吗？
恐怕很难做到，可能不符合kafka的设计理念。我不知道你为什么要这样做，也许你应该解决这个问题。

【解决方案2】：

您应该对主题进行重新分区，使其具有至少与活动的互斥消费者（即相同的 group.id）一样多的分区。这就是 Kafka 的工作原理。

【讨论】：

我同意。您还可以拥有比分区更多的消费者以实现一定程度的冗余（如果消费者死亡，额外的消费者可以接手工作）