Kafka Producer拦截器(Interceptor)
2023-04-18 13:06:07 时间
Kafka拦截器一共有两种:
- Producer端
- Consumer端
本篇主要讲述的是Kafka Producer端的拦截器,它主要用来对消息进行拦截或者修改,也可以用于Producer的Callback回调之前进行相应的预处理。
使用Kafka Producer端的拦截器非常简单,主要是实现ProducerInterceptor接口,此接口包含4个方法:
- ProducerRecord onSend(ProducerRecord record) Producer在将消息序列化和分配分区之前会调用拦截器的这个方法来对消息进行相应的操作。一般来说最好不要修改消息ProducerRecord的topic、key以及partition等信息,如果要修改,也需确保对其有准确的判断,否则会与预想的效果出现偏差。比如修改key不仅会影响分区的计算,同样也会影响Broker端日志压缩(Log Compaction)的功能。
- void onAcknowledgement(RecordMetadata metadata, Exception exception) 在消息被应答(Acknowledgement)之前或者消息发送失败时调用,优先于用户设定的Callback之前执行。这个方法运行在Producer的IO线程中,所以这个方法里实现的代码逻辑越简单越好,否则会影响消息的发送速率。
- void close() 关闭当前的拦截器,此方法主要用于执行一些资源的清理工作。
- configure(Map configs) 用来初始化此类的方法,这个是ProducerInterceptor接口的父接口Configurable中的方法。
一般情况下只需要关注并实现onSend或onAcknowledgement方法即可。
示例
- onSend统计发送的消息个数
- 通过onAcknowledgement统计发送消息的成功次数
相关文章
- .net core的学习小结
- 【从零开始学爬虫】采集丁香医生新冠问答数据
- 计算机内网建设对GPS北斗卫星授时服务器的技术需求
- 用SGDK开发世嘉MD游戏:入门篇
- 2023最新nacos的windows 10安装(保姆级)
- 读编程与类型系统笔记06_函数类型的高级应用
- MQ——如何选择消息队列
- 区块链特辑——solidity语言基础(五)
- 都想来分一杯羹,低代码到底服务谁?
- Nginx 高级篇
- 辛苦加班了一年,为何年终奖比同级同事少?
- MegCC 新版本来啦!新增 Benchmark 等工具,性能大幅提升!
- 开源动物行为分析实验箱(斯金纳箱)特色解析及试用条款
- 虚拟串口及其在串口转以太网中的应用
- C#、TS和Dart对比4.1:类型概述
- TapTap 算法平台的 Serverless 探索之路
- 读编程与类型系统笔记05_函数类型
- C#、TS和Dart对比2:变量和作用域
- 洛谷P1040. 加分二叉树
- MQ——如何保证消息不会丢失