概述
ConsumerInterceptor是Kafka中的一个重要组件，它允许开发人员在Kafka消费者端拦截和修改消息的处理过程。ConsumerInterceptor可以用于实现各种功能，从消息监控到数据转换和错误处理，为开发人员提供了更大的灵活性和可定制性。
ConsumerInterceptor的主要作用是在消息被消费之前和之后对其进行拦截和处理。它可以用于以下几个方面：
监控：通过ConsumerInterceptor，可以在消息被消费之前和之后记录和监控消息的元数据，例如消息的偏移量、主题、分区等信息。这对于跟踪和分析消息流的健康状况以及性能优化非常有用。
转换：ConsumerInterceptor还可以用于对消息进行转换和修改。通过拦截消息并对其进行操作，可以在消费者端对消息进行格式转换、数据解析或者其他自定义处理。例如，你可以将消息从一种格式转换为另一种格式，或者对消息进行特定的业务处理。
错误处理：当消费者在处理消息时发生错误或异常情况时，ConsumerInterceptor可以捕获这些错误并采取适当的措施。你可以在拦截器中实现自定义的错误处理逻辑，例如记录错误日志、发送告警通知或者进行重试操作，从而提高应用程序的可靠性和容错性。
使用场景
使用场景方面，ConsumerInterceptor可以在多种情况下发挥作用，例如：
监控和统计：你可以使用ConsumerInterceptor来收集和记录消费者端的统计信息，例如消费速率、处理延迟等。这样可以帮助你监控应用程序的性能并进行性能优化。
数据转换：如果你需要将消息从一种格式转换为另一种格式，例如将JSON消息转换为Avro格式，你可以使用ConsumerInterceptor来实现这个转换过程。
数据验证：ConsumerInterceptor可以用于验证消息的有效性和完整性。你可以在拦截器中实现验证逻辑，例如检查消息的签名或者校验消息的结构，以确保只有符合要求的消息被消费。
错误处理和重试：当消费者在处理消息时遇到错误，例如数据库连接失败或者网络故障，你可以使用ConsumerInterceptor来捕获这些错误并采取适当的措施。你可以在拦截器中实现自定义的错误处理逻辑，例如记录错误日志、发送告警通知或者进行消息重试。
总之，ConsumerInterceptor为开发人员提供了在消费者端对消息进行拦截、处理和定制的能力。通过使用ConsumerInterceptor，你可以实现一系列功能，包括监控、数据转换和错误处理，从而更好地控制和管理Kafka消费者端的消息处理过程。
实战
配置文件
spring: kafka: bootstrap-servers: 20.10.110.137:9888 # Kafka服务的地址 producer: key-serializer: org.apache.kafka.common.serialization.StringSerializer # key的序列化器 value-serializer: org.apache.kafka.common.serialization.StringSerializer # value的序列化器 acks: 1 # acks=0：表示producer不需要等待任何确认收到的信息。副本将立即加到socket缓冲区并认为已经发送。如果使用此选项，则存在丢失数据的风险，因为服务器在数据到达副本之前可能会崩溃。 retries: 0 # 失败重试次数，0表示不启用重试机制 batch-size: 16384 # 发送缓冲区大小，按照字节计算 linger-ms: 1 # 发送延时，单位毫秒 buffer-memory: 33554432 # 发送缓存区的大小，按照字节计算 compression-type: gzip # 压缩类型，默认是none，可选snappy、gzip、lz4 consumer: #Kafka中没有初始偏移或如果当前偏移在服务器上不再存在时,默认区最新 ，有三个选项 【latest, earliest, none】 auto-offset-reset: earliest #是否开启自动提交 enable-auto-commit: false #key的解码方式 key-deserializer: org.apache.kafka.common.serialization.StringDeserializer #value的解码方式 value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #消费者组groupid group-id: process-group #消费者最大拉取的消息数量 max-poll-records: 2000 #消费者最大等待时间 max-poll-interval-ms: 2000 listener: type: batch ack-mode: manual # 手动提交 concurrency: 12 # 并发数
配置类
package net.zf.module.system.config; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerContainerFactory; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler; import org.springframework.kafka.listener.ContainerProperties; import java.util.HashMap; import java.util.Map; /** * @author artisan */ @Slf4j @Configuration public class KafkaConfig { @Value("${spring.kafka.bootstrap-servers}") private String bootstrapServer; @Value("${spring.kafka.consumer.auto-offset-reset}") private String autoOffsetReset; @Value("${spring.kafka.consumer.enable-auto-commit}") private String enableAutoCommit; @Value("${spring.kafka.consumer.key-deserializer}") private String keyDeserializer; @Value("${spring.kafka.consumer.value-deserializer}") private String valueDeserializer; @Value("${spring.kafka.consumer.group-id}") private String group_id; @Value("${spring.kafka.consumer.max-poll-records}") private String maxPollRecords; @Value("${spring.kafka.consumer.max-poll-interval-ms}") private String maxPollIntervalMs; @Value("${spring.kafka.listener.concurrency}") private Integer concurrency; private final String consumerInterceptor = "net.zf.module.system.kafka.interceptor.FailureRateInterceptor"; /** * 消费者配置信息 */ @Bean public Map<String, Object> consumerConfigs() { Map<String, Object> props = new HashMap<>(32); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,bootstrapServer); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset); props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,enableAutoCommit); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,keyDeserializer); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,valueDeserializer); props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG,maxPollRecords); props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG,maxPollIntervalMs); props.put(ConsumerConfig.GROUP_ID_CONFIG,group_id); props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG,consumerInterceptor ); return props; } /** * 消费者批量工厂 */ @Bean public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> batchFactory() { ConcurrentKafkaListenerContainerFactory<String,String> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(consumerConfigs())); factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); factory.setBatchListener(true); factory.setConcurrency(concurrency); return factory; } /** * 异常处理器 * * @return */ @Bean public ConsumerAwareListenerErrorHandler consumerAwareListenerErrorHandler() { return (message, exception, consumer) -> { // log.error("消息{} , 异常原因{}", message, exception.getMessage()); log.error("consumerAwareListenerErrorHandler called"); return null; }; } }
这段代码是一个用于配置Kafka消费者的Spring配置类。它使用了Spring Kafka库来设置Kafka的消费者配置和相关的监听器。
以下是代码的主要部分的解释：
通过@Configuration注解将该类标记为一个Spring配置类。
使用@Value注解注入配置属性，这些属性来自于应用的配置文件（比如application.properties）。
consumerConfigs()方法创建了一个包含Kafka消费者配置信息的props对象，并将其返回。这些配置包括Kafka服务器地址、消费者组ID、序列化/反序列化类等。
batchFactory()方法创建了一个ConcurrentKafkaListenerContainerFactory对象，并设置了相关的属性。它使用了前面定义的消费者配置，并设置了批量消费和并发处理的参数。
consumerAwareListenerErrorHandler()方法创建了一个ConsumerAwareListenerErrorHandler对象，用于处理消费过程中出现的异常。在这个例子中，它只是打印了错误日志。
总体而言，这段代码的目的是配置Kafka消费者的相关属性，包括连接到Kafka服务器的配置、消费者组ID、序列化/反序列化类等。它还定义了一个批量消费的监听器工厂和一个异常处理器。这些配置可以通过注入KafkaListenerContainerFactory和ConsumerAwareListenerErrorHandler来在应用中使用。
自定义ConSumerInterceptor
package net.zf.module.system.kafka.interceptor; import lombok.extern.slf4j.Slf4j; import org.apache.kafka.clients.consumer.ConsumerInterceptor; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.common.TopicPartition; import org.springframework.stereotype.Component; import java.util.Map; /** * @author artisan */ @Slf4j @Component public class FailureRateInterceptor implements ConsumerInterceptor<Object, Object> { /** * 消息消费前的拦截处理 * * @param consumerRecords * @return */ @Override public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> consumerRecords) { // TODO log.info("FailureRateInterceptor#onConsume"); // 根据设定的规则计算失败率，并进行判断是否跳过消息的消费 // 返回ConsumerRecords对象, 继续执行下游的消费逻辑或者直接返回空的ConsumerRecords对象 (ConsumerRecords.EMPTY) return consumerRecords; } /** * 消息提交前进行拦截处理 * * @param map */ @Override public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) { log.info("FailureRateInterceptor#onCommit"); } /** * 拦截器关闭前进行拦截处理(如果有的话) */ @Override public void close() { log.info("FailureRateInterceptor#close"); } /** * 初始化配置(如果有的话) * * @param map */ @Override public void configure(Map<String, ?> map) { log.info("FailureRateInterceptor#configure"); } }
onConsume 可以控制 ConsumerRecords， 通过返回null ，可以暂停消费。
这段代码是一个自定义的Kafka消费者拦截器，实现了ConsumerInterceptor接口。拦截器可以在消息消费和提交的过程中插入自定义的逻辑，用于处理消息或拦截操作。
以下是代码的主要部分的解释：
@Slf4j注解用于自动生成日志记录器。
@Component注解将该类标记为Spring组件，使得它可以被自动扫描和注入到应用中。
实现了ConsumerInterceptor接口，并重写了其中的方法。
onConsume()方法在消费者消费消息之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
onCommit()方法在消息提交之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
close()方法在拦截器关闭之前被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
configure()方法在拦截器初始化配置时被调用。在这个例子中，它只是打印了日志信息，表示拦截器的执行。
拦截器的具体逻辑还没有实现，而是用// TODO标记了需要填充的部分。根据注释的描述，它可能会根据设定的规则计算消费失败率，并根据判断跳过或继续消费消息。
总体而言，这段代码定义了一个自定义的Kafka消费者拦截器。拦截器可以在消息消费和提交的过程中执行自定义的逻辑。在这个例子中，拦截器的逻辑还没有实现，只是打印了日志信息以表示拦截器的执行。你需要根据需求实现onConsume()方法中的拦截逻辑，以便根据设定的规则处理消息消费的失败率。
使用
package net.zf.module.system.kafka.consumer; import lombok.extern.slf4j.Slf4j; import net.zf.module.system.entity.AttackMessage; import net.zf.module.system.executors.factory.MessageExecutorFactory; import net.zf.module.system.service.es.AttackMessageESService; import net.zf.module.system.util.constants.KafkaTopicConstant; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.kafka.support.Acknowledgment; import org.springframework.stereotype.Component; import java.util.ArrayList; import java.util.List; /** * @author artisan */ @Component @Slf4j public class AttackKafkaConsumer { @Autowired private MessageExecutorFactory messageExecutorFactory; @Autowired private AttackMessageESService attackMessageESService; @KafkaListener(topicPattern = KafkaTopicConstant.ATTACK_MESSAGE + ".*", containerFactory = "batchFactory", errorHandler = "consumerAwareListenerErrorHandler") public void processMessage(List<String> records, Acknowledgment ack) { log.info("AttackKafkaConsumer 当前线程 {} , 本次拉取的数据总量:{} ", Thread.currentThread().getId(), records.size()); try { List<AttackMessage> attackMessages = new ArrayList(); records.stream().forEach(record -> { messageExecutorFactory.process(KafkaTopicConstant.ATTACK_MESSAGE).execute(record, attackMessages); }); if (!attackMessages.isEmpty()) { String response = attackMessageESService.addDocuments(attackMessages, false); log.info("AttackKafkaConsumer本次处理的数据总量:{}, 响应结果: {}", attackMessages.size(), response); } } finally { ack.acknowledge(); } } }
这段代码定义了一个名为AttackKafkaConsumer的类，它是一个Kafka消费者。它使用了Spring Kafka提供的@KafkaListener注解来指定消费者的相关配置。
以下是代码的主要部分的解释：
@Component注解将该类标记为Spring组件，使得它可以被自动扫描和注入到应用中。
@Slf4j注解用于自动生成日志记录器。
AttackKafkaConsumer类中注入了MessageExecutorFactory和AttackMessageESService两个依赖，通过@Autowired注解实现自动注入。
@KafkaListener注解标记了processMessage()方法作为Kafka消费者的消息处理方法。
topicPattern属性指定了要监听的Kafka主题的模式，使用了常量KafkaTopicConstant.ATTACK_MESSAGE并结合通配符.*。
containerFactory属性指定了用于创建Kafka监听容器的工厂bean的名称，使用了名为batchFactory的工厂。
errorHandler属性指定了用于处理消费者异常的错误处理器的bean的名称，使用了名为consumerAwareListenerErrorHandler的错误处理器。
processMessage()方法是消息的实际处理逻辑。它接收一个List<String>类型的消息记录和一个Acknowledgment对象作为参数。
首先，它记录了当前线程ID和本次拉取的数据总量的日志信息。
然后，它创建了一个空的AttackMessage列表，用于存储处理后的消息。
使用records.stream().forEach()遍历每条消息记录，并通过messageExecutorFactory调用process()方法来处理每条记录，同时将处理结果添加到attackMessages列表中。
在处理完所有消息后，如果attackMessages列表不为空，将调用attackMessageESService的addDocuments()方法将消息添加到Elasticsearch中，并记录处理的数据总量和响应结果的日志信息。
最后，在finally块中调用ack.acknowledge()手动确认消费完成。
总体而言，这段代码定义了一个Kafka消费者类AttackKafkaConsumer，并使用@KafkaListener注解指定了监听的主题、容器工厂和错误处理器。processMessage()方法是处理消息的具体逻辑，它遍历消息记录并调用适当的执行器进行处理，最后将处理结果添加到列表中，并通过Elasticsearch服务将消息存储到数据库中。消费完成后，手动确认消息的消费。
