Java中的Kafka拦截器详解
作者：dalianpai
Kafka拦截器
Producer 拦截器(interceptor)是在 Kafka 0.10 版本被引入的，主要用于实现 clients 端的定制化控制逻辑。
对于 producer 而言， interceptor 使得用户在消息发送前以及 producer 回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。
同时， producer 允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。
Intercetpor 的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：
(1) configure(configs) 获取配置信息和初始化数据时调用。
(2) onSend(ProducerRecord)： 该方法封装进 KafkaProducer.send 方法中，即它运行在用户主线程中。 Producer 确保在 消息被序列化以及计算分区前调用该方法。 用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区， 否则会影响目标分区的计算。
(3) onAcknowledgement(RecordMetadata, Exception)： 该方法会在消息从 RecordAccumulator 成功发送到 Kafka Broker 之后，或者在发送过程中失败时调用。 并且通常都是在 producer 回调逻辑触发之前。 onAcknowledgement 运行在producer 的 IO 线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢 producer 的消息发送效率。
(4) close： 关闭 interceptor，主要用于执行一些资源清理工作如前所述， interceptor 可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则 producer 将按照指定顺序调用它们，并仅仅是捕获每个 interceptor 可能抛出的异常记录到错误日志中而非在向上传递。这在使用过程中要特别留意。
代码演示
/**
* @author WGR
* @create 2021/8/9 -- 20:10
*/
public class CounterInterceptor implements ProducerInterceptor<String, String> {
private int errorCounter = 0;
private int successCounter = 0;
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
// 统计成功和失败的次数
if (exception == null) {
successCounter++;
} else {
errorCounter++;
}
}
@Override
public void close() {
// 保存结果
System.out.println("Successful sent: " + successCounter);
System.out.println("Failed sent: " + errorCounter);
}
@Override
public void configure(Map<String, ?> configs) {
}
}
/**
* @author WGR
* @create 2021/8/9 -- 20:07
*/
public class TimeInterceptor implements ProducerInterceptor<String, String> {
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
// 创建一个新的 record，把时间戳写入消息体的最前部
return new ProducerRecord(record.topic(),
record.partition(), record.timestamp(), record.key(),
System.currentTimeMillis() + "," + record.value().toString());
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
}
@Override
public void close() {
}
@Override
public void configure(Map<String, ?> configs) {
}
}把上面的拦截器加到producer 主程序
/**
* @author WGR
* @create 2021/8/9 -- 16:38
*/
public class ProducerFastStart {
//kafka集群地址
private static final String brokerList = "192.168.1.144:9092";
//主体名称
private static final String topic = "dalianpai";
public static void main(String[] args) {
Properties properties = new Properties();
//设置序列化器
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
//设置重试次数
properties.put(ProducerConfig.RETRIES_CONFIG,10);
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokerList);
// 2 构建拦截链
List<String> interceptors = new ArrayList<>();
interceptors.add(CounterInterceptor.class.getName());
interceptors.add(TimeInterceptor.class.getName());
properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
// 3 发送消息
for (int i = 0; i < 11; i++) {
ProducerRecord<String, String> record = new ProducerRecord<>(topic, "Kafka-demo-001", "hello, Kafka!"+i);
producer.send(record);
}
producer.close();
}
}消息结果
到此这篇关于Java中的Kafka拦截器详解的文章就介绍到这了,更多相关Kafka拦截器内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！