Kafka 9 KafkaConsumer multiple 似乎没有并行处理消息

Question

我在 3 个服务器上有一个 Kafka 集群，所以我创建了一个分区 3 的主题，复制因子为 3。

bin/kafka-topics.sh --create --zookeeper tstaapp001:52181,ewdlxsrv283:52181,devcapp001:52181 --replication-factor 3 --partitions 3 --topic TEST_PARTITION

我有一个使用 Kafka 9.0 从博客复制的多线程消费者示例程序，但在我的 8 核 PC 上运行时，它并没有真正并行接收消息。谁能告诉我我的设置或代码是否有问题？

public class Consumer implements Runnable {
    private final KafkaConsumer<String, String> consumer;
    private final List<String> topics;
    private final int id;

    public Consumer(int id, String groupId, List<String> topics) {
        this.id = id;
        this.topics = topics;
        Properties props = new Properties();
        props.put("bootstrap.servers",
                "tstaapp001:59092,ewdlxsrv283:59092,devcapp001:59092");
        props.put("group.id", groupId);
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<>(props);
    }

    @Override
    public void run() {
        try {
            consumer.subscribe(topics);

            while (true) {
                ConsumerRecords<String, String> records = consumer
                        .poll(Long.MAX_VALUE);
                for (ConsumerRecord<String, String> record : records) {
                    Map<String, Object> data = new HashMap<>();
                    data.put("partition", record.partition());
                    data.put("offset", record.offset());
                    data.put("value", record.value());
                    System.out.println(this.id + ": " + data);
                }
            }
        } catch (WakeupException e) {
            // ignore for shutdown
        } finally {
            consumer.close();
        }
    }

    public void shutdown() {
        consumer.wakeup();
    }

    public static void main(String[] args) {
        int numConsumers = 3;
        String groupId = "TEST-GROUP";
        List<String> topics = Arrays.asList("TEST_PARTITION");

        ExecutorService executor =     Executors.newFixedThreadPool(numConsumers);

        final List<Consumer> consumers = new ArrayList<>();
        for (int i = 0; i < numConsumers; i++) {
            Consumer consumer = new Consumer(i, groupId, topics);
            consumers.add(consumer);
            executor.submit(consumer);
        }

        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                for (Consumer consumer : consumers) {
                    consumer.shutdown();
                }
                executor.shutdown();
                try {
                    executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }
}

以下是输出：

1: {partition=1, offset=141, value=TEST1}
1: {partition=1, offset=142, value=TEST4}
1: {partition=1, offset=143, value=TEST7}
1: {partition=1, offset=144, value=TEST10}
1: {partition=1, offset=145, value=TEST13}
1: {partition=1, offset=146, value=TEST16}
1: {partition=1, offset=147, value=TEST19}
1: {partition=1, offset=148, value=TEST22}
1: {partition=1, offset=149, value=TEST25}
1: {partition=1, offset=150, value=TEST28}
1: {partition=1, offset=151, value=TEST31}
1: {partition=1, offset=152, value=TEST34}
1: {partition=1, offset=153, value=TEST37}
1: {partition=1, offset=154, value=TEST40}
1: {partition=1, offset=155, value=TEST43}
1: {partition=1, offset=156, value=TEST46}
1: {partition=1, offset=157, value=TEST49}
0: {partition=0, offset=145, value=TEST2}
0: {partition=0, offset=146, value=TEST5}
0: {partition=0, offset=147, value=TEST8}
0: {partition=0, offset=148, value=TEST11}
0: {partition=0, offset=149, value=TEST14}
0: {partition=0, offset=150, value=TEST17}
0: {partition=0, offset=151, value=TEST20}
0: {partition=0, offset=152, value=TEST23}
0: {partition=0, offset=153, value=TEST26}
0: {partition=0, offset=154, value=TEST29}
0: {partition=0, offset=155, value=TEST32}
0: {partition=0, offset=156, value=TEST35}
0: {partition=0, offset=157, value=TEST38}
0: {partition=0, offset=158, value=TEST41}
0: {partition=0, offset=159, value=TEST44}
0: {partition=0, offset=160, value=TEST47}
2: {partition=2, offset=142, value=TEST0}
2: {partition=2, offset=143, value=TEST3}
2: {partition=2, offset=144, value=TEST6}
2: {partition=2, offset=145, value=TEST9}
2: {partition=2, offset=146, value=TEST12}
2: {partition=2, offset=147, value=TEST15}
2: {partition=2, offset=148, value=TEST18}
2: {partition=2, offset=149, value=TEST21}
2: {partition=2, offset=150, value=TEST24}
2: {partition=2, offset=151, value=TEST27}
2: {partition=2, offset=152, value=TEST30}
2: {partition=2, offset=153, value=TEST33}
2: {partition=2, offset=154, value=TEST36}
2: {partition=2, offset=155, value=TEST39}
2: {partition=2, offset=156, value=TEST42}
2: {partition=2, offset=157, value=TEST45}
2: {partition=2, offset=158, value=TEST48}

Answer 1

马丁，

您的消费者代码看起来不错。您是否尝试过运行消费者的多个实例。例如看看我的消费者https://github.com/sdpatil/KafkaAPIClient/blob/master/src/main/java/com/spnotes/kafka/partition/Consumer.java

如果您说主题有 3 个分区，并且您想为所有 3 个分区并行消费消息（您不能并行消费一个分区），启动 3 个消费者实例，您会注意到 Kafka 分配了一个分区给每个消费者和他们将并行接收消息。在我的示例消费者中有 ConsumerRebalanceListener，它将帮助您调试正在发生的事情

苏尼尔

Answer 2

我建议您使用 rapids-kafka-client，这是一个为您提供并行性的库。

ConsumerConfig.<String, String>builder()
.prop(KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.prop(VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName())
.prop(GROUP_ID_CONFIG, "stocks")
.topics("stock_changed")
.consumers(7)
.callback((ctx, record) -> {
  log.info("status=consumed, value={}", record.value());
})
.build()
.consume()
.waitFor();