Kafka轮询分配策略不起作用

如果 Kafka 的轮询分配策略不起作用，可能有以下几个原因和解决方法：

1. 消费者组配置错误：确保消费者组的配置参数正确。消费者组的名称需要在同一应用程序中唯一，而且每个消费者组都应该有不同的消费者 ID。可以通过设置group.id和client.id来指定消费者组的名称和消费者 ID。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-consumer-group");
props.put("client.id", "my-consumer-client");

2. 分区分配器配置错误：确保正确配置了分区分配器。默认情况下，Kafka 使用的是RangeAssignor分区分配器，但你也可以自定义分区分配器。如果需要自定义分区分配器，可以通过设置partition.assignment.strategy属性来指定。

props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");

3. 消费者线程数量不正确：如果消费者线程数量小于分区数量，可能导致某些分区没有分配给任何消费者。确保消费者线程数量等于或大于分区数量。

int numConsumers = 3; // 消费者线程数量
int numPartitions = 5; // 分区数量

// 每个消费者线程处理的分区数量
int partitionsPerConsumer = numPartitions / numConsumers;

4. 消费者重新平衡：在消费者组中增加或减少消费者时，Kafka 会触发重新平衡。重新平衡期间，消费者可能被重新分配分区。可以通过设置max.poll.interval.ms属性来增加消费者超时时间，以防止重新平衡过程中的长时间轮询。

props.put("max.poll.interval.ms", "300000"); // 5分钟的超时时间

5. 代码示例：

以下是一个使用 Kafka 消费者的示例代码，其中包含了上述解决方法中的配置示例：

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

import java.util.Collections;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        String topic = "my-topic";
        String groupId = "my-consumer-group";
        String clientId = "my-consumer-client";

        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", groupId);
        props.put("client.id", clientId);
        props.put("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
        props.put("max.poll.interval.ms", "300000"); // 5分钟的超时时间

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

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

请根据你的实际需求调整上述示例代码中的配置参数。