为什么同一组id的Kafka消费者不平衡？

Question

我正在编写一个概念验证应用程序来使用Apache Kafka0.9.0.0中的消息，看看是否可以使用它而不是通用的JMS消息代理，因为Kafka提供了好处。这是我的基本代码，使用新的消费者API：

public class Main implements Runnable {

    public static final long DEFAULT_POLL_TIME = 300;
    public static final String DEFAULT_GROUP_ID = "ltmjTest";

    volatile boolean keepRunning = true;
    private KafkaConsumer<String, Object> consumer;
    private String servers;
    private String groupId = DEFAULT_GROUP_ID;
    private long pollTime = DEFAULT_POLL_TIME;
    private String[] topics;

    public Main() {
    }

    //getters and setters...

    public void createConsumer() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);

        configs.put("enable.auto.commit", "true");
        configs.put("auto.commit.interval.ms", "1000");
        configs.put("session.timeout.ms", "30000");

        configs.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        configs.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<>(configs);
        consumer.subscribe(asList(topics));
    }

    public static void main(String[] args) {
        Main main = new Main();
        if (args != null && args.length > 0) {
            for (String arg : args) {
                String[] realArg = arg.trim().split("=", 2);
                String argKey = realArg[0].toLowerCase();
                String argValue = realArg[1];
                switch (argKey) {
                case "polltime":
                    main.setPollTime(Long.parseLong(argValue));
                    break;
                case "groupid":
                    main.setGroupId(argValue);
                    break;
                case "servers":
                    main.setServers(argValue);
                    break;
                case "topics":
                    main.setTopics(argValue.split(","));
                    break;
            }
        }
        main.createConsumer();
        new Thread(main).start();
        try (Scanner scanner = new Scanner(System.in)) {
            while(true) {
                String line = scanner.nextLine();
                if (line.equals("stop")) {
                    main.setKeepRunning(false);
                    break;
                }
            }
        }
    }
}

我使用默认设置启动了一个kafka服务器，并使用shell工具kafka-console-producer.sh启动了一个kafka生产者，以便将消息写入我的主题。然后，我使用这段代码与两个使用者连接，发送正确的服务器来连接，发送主题来订阅，其他一切都使用默认值，这意味着两个使用者都有相同的组ID。我注意到只有我的一个消费者使用所有的数据。我从官方教程中读到默认行为应该是服务器必须平衡消费者：

如果所有的使用者实例都有相同的使用者组，那么这就像传统的队列平衡使用者的负载一样。

我如何修复消费者的行为像默认的？也许我漏掉了什么？

Answer 1

有一个trait kafka.consumer.PartitionAssignor说明如何为每个消费者分配分区。它有两个模块：RoundRobinAssignor和RangeSignor。默认的是RangeAsSignor。

可以通过设置参数“Partition.Assignment.Strategy”来更改。

循环赛文档：

范围分配器在每个主题的基础上工作。对于每个主题，我们按数字顺序列出可用分区，按词典顺序列出使用者。然后，我们将分区数除以使用者总数，以确定分配给每个使用者的分区数。如果它没有均匀划分，那么前几个消费者将有一个额外的分区。例如，假设有两个使用者C0和C1，两个主题t0和t1，每个主题有3个分区，产生分区t0p0、t0p1、t0p2、t1p0、t1p1和T1P2。赋值将是:C0:[t0p0,t0p1,t1p0,t1p1]C1:[t0p2,t1p2]

所以，如果我们所有的主题只有一个分区，那么只有一个使用者可以工作