一、拦截器介绍

Kafka Producer拦截器（Interceptor）主要用于实现clients端的定制化控制逻辑。对于Producer而言，Interceptor使得用户在消息发送前以及Producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，Producer允许用户指定多个Interceptor按序作用于同一条消息从而形成一个拦截链(Interceptor Chain)。Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法如下所示：

1、configure(configs)

获取配置信息和初始化数据时调用。

    @Overridepublic void configure(Map<String, ?> map) {}

2、onSend(ProducerRecord)

该方法封装进KafkaProducer.send方法中，即它运行在用户主线程中。Producer确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算。

    @Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {return null;}

3、onAcknowledgement(RecordMetadata, Exception)

该方法会在消息从RecordAccumulator成功发送到Kafka Broker之后，或者在发送过程中失败时调用。并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率。

    @Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {}

4、close

关闭interceptor，主要用于执行一些资源清理工作。如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非在向上传递。

    @Overridepublic void close() {}

二、拦截器案例

2.1 需求

实现一个由两个interceptor组成的拦截链。第一个interceptor会在消息发送前将时间戳信息加到消息value的最前部，第二个interceptor会在消息发送后更新成功发送消息数或失败发送消息数。

2.2 代码实现

1、首先实现添加时间戳的拦截器

package com.xzw.kafka.interceptor;import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.Map;/*** @author: xzw* @create_date: 2021/3/8 10:29* @desc: 添加时间戳的拦截器* @modifier:* @modified_date:* @desc:*/
public class TimeInterceptor implements ProducerInterceptor<String, String> {@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {return new ProducerRecord<String, String>(producerRecord.topic(), producerRecord.partition(),producerRecord.timestamp(), producerRecord.key(), System.currentTimeMillis() +producerRecord.value(),producerRecord.headers());}@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {}@Overridepublic void close() {}@Overridepublic void configure(Map<String, ?> map) {}
}

2、添加统计消息成功数和失败数的拦截器

package com.xzw.kafka.interceptor;import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;import java.util.Map;/*** @author: xzw* @create_date: 2021/3/8 11:11* @desc: 统计消息成功数和失败数的拦截器* @modifier:* @modified_date:* @desc:*/
public class CounterInterceptor implements ProducerInterceptor<String, String> {private long successNum = 0L;private long errorNum = 0L;@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {return producerRecord;}@Overridepublic void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {if (e == null) {successNum++;} else {errorNum++;}}@Overridepublic void close() {System.out.println("successNum = " + successNum + ";errorNum = " + errorNum);}@Overridepublic void configure(Map<String, ?> map) {}
}

3、定义生产者，添加拦截器配置信息

package com.xzw.kafka.producer;import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;/*** @author: xzw* @create_date: 2021/3/2 8:43* @desc: 添加了拦截器的异步发送* @modifier:* @modified_date:* @desc:*/
public class AsyncInterceptorProducer {public static void main(String[] args) {Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.ACKS_CONFIG, "all");props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);props.put(ProducerConfig.LINGER_MS_CONFIG, 1);//添加拦截器参数ArrayList<String> interceptors = new ArrayList<>();interceptors.add("com.xzw.kafka.interceptor.TimeInterceptor");interceptors.add("com.xzw.kafka.interceptor.CounterInterceptor");props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);//1、创建一个生产者对象KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);//2、调用生产者的send方法for (int i = 0; i < 10; i++) {producer.send(new ProducerRecord<String, String>("test", i + "", "data-" + i));}//3、关闭生产者producer.close();}
}

2.3 测试

启动一个消费者准备消费数据，运行定义的生产者可以发现，控制台打印出了成功的消息数和失败的消息数。

同时，消费者控制台也打印出了带有时间戳的数据：

本文到此已经接近尾声了，本文主要通过一个简单的例子说明了一下Kafka中的拦截器，内容比较简单。你们在此过程中遇到了什么问题，欢迎留言，让我看看你们都遇到了哪些问题~

四十七、Kafka中的拦截器（Interceptor）相关推荐

Springmvc中的拦截器interceptor及与过滤器filter的区别
一.Springmvc中的拦截器概述及与过滤器filter的区别 1).Springmvc中的拦截器interceptor用于对控制器controller进行预处理和后处理的技术; 2).可以定义拦截 ...
springboot中的拦截器interceptor和过滤器filter,多次获取request参数
大家好,我是烤鸭: 这是一篇关于springboot的拦截器(interceptor)和过滤器(Filter). 先说一下过滤器和拦截器. 区别: 1. servlet请求,顺序:Filter ...
谈谈spring中的拦截器interceptor
谈谈spring中的拦截器在web开发中,拦截器是经常用到的功能.它可以帮我们验证是否登陆.预先设置数据以及统计方法的执行效率等等.今天就来详细的谈一下spring中的拦截器.spring中拦截器主 ...
Hadoop生态Flume（四）拦截器(Interceptor)介绍与使用（2）
转载自 Flume中的拦截器(Interceptor)介绍与使用(二) lume中的拦截器(interceptor),用户Source读取events发送到Sink的时候,在events header ...
Kafka详解与总结(七)-Kafka producer拦截器(interceptor)
1. 拦截器原理 Producer拦截器(interceptor)是在Kafka 0.10版本被引入的,主要用于实现clients端的定制化控制逻辑. 对于producer而言,interceptor ...
JavaWeb中监听器Listener+过滤器filter+拦截器interceptor区别
JavaWeb中监听器Listener+过滤器filter+拦截器interceptor区别如果从整个项目中看,一个servlet请求的执行过程就变成了这样context-param–>lis ...
在SpringMVC中使用拦截器（interceptor）拦截CSRF***
关于什么是CSRF我这里就不多说了,以前转载的一篇文章(PS:https://www.zifangsky.cn/358.html)已经说得很清楚了.这里只是简单介绍如何在SpringMVC中使用拦截器 ...
Spring中过滤器(Filter)和拦截器(Interceptor)的区别和联系
在我们日常的开发中,我们经常会用到Filter和Interceptor.有时同一个功能.Filter可以做,Interceptor也可以做.有时就需要考虑使用哪一个比较好.这篇文章主要介绍一下,二者的 ...
(十六)ATP应用测试平台——java应用中的过滤器Filter、拦截器Interceptor、参数解析器Resolver、Aop切面，你会了吗？
前言过滤器Filter.拦截器Interceptor.参数解析器Resolver.Aop切面是我们应用开发中经常使用到的技术,到底该如何使用这些web附属功能, 本小节我们就分别介绍一下其各自的用法 ...

四十七、Kafka中的拦截器（Interceptor）