如何让Kafka Producer一次性批量发送多条记录（多消息）

1. 直接调用批量发送API #

Kafka Producer原生提供了批量发送的重载方法，你只需将List<Employee>转换为List<ProducerRecord<String, Employee>>集合，即可一次性提交：

// 假设Employee已实现序列化逻辑（自定义Serializer或实现Serializable）
List<Employee> employees = ...; // 你的目标员工集合
String topicName = "your_target_topic";

// 转换为ProducerRecord集合，可指定消息key（比如用员工ID）
List<ProducerRecord<String, Employee>> records = employees.stream()
    .map(emp -> new ProducerRecord<>(topicName, emp.getEmployeeId(), emp))
    .collect(Collectors.toList());

// 批量发送，支持异步回调处理结果
producer.send(records, (metadata, exception) -> {
    if (exception != null) {
        // 处理批量发送失败逻辑，比如日志记录或重试
        exception.printStackTrace();
    } else {
        // 发送成功后的业务处理
        System.out.println("批量发送完成，最后一条消息偏移量：" + metadata.offset());
    }
});

如果不需要指定消息key，可简化为new ProducerRecord<>(topicName, emp)构造方法。

2. 配置Producer自动攒批（生产环境首选） #

如果你不想手动处理集合转换，通过调整Producer的核心配置，能让它在后台自动攒批发送消息，平衡吞吐量和延迟：

• batch.size：单个批次的最大字节数（默认16KB），达到阈值自动发送
• linger.ms：等待攒批的最长时间（默认0），设置为5-10ms，即使批次未达大小，到时间也会发送
• buffer.memory：Producer缓存待发送消息的内存总量（默认32MB）

示例配置：

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, EmployeeSerializer.class.getName());
// 批量优化配置
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768); // 32KB
props.put(ProducerConfig.LINGER_MS_CONFIG, 5); // 等待5ms攒批
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864); // 64MB

KafkaProducer<String, Employee> producer = new KafkaProducer<>(props);

这种方式下，你依然可以逐条调用send()，Producer会自动在后台攒批提交，无需手动管理集合。

3. 关键注意事项 #

• 序列化保障：必须确保Employee类能被Kafka正确序列化，推荐自定义Serializer而非依赖默认的ByteArraySerializer，便于后续维护。
• 错误处理：批量发送时若部分消息失败，回调中的exception会捕获异常，需根据业务场景选择重试、丢弃或死信队列处理。
• 同步/异步选择：批量send()默认异步，若需同步等待结果，可调用producer.flush()或send().get()（生产环境不推荐后者，会阻塞线程）。

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