原代码
public class KafkaConsumerDemo {
public static void main(String[] args) {
int numConsumers = 5; // 增加消费者的数量
for (int i = 0; i < numConsumers; i++) {
new Thread(new KafkaConsumerThread()).start();
}
}
static class KafkaConsumerThread implements Runnable {
@Override
public void run() {
// 配置消费者属性
Properties props = new Properties();
props.put("bootstrap.servers", "114.15.78.14:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 调整消费者配置
props.put("fetch.min.bytes", "1024"); // 最小获取1KB的数据
props.put("fetch.max.wait.ms", "500"); // 最大等待500ms
// 创建消费者实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 订阅主题
consumer.subscribe(Collections.singletonList("test-topic"));
// 消费消息; 如果消息处理逻辑允许，可以批量处理消息，而不是逐条处理
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
if (!records.isEmpty()) {
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}
// 批量提交偏移量
consumer.commitSync();
}
}
}
}
}
在 Kafka 消费者代码中，有几个地方可以进行优化以提高消费速度：
增加消费者线程：你已经增加了消费者的数量，这是一个很好的开始。确保每个消费者线程都在不同的分区上消费，以实现并行处理。
调整 fetch.min.bytes 和 fetch.max.wait.ms：这两个参数控制了消费者从 Kafka
服务器获取数据的最小字节数和最大等待时间。如果你的消息较小，增加 fetch.min.bytes
可能会导致消费者等待更长时间以收集足够的数据，这可能会降低消费速度。如果你的消息较大，可以适当增加这个值。
异步提交偏移量：你当前使用的是 consumer.commitSync() 进行同步提交偏移量，这会阻塞当前线程直到提交完成。可以考虑使用
consumer.commitAsync() 异步提交，这样可以不阻塞消费线程。增加 max.poll.records：poll
方法有一个参数可以设置一次拉取的最大记录数。如果设置得太低，可能会导致频繁的网络请求，增加延迟。
合理设置 auto.commit.interval.ms：自动提交偏移量的间隔设置得太短可能会导致不必要的性能开销，可以适当增加这个值。
使用多线程处理消息：如果你的消息处理逻辑是 CPU 密集型的，可以考虑使用线程池来并行处理消息。
监控和调优：使用 Kafka 的监控工具来观察消费者的行为，找出瓶颈并进行调优。
下面是根据上述建议优化后的代码示例：
public class KafkaConsumerDemo {
public static void main(String[] args) {
int numConsumers = 5; // 增加消费者的数量
for (int i = 0; i < numConsumers; i++) {
new Thread(new KafkaConsumerThread()).start();
}
}
static class KafkaConsumerThread implements Runnable {
@Override
public void run() {
// 配置消费者属性
Properties props = new Properties();
props.put("bootstrap.servers", "114.15.78.14:9092");
props.put("group.id", "test-group");
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "5000"); // 增加自动提交偏移量的间隔
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
// 调整消费者配置
props.put("fetch.min.bytes", "1"); // 减少最小获取字节数
props.put("fetch.max.wait.ms", "100"); // 减少最大等待时间
props.put("max.poll.records", "500"); // 增加一次拉取的最大记录数
// 创建消费者实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// 订阅主题
consumer.subscribe(Collections.singletonList("test-topic"));
// 消费消息
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
if (!records.isEmpty()) {
processRecords(records); // 异步处理消息
consumer.commitAsync(); // 异步提交偏移量
}
}
}
private void processRecords(ConsumerRecords<String, String> records) {
// 异步处理消息的逻辑
for (ConsumerRecord<String, String> record : records) {
System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
// 这里可以添加消息处理逻辑，例如使用线程池并行处理
}
}
}
}
