四十七、Kafka中的拦截器(Interceptor)

前两篇文章我们分别介绍了Kafka生产者和消费者的API,本文我们介绍一下Kafka中拦截器的知识。关注专栏《破茧成蝶——大数据篇》,查看更多相关的内容~


目录

一、拦截器介绍

二、拦截器案例

2.1 需求

2.2 代码实现

2.3 测试


一、拦截器介绍

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

1、configure(configs)

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

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

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

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

4、close

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

    @Override     public 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> {     @Override     public 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());     }      @Override     public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {      }      @Override     public void close() {      }      @Override     public 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;      @Override     public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {         return producerRecord;     }      @Override     public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {         if (e == null) {             successNum++;         } else {             errorNum++;         }     }      @Override     public void close() {         System.out.println("successNum = " + successNum + ";errorNum = " + errorNum);     }      @Override     public 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中的拦截器(Interceptor)

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

四十七、Kafka中的拦截器(Interceptor)

 

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

版权声明:玥玥 发表于 2021-03-11 12:42:16。
转载请注明:四十七、Kafka中的拦截器(Interceptor) | 女黑客导航