Kafka Producer拦截器（Interceptor）

Posted 2021-08-06 JavaEdge.

Kafka拦截器一共有两种：

• Producer端
• Consumer端

本篇主要讲述的是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方法即可。

示例

• onSend统计发送的消息个数

• 通过onAcknowledgement统计发送消息的成功次数