zoukankan      html  css  js  c++  java
  • 24.Kafka(二):API

    一、Producer API

    1.1 消息发送流程

    KafkaProducer发送消息采用的是异步发送的方式。在消息发送的过程中,涉及到了两个线程——main线程和sender线程,以及一个线程共享变量——RecordAccumulatormain线程将消息发送给RecordAccumulatorsender线程不断从RecordAccumulator中拉取消息发送到Kafka broker
    1
    相关参数:

    • batch.size:只有数据积累到batch.size之后,sender才会发送数据。
    • linger.ms:如果数据迟迟未达到batch.sizesender等待linger.time之后就会发送数据。

    1.2 异步发送API

    导入依赖

    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.11.0.0</version>
    </dependency>
    
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.25</version>
        <scope>compile</scope>
    </dependency>
    

    编写代码

    需要用到的类:

    • KafkaProducer:需要创建一个生产者对象,用来发送数据
    • ProducerConfig:获取所需的一系列配置参数
    • ProducerRecord:每条数据都要封装成一个ProducerRecord对象

    ①不带回调函数的API

    public class AsyncProducer {
    
    public static void main(String[] args) throws ExecutionException, InterruptedException {
    
         Properties props = new Properties();
         props.put("bootstrap.servers", "hadoop100:9092");//kafka集群,broker-list
         props.put("acks", "all");
         props.put("retries", 1);//重试次数
         props.put("batch.size", 16384);//批次大小
         props.put("linger.ms", 1);//等待时间
         props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
         Producer<String, String> producer = new KafkaProducer<>(props);
         for (int i = 0; i < 100; i++) {
             ProducerRecord<String, String> record = new ProducerRecord<>("first",
                     Integer.toString(i), Integer.toString(i));
             producer.send(record).get();
         }
         producer.close();
     }
     
    }
    

    ②带回调函数的API

    回调函数会在producer收到ack时调用,为异步调用,该方法有两个参数,分别是RecordMetadataException,如果Exceptionnull,说明消息发送成功,如果Exception不为null,说明消息发送失败。

    注意:消息发送失败会自动重试,不需要我们在回调函数中手动重试。

    public class CustomProducer {
    
        public static void main(String[] args) throws ExecutionException, InterruptedException {
          		
            for (int i = 0; i < 100; i++) {
                producer.send(new ProducerRecord<String, String>("first", Integer.toString(i),
                        //回调函数Callback会在Producer收到ack时异步调用
                        Integer.toString(i)), (metadata, exception) -> {
                            if (exception == null) {
                                System.out.println("success->" + metadata.offset());
                            } else {
                                exception.printStackTrace();
                            }
                        });
            }
            producer.close();
        }
        
    }
    

    1.3 同步发送API

    同步发送的意思就是,一条消息发送之后,会阻塞当前线程,直至返回ack

    由于send方法返回的是一个Future对象,根据Futrue对象的特点,我们也可以实现同步发送的效果,只需在调用Future对象的get方发即可。

    public class SyncProvider {
    
     public static void main(String[] args) throws ExecutionException, InterruptedException {
     
         Properties props = new Properties();
         props.put("bootstrap.servers", "hadoop100:9092");//kafka集群,broker-list
         props.put("acks", "all");
         props.put("retries", 1);//重试次数
         props.put("batch.size", 16384);//批次大小
         props.put("linger.ms", 1);//等待时间
         props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
         props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
         props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
         Producer<String, String> producer = new KafkaProducer<>(props);
         for (int i = 0; i < 100; i++) {
             ProducerRecord<String, String> record = new ProducerRecord<>("first",
                     Integer.toString(i), Integer.toString(i));
             producer.send(record).get();
         }
         producer.close();
     }
     
    }
    

    二、Consumer API

    Consumer消费数据时的可靠性是很容易保证的,因为数据在Kafka中是持久化的,故不用担心数据丢失问题。

    由于Consumer在消费过程中可能会出现断电宕机等故障,Consumer恢复后,需要从故障前的位置的继续消费,所以Consumer需要实时记录自己消费到了哪个offset,以便故障恢复后继续消费。

    所以offset的维护是Consumer消费数据是必须考虑的问题。

    2.1 手动提交offset

    ① 编写代码

    需要用到的类:

    • KafkaConsumer:需要创建一个消费者对象,用来消费数据
    • ConsumerConfig:获取所需的一系列配置参数
    • ConsuemrRecord:每条数据都要封装成一个ConsumerRecord对象
    public class HandConsumer {
    	public static void main(String[] args) {
    	 Properties props = new Properties();
    	 props.put("bootstrap.servers", "hadoop100:9092");
    	 props.put("group.id", "test");//消费者组,只要group.id相同,就属于同一个消费者组
    	 props.put("enable.auto.commit", false);//手动提交offset
    	 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    	 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    	 KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
    	 consumer.subscribe(Arrays.asList("first"));
    	 while (true) {
    	     ConsumerRecords<String, String> records = consumer.poll(100);
    	     for (ConsumerRecord record : records) {
    	         System.out.printf("offset =%d,key=%s,value=%s
    ", record.offset(),
    	                 record.key(), record.value());
    	     }
    	     consumer.commitSync();
    	 }
    	}
    }
    

    ②编写代码

    手动提交offset的方法有两种:分别是commitSync(同步提交)和commitAsync(异步提交)。两者的相同点是,都会将本次poll的一批数据最高的偏移量提交;不同点是,commitSync会失败重试,一直到提交成功(如果由于不可恢复原因导致,也会提交失败);而commitAsync则没有失败重试机制,故有可能提交失败。

    ③数据重复消费问题
    在这里插入图片描述
    已消费消息,但未提交对应的offset

    2.2 自动提交offset

    public class CustomConsumer {
    
       public static void main(String[] args) {
           Properties props = new Properties();
           props.put("bootstrap.servers", "hadoop102:9092");
           props.put("group.id", "test");
           props.put("enable.auto.commit", "true");
           props.put("auto.commit.interval.ms", "1000");
           props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
           props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
           KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
           consumer.subscribe(Arrays.asList("first"));
           while (true) {
               ConsumerRecords<String, String> records = consumer.poll(100);
               for (ConsumerRecord<String, String> record : records)
             System.out.printf("offset =%d,key=%s,value=%s
    ", record.offset(),
                     record.key(), record.value());
           }
       }
    }
    

    三、自定义Interceptor

    3.1 拦截器原理

    Producer拦截器(interceptor)是在Kafka 0.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, Exception)
      该方法会在消息从RecordAccumulator成功发送到Kafka Broker之后,或者在发送过程中失败时调用。并且通常都是在producer回调逻辑触发之前。onAcknowledgement运行在producerIO线程中,因此不要在该方法中放入很重的逻辑,否则会拖慢producer的消息发送效率。
    4. close
      关闭interceptor,主要用于执行一些资源清理工作

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

    3.2 拦截器案例

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

    案例实操:

    ①增加时间戳拦截器

    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());
        }
    
        @Override
        public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
    
        }
    
        @Override
        public void close() {
    
        }
    
        @Override
        public void configure(Map<String, ?> map) {
    
        }
    }
    

    ②统计发送消息成功和发送失败消息数,并在producer关闭时打印这两个计数器

    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 recordMetadata, Exception e) {
            //统计成功和失败的次数
            if (e == null) {
                successCounter++;
            } else {
                errorCounter++;
            }
        }
    
        @Override
        public void close() {
            System.out.println("success:" + successCounter);
            System.out.println("error:" + errorCounter);
        }
    
        @Override
        public void configure(Map<String, ?> map) {
    
        }
    }
    

    producer代码

    public class InterceptorProducer {
        public static void main(String[] args) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "hadoop100:9092");//kafka集群,broker-list
            props.put("acks", "all");
            props.put("retries", 1);//重试次数
            props.put("batch.size", 16384);//批次大小
            props.put("linger.ms", 1);//等待时间
            props.put("buffer.memory", 33554432);//RecordAccumulator缓冲区大小
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 2 构建拦截链
            List<String> interceptors = new ArrayList<>();
            interceptors.add("com.hucheng.kafka.interceptor.TimeInterceptor");
            interceptors.add("com.hucheng.kafka.interceptor.CounterInterceptor");
            props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
    
            String topic = "first";
            Producer<String, String> producer = new KafkaProducer<>(props);
    
            // 3 发送消息
            for (int i = 0; i < 10; i++) {
    
                ProducerRecord<String, String> record = new ProducerRecord<>(topic, "message" + i);
                producer.send(record);
            }
    
            // 4 一定要关闭producer,这样才会调用interceptor的close方法
            producer.close();
    
    
        }
    }
    
    

    ④启动测试
    在这里插入图片描述
    在这里插入图片描述

  • 相关阅读:
    常遇问题及解决
    cs231笔记1
    scikit-learn模型参数保存和多分类策略(one vs one和one vs rest)
    练习1_scikit_learn自带数据集_sklearn和svm
    记一次连不上wifi网的处理
    剑指offer | 从1到n整数中1出现的次数 | 22
    剑指offer | 数组中出现次数超过一半的数字 | 21
    剑指offer | 不用加减乘除做加法 | 20
    剑指offer | 二进制中1的个数 | 19
    剑指offer | 链表中环的入口结点 | 18
  • 原文地址:https://www.cnblogs.com/hucheng1997/p/13565871.html
Copyright © 2011-2022 走看看