1. 优化消息发送效率
批量发送
- 原理：将多条消息批量打包发送，减少网络请求次数，从而提高发送效率。
- 适用性：适用于对消息实时性要求不是特别高，但对发送性能要求较高的场景，如日志收集等。
- 代码示例（Kafka Producer）：
Properties props = new Properties();
props.put("bootstrap.servers", "your - kafka - brokers");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384); // 批次大小，默认16KB
props.put("linger.ms", 1); // 延迟时间，默认0，即不等待直接发送，设置1表示最多等待1ms以收集更多消息形成批次
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
producer.send(new ProducerRecord<>("your - topic", "key" + i, "value" + i));
}
producer.close();
异步发送
- 原理：使用异步方式发送消息，生产者无需等待消息发送确认，继续执行后续操作，提高系统的并发处理能力。
- 适用性：适用于对系统响应时间敏感，允许一定程度上的消息发送不确定性（如允许少量消息发送失败重试）的场景，如实时监控数据上报。
- 代码示例（Kafka Producer）：
Properties props = new Properties();
props.put("bootstrap.servers", "your - kafka - brokers");
props.put("acks", "all");
props.put("retries", 0);
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 100; i++) {
producer.send(new ProducerRecord<>("your - topic", "key" + i, "value" + i), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception != null) {
exception.printStackTrace();
} else {
System.out.println("Message sent to partition " + metadata.partition() + " at offset " + metadata.offset());
}
}
});
}
producer.close();
2. 优化消息接收效率
多线程消费
- 原理：创建多个消费者线程并行处理消息，提高消息消费速度。
- 适用性：适用于消息处理逻辑相对独立且耗时的场景，如数据处理、计算任务等。
- 代码示例（Kafka Consumer）：
Properties props = new Properties();
props.put("bootstrap.servers", "your - kafka - brokers");
props.put("group.id", "your - group - id");
props.put("auto.offset.reset", "earliest");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("your - topic"));
ExecutorService executor = Executors.newFixedThreadPool(5);
for (int i = 0; i < 5; i++) {
executor.submit(() -> {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.println("Thread " + Thread.currentThread().getName() + " received message: " + record.value());
}
}
});
}
合理设置fetch参数
- 原理：通过调整每次拉取消息的数量（
fetch.max.bytes）和等待时间（fetch.min.bytes、fetch.wait.max.ms），平衡网络开销和消息获取及时性。
- 适用性：适用于不同网络环境和消息负载的场景。如网络带宽充足时，可适当增大
fetch.max.bytes以减少拉取次数；网络不稳定时，可合理设置等待时间参数。
- 代码示例（Kafka Consumer）：
Properties props = new Properties();
props.put("bootstrap.servers", "your - kafka - brokers");
props.put("group.id", "your - group - id");
props.put("auto.offset.reset", "earliest");
props.put("fetch.max.bytes", 5242880); // 每次拉取最大字节数，默认5MB
props.put("fetch.min.bytes", 1); // 每次拉取最小字节数，默认1
props.put("fetch.wait.max.ms", 500); // 等待拉取到足够消息的最大时间，默认500ms
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("your - topic"));
3. 处理消息堆积问题
增加分区
- 原理：增加Kafka主题的分区数，提高并行处理能力，加快消息消费速度，从而缓解消息堆积。
- 适用性：适用于消费者处理能力有提升空间，且业务允许增加分区并行处理的场景。但增加分区会带来一些管理成本，如分区分配策略调整等。
- 代码示例（创建Kafka主题时设置分区数）：
kafka - topics.sh --create --bootstrap - servers your - kafka - brokers --replication - factor 1 --partitions 10 --topic your - topic
优化消费者处理逻辑
- 原理：减少消费者处理消息的时间，如优化业务逻辑、减少I/O操作等，提高消费速度。
- 适用性：适用于任何消息堆积场景，优化消费者处理逻辑是通用的解决方案。例如在处理数据库写入时，可采用批量写入等方式减少I/O次数。
启用死信队列
- 原理：将处理失败的消息发送到死信队列（如单独的Kafka主题），避免这些消息阻塞正常消息的处理，后续再对死信队列中的消息进行专门处理。
- 适用性：适用于对消息处理准确性要求较高，不允许消息丢失，但处理过程可能出现失败的场景，如订单处理等业务。
- 代码示例（简单模拟死信队列处理，以Kafka Streams为例）：
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("your - topic");
source.filter((key, value) -> {
// 模拟消息处理失败判断
return false;
}).to("dead - letter - topic");
