当kafka消费的数据滞后1000条时，打印告警信息

server/2024/11/17 18:03:31/

要在 Kafka
消费者中实现当数据滞后1000条时打印告警信息，你需要在消费循环中添加逻辑来检查当前消费者的偏移量与主题中的最新偏移量之间的差异。如果这个差异大于1000，就打印告警信息。以下是修改后的代码示例：

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; // 设置告警阈值@Overridepublic void run() {// 配置消费者属性Properties props = new Properties();props.put("bootstrap.servers", "4.15.18.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); // 异步处理消息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);}}}}
}