如何高效发送批量消息到Kafka Topic 及Kafka Streams处理方案

现有实现性能瓶颈分析 #

你当前的逐条调用kafkaTemplateForAvro.send()的写法，默认没有利用Kafka Producer的批量发送能力，加上如果参数配置不合理，会产生大量小IO请求，是10万条数据耗时1分30秒的核心原因。

更低耗时的实现方案 #

1. 调整Kafka Producer核心配置 #

这是性价比最高的优化方式，修改Spring Kafka的producer配置项：

• batch.size：从默认16KB调整到128KB~512KB，允许积攒更多消息再批量发送
• linger.ms：从默认0调整到5~10，允许最多等待指定毫秒数聚合小批量消息，不会显著增加延迟但能大幅提升吞吐量
• compression.type：设置为lz4或zstd，批量压缩消息后再发送，减少网络传输开销
• acks：如果业务可以容忍极少量消息丢失，设置为1（只需要分区leader确认写入即可），强一致场景再保留all
• buffer.memory：调整到32MB以上，避免Producer缓存不足导致发送阻塞

2. 优化发送逻辑，批量等待异步结果 #

kafkaTemplate.send()本身是异步调用，返回ListenableFuture，不需要逐条等待结果，收集所有发送任务的Future后统一等待即可，优化后代码示例：

Map<GenericRecord, GenericRecord> allMessages;
List<ListenableFuture<SendResult<GenericRecord, GenericRecord>>> sendFutures = new ArrayList<>(allMessages.size());

// 批量提交所有发送请求，充分利用Producer内置的批量聚合能力
allMessages.forEach((key, value) -> {
    sendFutures.add(kafkaTemplateForAvro.send(sink, key, value));
});

// 统一等待所有请求完成，处理异常
sendFutures.forEach(future -> {
    try {
        // 可根据业务需要添加超时时间，避免永久阻塞
        future.get(30, TimeUnit.SECONDS);
    } catch (Exception e) {
        // 自定义发送失败的处理逻辑，比如重试、记录失败日志等
    }
});

按照以上配置和逻辑优化后，10万条消息的发送耗时通常可以降到10秒以内。

Kafka Streams实现消息集合到Topic的处理流程 #

首先需要明确：Kafka Streams是流处理框架，更适合持续从上游Topic消费、处理后写入下游Topic的流式场景，如果你的原始数据是内存中的固定集合，直接用优化后的Kafka Template性价比更高，不需要额外引入Kafka Streams。

如果是流式场景下的消息处理写入，实现流程如下：

1. 配置Kafka Streams基础参数，包括服务地址、Avro序列化器、Schema Registry地址等
2. 构造流处理拓扑，定义数据消费、处理、写入的完整流程
3. 启动Streams实例，注册优雅关闭钩子

示例代码：

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "message-sink-process-app");
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-bootstrap-address:9092");
streamsProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
streamsProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
streamsProps.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "your-schema-registry-address");

StreamsBuilder streamsBuilder = new StreamsBuilder();
// 从上游源Topic消费数据
KStream<GenericRecord, GenericRecord> sourceStream = streamsBuilder.stream("your-source-topic");

// 可自定义中间处理逻辑，比如过滤、字段转换、聚合等
sourceStream
    .filter((key, value) -> value != null) // 示例：过滤空值消息
    .to("your-sink-topic"); // 写入目标Topic

KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), streamsProps);
// 启动流处理任务
kafkaStreams.start();
// 注册JVM关闭钩子，实现优雅停机
Runtime.getRuntime().addShutdownHook(new Thread(kafkaStreams::close));

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