Kafka Producer之拦截器

1. Producer流程

新建ProducerRecord类后，传入topic、key、value等数据构建Record之后，隔断发送至kafka集群还必要经历若干过程。

• 拦截器列表，对数据进行过滤，更改等行为，处置惩罚异常不会导致流程停止。
  
• 获取Kafka集群元数据
  
• 对数据进行序列化
  
• 根据元数据选择分区和Broker
  
• 数据校验
  
• 进入数据发送缓存区，批次发送
  
• send
  

2. 代码测试

1. public class KafkaProducerInterceptorTest {
2.     public static void main(String[] args) throws InterruptedException {
3.         //创建producer
4.         HashMap<String, Object> config = new HashMap<>();
5.         config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:19092");
6.         config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
7.         config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
8.         //指定拦截器
9.         config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ValueInterceptorTest.class.getName());
10.         KafkaProducer<String, String> producer = new KafkaProducer<String, String>(config);
12.         for (int i = 0; i < 10; i++) {
13.             //创建record
14.             ProducerRecord<String, String> record = new ProducerRecord<String, String>(
15.                     "test1",
16.                     "key"+i,
17.                     "我是你爹"+i
18.             );
19.             //发送record
20.             producer.send(record);
21.             Thread.sleep(500);
22.         }
24.         //关闭producer
25.         producer.close();
26.     }
27. }

复制代码

拦截器自界说类

1. package org.dragon.producer;
3. import org.apache.kafka.clients.producer.ProducerInterceptor;
4. import org.apache.kafka.clients.producer.ProducerRecord;
5. import org.apache.kafka.clients.producer.RecordMetadata;
7. import java.util.Map;
9. /**
10. * 自定义value拦截器试验<br/>
11. * 1. 实现接口
12. * 2. 定义泛型
13. * 3. 重写方法
14. *
15. *
16. * @author mumu
17. * @date 2024/07/15
18. */
19. public class ValueInterceptorTest implements ProducerInterceptor<String, String> {
21.     /**
22.      * 发送数据时会调用这个方法<br/>
23.      * 让value复制2次
24.      *
25.      * @param producerRecord 生产者记录
26.      * @return {@link ProducerRecord}<{@link String}, {@link String}>
27.      */
28.     @Override
29.     public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
30.         return new ProducerRecord<String, String>(producerRecord.topic(), producerRecord.key(), producerRecord.value() + producerRecord.value());
31.     }
33.     /**
34.      * 发送数据完毕，服务器返回的响应，会调用此方法。
35.      *
36.      * @param recordMetadata 记录元数据
37.      * @param e              e
38.      */
39.     @Override
40.     public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
42.     }
44.     /**
45.      * 生产者关闭，会调用此方法
46.      */
47.     @Override
48.     public void close() {
50.     }
52.     /**
53.      * 创建生产者对象时调用
54.      *
55.      * @param map 地图
56.      */
57.     @Override
58.     public void configure(Map<String, ?> map) {
60.     }
61. }

复制代码

3. 测试结果

可以看到value是复制了2次，乐成。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。