Kafka Consumer接收记录后无法存入ArrayList的问题求助

从你的代码和描述来看，既然能打印出Received data: ...，说明Kafka Consumer已经成功接收到记录，result.add(record.value())代码也已经执行。问题核心在于变量作用域：result是方法内部的局部变量，方法执行完毕后会被销毁，外部无法获取它的值；另外也可能是TIMEOUT_MILLIS设置过短，导致循环提前退出，没来得及处理所有记录。

方案1：修改方法返回值，返回接收的记录列表 #

将方法的返回类型从void改为List<String>，最后返回result，调用方就能直接拿到数据：

public List<String> subscribeFromKafka() throws Exception {

    List<String> result = new ArrayList<>();
    
    Properties props = new Properties();
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test");
    props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    props.setProperty("auto.commit.interval.ms", "1000");
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        System.out.println("Receiving from Kafka . . .");
        consumer.subscribe(Arrays.asList(KAFKA_TOPIC_URL));
        long endTimeMillis = System.currentTimeMillis() + TIMEOUT_MILLIS;
    
        while (System.currentTimeMillis() < endTimeMillis) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received data: " + record.value());
                result.add(record.value());
            }
        }

        System.out.println(result + " this is the result from Kafka.");
    } catch (Exception e) {
        throw new Exception("Failed to subscribe from Kafka", e);
    }
    return result;
}

// 调用方式
List<String> kafkaRecords = subscribeFromKafka();
// 处理kafkaRecords中的数据

方案2：传入外部列表作为参数 #

直接将需要填充的列表作为参数传入方法，方法内部直接往该列表中添加记录，外部调用方可以直接访问这个列表：

public void subscribeFromKafka(List<String> result) throws Exception {
    // 不再新建局部列表，直接使用传入的列表
    Properties props = new Properties();
    props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_SERVERS);
    props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "test");
    props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
    props.setProperty("auto.commit.interval.ms", "1000");
    props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
    try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
        System.out.println("Receiving from Kafka . . .");
        consumer.subscribe(Arrays.asList(KAFKA_TOPIC_URL));
        long endTimeMillis = System.currentTimeMillis() + TIMEOUT_MILLIS;
    
        while (System.currentTimeMillis() < endTimeMillis) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                System.out.println("Received data: " + record.value());
                result.add(record.value());
            }
        }

        System.out.println(result + " this is the result from Kafka.");
    } catch (Exception e) {
        throw new Exception("Failed to subscribe from Kafka", e);
    }
}

// 调用方式
List<String> kafkaRecords = new ArrayList<>();
subscribeFromKafka(kafkaRecords);
// 此时kafkaRecords中已包含接收到的记录

额外检查点 #

• 确认TIMEOUT_MILLIS的数值：如果设置过小（比如100ms），Consumer可能还没完成集群握手、拉取数据就退出循环了，建议设置为3000-5000ms，或根据业务需求调整。
• 持续消费场景：如果需要一直监听Kafka，不要用固定超时的循环，改用无限循环配合退出条件（比如收到停止信号、特定消息等）。

内容的提问来源于stack exchange，提问作者pagesays404