Apache Kafka - ConsumerInterceptor 实战（2）
Apache Kafka - ConsumerInterceptor 实战（2）
小小工匠
发布于 2023-05-27 14:32:41
发布于 2023-05-27 14:32:41
在这里插入图片描述
在这里插入图片描述
Pre
Apache Kafka - ConsumerInterceptor 实战 (1) 用代码的方式实现了ConsumerInterceptor , 接下来我们用 配置的方式来实现一下 。
思路
如何找配置类 KafkaProperties
在这里插入图片描述
有些属性是很明显的有的，其他没有的一般都在 Map里
在这里插入图片描述
那map的 key value 从哪里找呢？
找原生的配置 Kafka Consumer的 都在 ConsumerConfig
在这里插入图片描述
找到
public static final String INTERCEPTOR_CLASSES_CONFIG = "interceptor.classes";
在这里插入图片描述
OK，继续
示例
配置文件
在这里插入图片描述
自定义 拦截器
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对象
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");
}
}使用
在这里插入图片描述
测试
启动服务，发送消息，进行消费
在这里插入图片描述
小结
在Spring Boot中配置Kafka消费者的拦截器需要进行以下步骤：
- 首先，创建一个拦截器类，实现Kafka的
ConsumerInterceptor接口，定义拦截器的逻辑。 - 在应用的配置文件（例如
application.properties或application.yml）中，添加拦截器相关的配置项，其中包括设置interceptor.class属性为拦截器类的全限定名。
下面是一个示例，演示如何在Spring Boot中配置Kafka消费者的拦截器：
- 创建拦截器类：
@Slf4j
@Component
public class MyConsumerInterceptor implements ConsumerInterceptor<Object, Object> {
@Override
public ConsumerRecords<Object, Object> onConsume(ConsumerRecords<Object, Object> records) {
// 在消息消费前的处理逻辑
// ...
return records;
}
@Override
public void onCommit(Map<TopicPartition, OffsetAndMetadata> offsets) {
// 在消息提交前的处理逻辑
// ...
}
@Override
public void close() {
// 拦截器关闭前的处理逻辑
// ...
}
@Override
public void configure(Map<String, ?> configs) {
// 初始化配置的处理逻辑
// ...
}
}- 在应用的配置文件中设置拦截器相关的配置项：
spring.kafka.consumer.properties.interceptor.classes=com.example.MyConsumerInterceptor或者在application.yml文件中：
spring:
kafka:
consumer:
properties:
interceptor.classes: com.example.MyConsumerInterceptor这样配置之后，Spring Boot会自动创建Kafka消费者，并将指定的拦截器应用于消费者。在消费者处理消息的过程中，拦截器的方法将会被调用，可以在这些方法中编写自定义的逻辑来处理消息或拦截操作。
本文参与 腾讯云自媒体分享计划，分享自作者个人站点/博客。
原始发表：2023-05-25，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录