【RAEDME】

本文中， java客户端作为生产者， centos中consumer线程作为消费者；

 

【1】拦截器简述

1）拦截器是什么？ 很明显，为了实现面向切面编码，即在 具体逻辑的上下文 添加一些逻辑；如

逻辑1
具体逻辑
逻辑2

2）什么时候调用拦截器？这就要从 kafka生产者发送数据说起了；

kafka生产者使用了2个线程来发送数据： 
step1）生产者中的main线程把数据经过 拦截器-》序列化器-》分区器 处理；然后再把数据写到 RecordAccumulator； 
step2）send 线程从 RecordAccumulator 中取出数据写入到broker list；

【2】拦截器实现

/* 添加拦截器 */
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(TimeInterceptor.class.getName(), CounterInterceptor.class.getName()));

1）需求

第1个拦截器， 在消息发送前将时间戳加到消息value的 最前面； 
第2个拦截器，在消息发送后更新成功发送消息数或失败发送消息数； 

2）代码实现

-- 添加拦截器

/* 添加拦截器 */
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(TimeInterceptor.class.getName(), CounterInterceptor.class.getName()));

-- 带有拦截器的生产者

/*** 带有拦截器的生产者*/ 
public class InterceptorProducer {public static void main(String[] args) {/* 1.创建kafka生产者的配置信息 */Properties props = new Properties();/*2.指定连接的kafka集群, broker-list */props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "centos201:9092");  /*3.ack应答级别*/props.put(ProducerConfig.ACKS_CONFIG, "all");/*4.重试次数*/ props.put(ProducerConfig.RETRIES_CONFIG, 3); /*5.批次大小，一次发送多少数据，当数据大于16k，生产者会发送数据到 kafka集群 */props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16 * KfkNumConst._1K);  /*6.等待时间， 等待时间超过1毫秒，即便数据没有大于16k， 也会写数据到kafka集群 */props.put(ProducerConfig.LINGER_MS_CONFIG, 1); /*7. RecordAccumulator 缓冲区大小*/ props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 32 * KfkNumConst._1M);  /*8. key, value 的序列化类 */ props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());/* 添加拦截器 */props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, Arrays.asList(TimeInterceptor.class.getName(), CounterInterceptor.class.getName())); System.out.println(props); /* 9.创建生产者对象 */KafkaProducer<String, String> producer = new KafkaProducer<>(props);  /* 10.发送数据 */ for (int i = 0; i < 10; i++) { producer.send(new ProducerRecord<>("first", "key"+i, "value-first-20210101--J" + i));}/* 11.关闭资源 */  producer.close(); // 间接调用了拦截器的close 方法  System.out.println("kafka生产者写入数据完成"); } 
}

/*** 时间拦截器-在消息前添加时间戳 */
public class TimeInterceptor implements ProducerInterceptor<String, String>{@Overridepublic void configure(Map<String, ?> configs) {}/*生产者确保在 消息被序列化以及计算分区前调用该方法。*/@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {// 获取消息值 String value = record.value();return new ProducerRecord<>(record.topic(), record.partition(), record.key(), System.currentTimeMillis() + "," + record.value()); }/*该方法会在消息从 RecordAccumulator 成功发送到Kafka Broker之后，或者在发送过程中失败时调用； 且通常在 生产者回调逻辑触发之前调用*/@Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {}@Overridepublic void close() {}
}

/*** 计数拦截器*/
public class CounterInterceptor implements ProducerInterceptor<String, String>{int sucCounter = 0;int errCounter = 0;@Overridepublic void configure(Map<String, ?> configs) {}/*生产者确保在 消息被序列化以及计算分区前调用该方法。*/@Overridepublic ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {return record; }/*该方法会在消息从 RecordAccumulator 成功发送到Kafka Broker之后，或者在发送过程中失败时调用； 且通常在 生产者回调逻辑触发之前调用*/@Overridepublic void onAcknowledgement(RecordMetadata metadata, Exception exception) {if(metadata !=null ) {sucCounter++;} else {errCounter++; }}@Overridepublic void close() {System.out.println("sucCounter =" + sucCounter + ", errCounter=" + errCounter);  }
}

-- java生产者打印日志 

sucCounter =10, errCounter=0
kafka生产者写入数据完成

3）centos消费者消费结果

[root@centos201 ~]# kafka-console-consumer.sh --topic first --zookeeper centos201:2181   
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper].
1609599630884,value-first-20210102--A0
1609599631203,value-first-20210102--A1
1609599631204,value-first-20210102--A6
1609599631203,value-first-20210102--A2
1609599631203,value-first-20210102--A3
1609599631204,value-first-20210102--A4
1609599631204,value-first-20210102--A5
1609599631204,value-first-20210102--A7
1609599631204,value-first-20210102--A8
1609599631204,value-first-20210102--A9