kafka自定义拦截器通过实现producerinterceptor或consumerinterceptor接口并配置interceptor.classes实现消息过滤与监控。生产者端onsend()用于发送前过滤和打标，onacknowledgement()用于发送后监控；消费者端onconsume()用于消费前过滤，oncommit()用于提交后监控；需注意线程安全、阻塞操作规避及拦截器链执行逻辑。
Kafka 自定义拦截器实现消息过滤与监控，核心在于分别实现 ProducerInterceptor（生产者端）或 ConsumerInterceptor（消费者端）接口，并通过配置项 interceptor.classes 注册。过滤和监控可单独使用，也可组合部署，关键在于方法调用时机和线程安全设计。
生产者端：发送前过滤 + 发送后监控
适合在消息发出前剔除非法内容、统一打标，或统计成功率等指标。
-
onSend()：在序列化和分区计算前触发，可修改消息值（如添加前缀）、或直接丢弃不符合规则的消息（返回
null或抛异常） - onAcknowledgement()：消息确认后立即执行，适合记录成功/失败数、耗时、异常类型；注意该方法运行在 I/O 线程，避免重逻辑
- 配置方式：在
Properties或 YAML 中设置interceptor.classes，值为全限定类名列表，支持多个拦截器按序链式执行
消费者端：消费前过滤 + 提交后监控
适合按业务字段（如用户 ID、关键词、状态码）筛掉无需处理的消息，同时追踪消费进度和偏移提交行为。
-
onConsume()：在
poll()返回给业务代码前调用，可遍历ConsumerRecords，移除不满足条件的 record（例如 value 包含 “TEST” 或 timestamp 超过 24 小时） - onCommit()：位移提交完成后触发，适合记录提交时间、分区 offset 范围、失败重试次数等运维指标
- 注意：过滤后的 records 仍需保持原有结构（topic/partition/timestamp 不变），否则可能影响再平衡或重试逻辑
开发与配置要点
写拦截器时需兼顾功能性和稳定性：
- 实现
configure(Map<string>)</string>方法读取自定义参数（如过滤关键词列表、监控采样率） - 所有共享状态（如计数器）必须线程安全，推荐用
AtomicLong或ConcurrentHashMap - 避免在
onAcknowledgement和onCommit中做网络调用、文件写入等阻塞操作 - 拦截器链中任一环节抛出异常会被捕获并记录，但不会中断后续拦截器执行——需主动判断返回值或异常类型来控制流程
典型配置示例
以 Spring Kafka 为例，在 application.yml 中注册：
kafka:
producer:
properties:
interceptor.classes: com.example.kafka.interceptor.ContentFilterInterceptor,com.example.kafka.interceptor.MetricsInterceptor
consumer:
properties:
interceptor.classes: com.example.kafka.interceptor.KeywordFilterInterceptor
