Kafka Producer拦截器（Interceptor）

最新推荐文章于 2025-03-24 15:37:43 发布

JavaEdge聊AI

最新推荐文章于 2025-03-24 15:37:43 发布

阅读量1.1k

点赞数 1

分类专栏：消息队列

本文链接：https://blog.csdn.net/qq_33589510/article/details/118578793

版权

84 篇文章

订阅专栏

Kafka拦截器一共有两种：

本篇主要讲述的是Kafka Producer端的拦截器，它主要用来对消息进行拦截或者修改，也可以用于Producer的Callback回调之前进行相应的预处理。

使用Kafka Producer端的拦截器非常简单，主要是实现ProducerInterceptor接口，此接口包含4个方法：

ProducerRecord<K, V> onSend(ProducerRecord<K, V> record)
Producer在将消息序列化和分配分区之前会调用拦截器的这个方法来对消息进行相应的操作。一般来说最好不要修改消息ProducerRecord的topic、key以及partition等信息，如果要修改，也需确保对其有准确的判断，否则会与预想的效果出现偏差。比如修改key不仅会影响分区的计算，同样也会影响Broker端日志压缩（Log Compaction）的功能。
void onAcknowledgement(RecordMetadata metadata, Exception exception)
在消息被应答（Acknowledgement）之前或者消息发送失败时调用，优先于用户设定的Callback之前执行。这个方法运行在Producer的IO线程中，所以这个方法里实现的代码逻辑越简单越好，否则会影响消息的发送速率。
void close()
关闭当前的拦截器，此方法主要用于执行一些资源的清理工作。
configure(Map<String, ?> configs)
用来初始化此类的方法，这个是ProducerInterceptor接口的父接口Configurable中的方法。

一般情况下只需要关注并实现onSend或onAcknowledgement方法即可。