我有一个消费者 worker 应用程序，它在内部启动 X 个线程，每个线程都生成它的 KafkaCosnumer。 Cosnumers 具有相同的 groupId 并且订阅了相同的主题。因此，每个消费者都能公平地分配分区。

处理的本质是我不能丢失消息，也不能允许重复。我运行的kafka版本是0.10.2.1。

这是我面临的问题:消费者线程 1 开始消费消息，并且在 poll() 上获取了一批消息。我还实现了 ConsumerRebalanceListener，这样每次成功处理消息时，它都会被添加到 offsets 映射中。 (见下面的代码。)因此，一旦重新平衡发生，我可以在我的分区重新分配给其他消费者之前提交我的偏移量。 有时，为了处理该批处理，它需要比 max.poll.interval.ms 更长的时间，这是重新平衡发生的地方，分区从消费者 1 中拉出并分配给消费者 2。消费者 1 没有不知道分区已被撤销并继续处理消息，与此同时，消费者 2 从最后一个偏移量(由 RebalanceListener 提交)获取并处理相同的消息。

有没有办法通知消费者他已撤销分区，以便他可以停止处理循环中的消息，这些消息已经分配给其他消费者？

public class RebalanceListener<K, V> implements ConsumerRebalanceListener {

    private final KafkaConsumer<K, V> consumer;

    private static final ConcurrentMap<TopicPartition, OffsetAndMetadata> CURRENT_OFFSETS =
            Maps.newConcurrentMap();

    private static final Logger LOGGER = LoggerFactory.getLogger(RebalanceListener.class);

    public RebalanceListener(KafkaConsumer<K, V> consumer) {
        this.consumer = consumer;
    }

    public void addOffset(String topic, int partition, long offset) {
        LOGGER.debug("message=Adding offset to offsets map, topic={}, partition={}, offset={}",
                topic, partition, offset);
        CURRENT_OFFSETS.put(new TopicPartition(topic, partition),
                new OffsetAndMetadata(offset, "commit"));
    }

    public Map<TopicPartition, OffsetAndMetadata> getCurrentOffsets() {
        return CURRENT_OFFSETS;
    }

    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        LOGGER.debug("message=following partitions have been revoked from consumer: [{}]",
                partitions.stream().map(
                        topicPartition -> topicPartition.topic() + ":" + topicPartition.partition())
                        .collect(joining(",")));
        LOGGER.debug("message=Comitting offsets for partititions [{}]",
                CURRENT_OFFSETS.keySet().stream().map(
                        topicPartition -> topicPartition.topic() + ":" + topicPartition.partition())
                        .collect(joining(",")));
        consumer.commitSync(CURRENT_OFFSETS);
    }

    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        LOGGER.debug("message=following partitions have been assigned to consumer: [{}]",
                partitions.stream().map(
                        topicPartition -> topicPartition.topic() + ":" + topicPartition.partition())
                        .collect(joining(",")));
    }

}

我想我可以在 RebalanceListener 内部创建 consumerId -- TopicPartition 的并发映射，然后在处理每条消息之前检查当前消费者是否仍然关联与记录(每个 ConsumerRecord 都有 topic 和 partition 字段)。 如果没有 - 打破循环并进行下一个 poll()。

如果我的工作应用程序将在一个实例中运行，这将是一个可行的解决方案，即使有多个 KafkaConsumer 线程在旋转。但是一旦我按比例放大，我将无法在静态映射中存储偏移量和消费者主题分区映射。这必须是某种集中式存储、数据库，或者比方说 Redis。

但是，在每次我处理一个项目之前，我都必须询问我的记录是否可以被当前的消费者线程合法地处理。在扩展的工作应用程序的情况下，它将是对外部存储的网络调用，这将破坏使用 kafka 的目的，因为它会减慢处理速度。我可能只是选择在处理单个项目后执行偏移量提交。

最佳答案

你需要实现 onPartitionsRevoked()

https://kafka.apache.org/0110/javadoc/org/apache/kafka/clients/consumer/ConsumerRebalanceListener.html#onPartitionsRevoked(java.util.Collection)

It is guaranteed that all consumer processes will invoke onPartitionsRevoked prior to any process invoking onPartitionsAssigned. So if offsets or other state is saved in the onPartitionsRevoked call it is guaranteed to be saved by the time the process taking over that partition has their onPartitionsAssigned callback called to load the state.

关于java - 卡夫卡再平衡。重复处理问题，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/47103436/