Kafka拦截器概述_kafka 拦截器场景

拦截器：允许应用在不修改逻辑的情况下，动态的实现一组可插拔的时间处理逻辑链；它被曝光贵阳在主业务操作的前后多个时间点插入对应的 “拦截逻辑”；
REQUEST --> INTERCEPTOR1 --> INTERCEPTOR2 --> BUSINESS CONTROLLER
–> INTERCEPTOR3 --> INTERCEPTOR4 --> RESPONSE
Kafka拦截器引入版本：0.10.0.0, 可以在消息发送前后动态植入不同的处理逻辑；

Kafka 拦截器

kafka拦截器支持链式编程，可以将一组拦截器串联成一个大的拦截器， Kafka会按照添加顺序依次执行拦截器逻辑；拦截器逻辑最好都是轻量级的；

生产者拦截器

生产者拦截器允许在发送消息前及消息提交成功后植入动态逻辑；
生产者拦截器配置参数：interceptor.classes ，指定为拦截器的全限定名；

Properties props = new Properties();
List<String> interceptors = new ArrayList<>();
interceptors.add("com.yourcompany.kafkaproject.interceptors.AddTimestampInterceptor"); // 拦截器 1
interceptors.add("com.yourcompany.kafkaproject.interceptors.UpdateCounterInterceptor"); // 拦截器 2
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
1
2
3
4
5

生产者拦截器实现

Interface: org.apache.kafka.clients.producer.ProducerInterceptor
method:
1. onSend：消息执行前被调用；
2. onAcknowledgement：消息提交成功或发送失败时被调用；该方法调用要早于消息发送回执callback; 该方法和onSend执行异步，如果有共享变量需要自行保证线程安全；

消费者拦截器

消费者拦截器支持在消费消息前以及提交为以后植入动态逻辑
消费者拦截器配置参数：interceptor.classes ， 同生产者

消费者拦截器实现

Interface: org.apache.kafka.clients.consumer.ConsumerInterceptor
method:
1. onConsume：消息返回给Consumer之前被调用；
2. onCommit：Consumer 在提交位移之后调用；

Kafka拦截器使用场景

• 客户端监控
• 端到端系统性能检测 ： 比如监控一条消息从生产到最后消费的端到端延时；
• 消息审计

实例化 – 延时监控

待补充…