java - Kafka Consumer 对多线程访问不安全异常

Question

当我使用 kafka 创建消费者时，出现以下异常：

例外：

Kafka Consumer 对多线程访问不安全异常

100多个消费者同时订阅主题，同时获取记录。

我添加了以下代码

//KafkaConsumer Configuration 

KafkaConsumer<String, Object> kafkaConsumer;
Properties properties = new Properties();
properties.put("bootstrap.servers", KAFKA_SERVER_URL + ":" + KAFKA_SERVER_PORT);
properties.put("group.id", GROUP_ID);
properties.put("enable.auto.commit", "false");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "earliest");
properties.put("key.deserializer", JsonDeserializer.class);
properties.put("value.deserializer", JsonDeserializer.class);
kafkaConsumer = new KafkaConsumer<>(properties);


//Subscribe Consumer

kafkaConsumer.subscribe(Arrays.asList("Topic_Name"));

//Get Records From Beginning

while (true) {
    kafkaConsumer.seekToBeginning(kafkaConsumer.assignment());
    ConsumerRecords<String, Object> records = kafkaConsumer.poll(Duration.ofMillis(1000));
    for (ConsumerRecord<String, Object> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), 
            record.key(), record.value());
    }
}

//Get Records Using WebSocket this section continuously

ConsumerRecords<String, Object> records = kafkaConsumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord<String, Object> record : records) {
    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), 
        record.key(), record.value());
}

标签： javaapache-kafkakafka-consumer-api