kafka-自定义拦截器（interceptor）

1.拦截器原理

　　producer拦截器（interceptor）是在kafka0.10版本被引入的，主要用于实现clients端的制定化控制逻辑；

　　对于producer而言，interceptor是的用户在消息发送前以及producer回调逻辑前有机会对消息做一些定制化需求，比如修改消息同时，producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链（interceptor chain），interceptor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：

　　1.configure（configs）:

　　　　获取配置信息和初始化数据时调用；

　　2.onSend（producerRecord）：

　　　　该方法封装进入kafkaProducer.send方法中，即它运行在用户主线程中，producer确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算；

　　3.onAcknowledgement（RecordMetadata,Excpetion）：

　　　　该方法在消息从RecordAccumnlator成功发送到kafka Broker之后，或者在发送过程中失败时调用，并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重要的逻辑，否则会拖慢producer的消息发送效率；

　　4.close：

　　　　关闭interceptor，主要用于执行一些资源清理的工作；

　　　　如前所述，interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个interceptor，则producer将按照执行顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非在向上传递，这在使用过程中要特别留意。

2.拦截器案例

　　需求：

　　　　实现一个简单的双interceptor组成的拦截链。第一个interceptor会在消息发送前将时间戳信息加到消息value的最前部；第二个interceptor会在消息发送后更新成功发送消息或失败发送消息数；

　　2.1 创建第一个interceptor

package com.wn.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class TimeInterceptor implements ProducerInterceptor<String,String> {

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        //取出数据value
        String value = producerRecord.value();
        //创建新的producer对象，并返回
        return new ProducerRecord<String,String>(producerRecord.topic(),producerRecord.partition(),
        producerRecord.key(),System.currentTimeMillis()+","+value);
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {

    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

　　2.2 创建第二个interceptor

package com.wn.interceptor;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Map;

public class CountInterceptor implements ProducerInterceptor<String,String> {
    int success;
    int error;

    @Override
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
        return producerRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
        if (recordMetadata!=null){
            success++;
        }else{
            error++;
        }
    }

    @Override
    public void close() {
        System.out.println("success:"+success);
        System.out.println("error:"+error);
    }

    @Override
    public void configure(Map<String, ?> map) {

    }
}

　　2.3 创建生产者（添加两个拦截器）

package com.wn.interceptor;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.ArrayList;
import java.util.Properties;

public class InterceptorProducer {
    public static void main(String[] args){
        //创建kafka生产者的配置信息
        Properties properties = new Properties();
        //kafka集群   ProducerConfig
        properties.put("bootstrap.servers","192.168.138.55:9092,192.168.138.66:9092,192.168.138.77:9092");
        //ack应答级别
        properties.put("acks","all");
        //重试次数
        properties.put("retries",3);
        //批次大小 16K
        properties.put("batch.size",16384);
        //等待时间
        properties.put("linger.ms",1);
        //RecordAccumulator缓冲区大小 32M
        properties.put("buffer.memory",33554432);
        //key,value序列化类
        properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
        //添加拦截器
        ArrayList<Object> interceptors = new ArrayList<>();
        interceptors.add("com.wn.interceptor.TimeInterceptor");
        interceptors.add("com.wn.interceptor.CountInterceptor");
        properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,interceptors);
        //创建生产者对象
        KafkaProducer<String,String> producer = new KafkaProducer<>(properties);
        //发送数据
        for (int i=0;i<5;i++){
            producer.send(new ProducerRecord<String, String>("aaa","aaabbb--"+i));
        }
        //关闭资源
        producer.close();

    }
}

　　2.4 开启zookeeper+kafka

　　2.5 创建消费者

bin/kafka-console-consumer.sh --zookeeper 192.168.138.55:2181,192.168.138.66:2181,192.168.138.77:2181 --topic aaa

　　2.6 启动生产者，查看效果