Kafka中的拦截器(Interceptor)是0.10.x.x版本引入的一个功能,一共有两种:Kafka Producer端的拦截器和Kafka Consumer端的拦截器。本篇主要讲述的是Kafka Producer端的拦截器,它主要用来对消息进行拦截或者修改,也可以用于Producer的Callback回调之前进行相应的预处理。
使用Kafka Producer端的拦截器非常简单,主要是实现ProducerInterceptor接口,此接口包含4个方法:
一般情况下只需要关注并实现onSend或者onAcknowledgement方法即可。下面我们来举个案例,通过onSend方法来过滤消息体为空的消息以及通过onAcknowledgement方法来计算发送消息的成功率。
public class ProducerInterceptorDemo implements ProducerInterceptor<String,String> {
private volatile long sendSuccess = 0;
private volatile long sendFailure = 0;
@Override
public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
if(record.value().length()<=0)
return null;
return record;
}
@Override
public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
if (exception == null) {
sendSuccess++;
} else {
sendFailure ++;
}
}
@Override
public void close() {
double succe***atio = (double)sendSuccess / (sendFailure + sendSuccess);
System.out.println("[INFO] 发送成功率="+String.format("%f", succe***atio * 100)+"%");
}
@Override
public void configure(Map<String, ?> configs) {}
}
自定义的ProducerInterceptorDemo类实现之后就可以在Kafka Producer的主程序中指定,示例代码如下:
public class ProducerMain {
public static final String brokerList = "localhost:9092";
public static final String topic = "hidden-topic";
public static void main(String[] args) throws ExecutionException, InterruptedException {
Properties properties = new Properties();
properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
properties.put("bootstrap.servers", brokerList);
properties.put("interceptor.classes", "com.hidden.producer.ProducerInterceptorDemo");
Producer<String, String> producer = new KafkaProducer<String, String>(properties);
for(int i=0;i<100;i++) {
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic, "msg-" + i);
producer.send(producerRecord).get();
}
producer.close();
}
}
Kafka Producer不仅可以指定一个拦截器,还可以指定多个拦截器以形成拦截链,这个拦截链会按照其中的拦截器的加入顺序一一执行。比如上面的程序多添加一个拦截器,示例如下:
properties.put("interceptor.classes", "com.hidden.producer.ProducerInterceptorDemo,com.hidden.producer.ProducerInterceptorDemoPlus");1
这样Kafka Producer会先执行拦截器ProducerInterceptorDemo,之后再执行ProducerInterceptorDemoPlus。
有关interceptor.classes参数,在kafka 1.0.0版本中的定义如下:
NAME | DESCRIPTION | TYPE | DEFAULT | VALID VALUES | IMPORTANCE |
---|---|---|---|---|---|
interceptor.calssses | A list of classes to use as interceptors. Implementing the org.apache.kafka.clients.producer.ProducerInterceptor interface allows you to intercept (and possibly mutate) the records received by the producer before they are published to the Kafka cluster. By default, there no interceptors. | list | null | low |
本文的重点是你有没有收获与成长,其余的都不重要,希望读者们能谨记这一点。同时我经过多年的收藏目前也算收集到了一套完整的学习资料,包括但不限于:分布式架构、高可扩展、高性能、高并发、Jvm性能调优、Spring,MyBatis,Nginx源码分析,Redis,ActiveMQ、、Mycat、Netty、Kafka、Mysql、Zookeeper、Tomcat、Docker、Dubbo、Nginx等多个知识点高级进阶干货,希望对想成为架构师的朋友有一定的参考和帮助
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。