要优化 Kafka 消费者性能，你可以考虑以下策略：
-
并行消费：通过增加消费者组中的消费者数量来并行处理更多的消息，从而提升消费速度。
-
批量消费：配置
fetch.min.bytes和fetch.max.wait.ms参数来控制批量消费的大小和等待时间，减少网络开销。 -
手动提交偏移量：使用手动提交偏移量（通过设置
enable.auto.commit=false并使用commitSync或commitAsync方法），提高消费的可靠性和灵活性。 -
优化配置：根据具体场景优化 Kafka 配置，如调整日志保留策略（
log.retention.hours、log.retention.bytes等）、消费者拉取策略（fetch.min.bytes、fetch.max.wait.ms等）；根据实际需求设置合适的复制因子（replication.factor）和最小同步副本数（min.insync.replicas）等。 -
监控和维护：使用 Kafka 提供的 JMX（Java Management Extensions）指标，或集成第三方监控工具（如 Prometheus、Grafana）来实时监控 Kafka 集群的性能。
-
日志管理：定期检查和清理日志文件，确保磁盘空间充足。配置
log.cleanup.policy参数（如 delete 或 compact）来控制日志清理策略。 -
集群维护：定期进行 Kafka 和 Zookeeper 集群的维护和升级，确保系统的稳定性和安全性。
-
分区设计：合理设计消息的分区策略，可以均衡负载，提升整体吞吐量。
-
批处理和压缩：启用数据压缩功能（如GZIP或Snappy），可以减少网络传输的数据量，进而提升吞吐量。
-
硬件资源优化：监控硬件资源使用情况，发现潜在的性能瓶颈；优化硬件配置和资源分配策略，确保资源得到充分利用。
-
Broker 配置调优：调整 Broker 配置，如
log.segment.bytes优化日志存储结构，提升读写性能。 -
Zookeeper 优化：合理配置 Kafka 的副本数量和 ISR（In-Sync Replicas）列表，优化写入性能。
通过实施这些优化策略，你可以提升 Kafka 消费者性能，确保 Kafka 集群的高效运行。
package com.mita.web.core.config.kafka;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
/**
* @author sunpeiyang
* @date 2024/11/12 14:54
*/
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 {
private static final int ALERT_THRESHOLD = 1000; // 设置告警阈值
@Override
public void run() {
// 配置消费者属性
Properties props = new Properties();
props.put("bootstrap.servers", "ip: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); // 异步处理消息
checkLag(ALERT_THRESHOLD, consumer, "test-topic"); // 检查滞后并告警
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());
// 这里可以添加消息处理逻辑，例如使用线程池并行处理
}
}
private void checkLag(int threshold, KafkaConsumer<String, String> consumer, String topic) {
for (TopicPartition partition : consumer.assignment()) {
long currentOffset = consumer.position(partition);
long endOffset = consumer.endOffsets(Collections.singleton(partition)).values().iterator().next();
long lag = endOffset - currentOffset;
if (lag > threshold) {
System.out.printf("Alert: Consumer lag for partition %s is %d, which exceeds the threshold of %d%n", partition, lag, threshold);
}
}
}
}
}
以上代码基本上就能完全覆盖了相关kafka的性能优化，目前每秒的数据处理量是： 一万条左右，正常业务足够用了
