Kafka producer拦截器--拦截器链

怼烎@ 2022-03-30 05:20 307阅读 0赞

##### 背景 #####

> producer 拦截器（interceptor）实在kafka0.10版本被引入的，主要实现客户端的定制化需求逻辑控制  
> producer允许多个用户指定多个interceptor按照顺序作用于同一条信息，从而形成一个拦截器链

##### 条件 #####

*  必须实现ProducerInterceptor 接口

public interface ProducerInterceptor<K, V> extends Configurable {
      public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record);
      public void onAcknowledgement(RecordMetadata metadata, Exception exception); 		
      public void close();
    }

###### 拦截器1 ######

package 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;
    
    /*
    kafka拦截器：
    1. 需要实现producerinterceoptor接口
    
    小需求：在每条数据后面加：时间戳
     */
    public class KafkaTimeInterceptor implements ProducerInterceptor<String, String> {
    
        /*
        onSend：在序列化前和分配分区之前进行拦截
         */
        @Override
        public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
            String value = record.value();//获取record的value
            String newValue = value+":"+System.currentTimeMillis();
            String topic = record.topic();
            //通过topic和处理后的newValue进行发送返回
            return new ProducerRecord<>(topic, newValue);
    
        }
    
        /*
        在消息被应答（Acknowledgement）之前或者消息发送失败时调用，优先于用户设定的Callback之前执行
        在执行callback之前进行调用
         */
        @Override
        public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    
        }
        /*
        关闭资源的时候进行调用
         */
        @Override
        public void close() {
    
        }
        /*
        获取或设置相关配置
        可以获取配置
        也可以对配置进行设置
         */
        @Override
        public void configure(Map<String, ?> configs) {
    
        }
    }

###### 拦截器2 ######

package 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 KafkaCountInterceptor implements ProducerInterceptor<String,String> {
        private long successCount=0;
        private long errorCount=0;
    
        @Override
        public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
            return record;
        }
        /*
        调用回调函数前，调用此方法，
         */
        @Override
        public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
            if(metadata!=null){
                successCount++;
            }else{
                errorCount++;
            }
        }
    
        @Override
        public void close() {
            System.out.println("success:"+successCount);
            System.out.println("error:"+errorCount);
        }
    
        @Override
        public void configure(Map<String, ?> configs) {
    
        }
    }

###### producer 生产者 ######

*  属性设置时，需要添加关于拦截器链的相关属性设置

package interceptor;
    
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    /*
    主类：
    */
    @SuppressWarnings("all")
    public class InterceptorProducer {
        public static void main(String[] args) {
            Properties props = new Properties();
            //设置相关属性
            //设置kafka集群
            props.put("bootstrap.servers", "hadoop102:9092");
            //设置应答级别
            props.put("acks", "all");
            //设置失败重试次数
            props.put("retries", 2);
            //设置批处理大小
            props.put("batch.size", 16384);
            //设置延时时间
            props.put("linger.ms", 1);
            //设置缓冲大小
            props.put("buffer.memory", 33554432);
            //设置key序列化
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            //设置value的序列化
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            //设置分区类设置--自定义分区，value设置为全类名
            //如果设置了自定义分区，而且record中没有指定分区，则自定义分区中的分区会覆盖掉key的影响
            //props.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.wangyg.NewMyPartition");
    
            //拦截链：多个拦截器构成的list
            props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,
                    Arrays.asList("interceptor.KafkaTimeInterceptor", "interceptor.KafkaCountInterceptor"));
    
            Producer<String, String> producer = new KafkaProducer<>(props);
            for (int i = 0; i < 100; i++) {
                // producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), Integer.toString(i)));
                ProducerRecord<String, String> record = new ProducerRecord<>("first", i + "", i + ":dilireba");
                producer.send(record, new Callback() {
                    //callback回调函数
                    @Override
                    public void onCompletion(RecordMetadata metadata, Exception exception) {
                        //不为空，获取相关元数据
                        if (metadata != null) {
                            long offset = metadata.offset(); //获取offset偏移量
                            int partition = metadata.partition();//获取分区
                            String topic = metadata.topic();//获取主题
                            System.out.println("topic:" + topic + " partition:" + partition + " offset" + offset);
                        }
                    }
                });
            }
            producer.close();
        }
    
    
    }

> 启动后，可以观察相关信息  
> 在idea中可以观察输出结果