Kafka生产者：如何在kafka消费者之间公平地分配消息（而不是分区之间）

要在Kafka消费者之间公平地分配消息（而不是分区之间），可以使用Kafka的消费者组和自定义的分配策略。

下面是一个示例代码，演示如何使用自定义分配策略来实现公平地分配消息给消费者：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.util.*;

public class FairMessageDistribution {

    public static void main(String[] args) {
        String topic = "your_topic";
        String groupId = "your_group_id";
        String bootstrapServers = "your_bootstrap_servers";

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topic), new FairMessageDistribution.PartitionAssignor());

        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println("Received message: " + record.value());
                }
            }
        } finally {
            consumer.close();
        }
    }

    private static class PartitionAssignor implements ConsumerRebalanceListener {

        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // 当分区重新分配时，不做处理
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            // 在分区重新分配时，重新计算分配策略
            Map<TopicPartition, Long> partitionToOffset = new HashMap<>();
            for (TopicPartition partition : partitions) {
                partitionToOffset.put(partition, 0L); // 初始化每个分区的偏移量为0
            }

            ConsumerRecords<String, String> records = null;
            while (records == null || records.isEmpty()) {
                // 消费一条消息，以获取消费者组中其他消费者的偏移量信息
                records = consumer.poll(100);
            }

            for (ConsumerRecord<String, String> record : records) {
                TopicPartition partition = new TopicPartition(record.topic(), record.partition());
                // 更新分区的偏移量为消费者组中其他消费者的偏移量
                partitionToOffset.put(partition, record.offset());
            }

            // 将偏移量设置给消费者
            for (TopicPartition partition : partitions) {
                consumer.seek(partition, partitionToOffset.get(partition));
            }
        }
    }
}

上述代码中，我们创建了一个自定义的分区分配策略PartitionAssignor，并在消费者的subscribe()方法中传入该策略实例。在PartitionAssignor中，我们通过消费一条消息来获取消费者组中其他消费者的偏移量信息，并根据这些偏移量来重新分配分区给消费者。最后，我们调用consumer.seek()方法将偏移量设置给消费者，以确保公平地分配消息。

请注意，上述代码仅提供了一个思路和示例，实际应用中可能需要根据具体的业务逻辑和需求进行调整和优化。