kafka

消费者Api示例

一个正常的消费逻辑需要具备以下几个步骤：

Javapackage com.doitedu;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.header.Header;import org.apache.kafka.common.header.Headers;import org.apache.kafka.common.record.TimestampType;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.Arrays;import java.util.Iterator;import java.util.Optional;import java.util.Properties;public class ConsumerDemo {    public static void main(String[] args) {        //1.创建kafka的消费者对象，附带着把配置文件搞定        Properties props = new Properties();        //props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");        //props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        //props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());        // 定义kakfa 服务的地址，不需要将所有broker指定上       // props.put("bootstrap.servers", "linux01:9092,linux02:9092,linux03:9092");        // 制定consumer group        props.put("group.id", "g3");        // 是否自动提交offset  __consumer_offset   有多少分区  50         props.put("enable.auto.commit", "true");        // 自动提交offset的时间间隔   -- 这玩意设置的大小怎么控制        props.put("auto.commit.interval.ms", "5000");  //50000   1000        // key的反序列化类        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        // value的反序列化类        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");        // 如果没有消费偏移量记录，则自动重设为起始offset：latest, earliest, none        props.put("auto.offset.reset","earliest");        KafkaConsumer consumer = new KafkaConsumer(props);        //2.订阅主题(确定需要消费哪一个或者多个主题)        consumer.subscribe(Arrays.asList("test02"));        //3.开始从topic中获取数据        while (true){            ConsumerRecords records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));            for (ConsumerRecord record : records) {                //这是数据所属的哪一个topic                String topic = record.topic();                //该条数据的偏移量                long offset = record.offset();                //这条数据是哪一个分区的                int partition = record.partition();                //这条数据记录的时间戳,但是这个时间戳有两个类型                long timestamp = record.timestamp();                //上面时间戳的类型，这个类型有两个，一个是CreateTime(这条数据创建的时间), LogAppendTime(这条数据往日志里面追加的时间)                TimestampType timestampType = record.timestampType();                //这个数据key的值                String key = record.key();                //这条数据value的值                String value = record.value();                //分区leader的纪元                Optional integer = record.leaderEpoch();                //key的长度                int keySize = record.serializedKeySize();                //value的长度                int valueSize = record.serializedValueSize();                //数据的头部信息                Headers headers = record.headers();//            for (Header header : headers) {//                String hKey = header.key();//                byte[] hValue = header.value();//                String valueString = new String(hValue);//                System.out.println("header的key值 = " + hKey + "header的value的值 = "+ valueString);//            }                System.out.printf("topic = %s ,offset = %d, partition = %d, timestampType = %s ,timestamp = %d , key = %s , value = %s ,leader的纪元 = %d , key序列化的长度 = %d ,value 序列化的长度 = %d \r\n" ,                        topic,offset,partition,timestampType + "",timestamp,key,value,integer.get(),keySize,valueSize);            }        }        //4.关闭消费者对象//        consumer.close();    }}

4.2.1subscribe订阅主题

subscribe有如下重载方法：

Javapublic void subscribe(Collection topics,ConsumerRebalanceListener listener) public void subscribe(Collection topics) public void subscribe(Pattern pattern, ConsumerRebalanceListener listener) public void subscribe(Pattern pattern)

Javaconsumer.subscribe(Arrays.asList(topicl ));

如果消费者采用的是正则表达式的方式（subscribe(Pattern)）订阅， 在之后的过程中，如果有人又创建了新的主题，并且主题名字与正表达式相匹配，那么这个消费者就可以消费到新添加的主题中的消息。如果应用程序需要消费多个主题，并且可以处理不同的类型，那么这种订阅方式就很有效。

正则表达式的方式订阅的示例

Javaconsumer.subscribe(Pattern.compile ("topic.*" ));

利用正则表达式订阅主题，可实现动态订阅

4.2.2assign订阅主题

消费者不仅可以通过 KafkaConsumer.subscribe() 方法订阅主题，还可直接订阅某些主题的指定分区；

 在 KafkaConsumer 中提供了 assign() 方法来实现这些功能，此方法的具体定义如下：

Javapackage com.doitedu;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.ConsumerRecord;import org.apache.kafka.clients.consumer.ConsumerRecords;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.header.Headers;import org.apache.kafka.common.record.TimestampType;import org.apache.kafka.common.serialization.StringDeserializer;import java.time.Duration;import java.util.Arrays;import java.util.Optional;import java.util.Properties;public class ConsumerDemo1 {     public static void main(String[] args) {         //1.创建kafka的消费者对象，附带着把配置文件搞定        Properties props = new Properties();         props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"linux01:9092,linux02:9092,linux03:9092");         props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"doit01");         KafkaConsumer consumer = new KafkaConsumer(props);         //2.订阅主题(确定需要消费哪一个或者多个主题)//        consumer.subscribe(Arrays.asList("test03"));//        consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));//        //我现在想手动指定，我需要从哪边开始消费//        //如果用subscribe去订阅主题的时候，他内部会给这个消费者组来一个自动再均衡//        consumer.seek(new TopicPartition("test03",0),2);        TopicPartition tp01 = new TopicPartition("test03", 0);         //他就是手动去订阅主题和partition，有了这个就不需要再去订阅subscribe主题了，手动指定以后，他的内部就不会再来自动均衡了        consumer.assign(Arrays.asList(tp01)); // 手动订阅指定主题的指定分区的指定位置         consumer.seek(new TopicPartition("test03",0),2);         //3.开始从topic中获取数据        while (true){             ConsumerRecords records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));             for (ConsumerRecord record : records) {                 //这是数据所属的哪一个topic                String topic = record.topic();                 //该条数据的偏移量                long offset = record.offset();                 //这条数据是哪一个分区的                int partition = record.partition();                 //这条数据记录的时间戳,但是这个时间戳有两个类型                long timestamp = record.timestamp();                 //上面时间戳的类型，这个类型有两个，一个是CreateTime(这条数据创建的时间), LogAppendTime(这条数据往日志里面追加的时间)                TimestampType timestampType = record.timestampType();                 //这个数据key的值                String key = record.key();                 //这条数据value的值                String value = record.value();                 //分区leader的纪元                Optional integer = record.leaderEpoch();                 //key的长度                int keySize = record.serializedKeySize();                 //value的长度                int valueSize = record.serializedValueSize();                 //数据的头部信息                Headers headers = record.headers();//            for (Header header : headers) {//                String hKey = header.key();//                byte[] hValue = header.value();//                String valueString = new String(hValue);//                System.out.println("header的key值 = " + hKey + "header的value的值 = "+ valueString);//            }                System.out.printf("topic = %s ,offset = %d, partition = %d, timestampType = %s ,timestamp = %d , key = %s , value = %s ,leader的纪元 = %d , key序列化的长度 = %d ,value 序列化的长度 = %d \r\n" ,                         topic,offset,partition,timestampType + "",timestamp,key,value,integer.get(),keySize,valueSize);             }         }         //4.关闭消费者对象//        consumer.close();    } }

这个方法只接受参数partitions，用来指定需要订阅的分区集合。示例如下：

Javaconsumer.assign(Arrays.asList(new TopicPartition ("tpc_1" , 0),new TopicPartition(“tpc_2”,1))) ;

4.2.3subscribe与assign的区别

在多个消费者的情况下可以根据分区分配策略来自动分配各个消费者与分区的关系。当消费组的消费者增加或减少时，分区分配关系会自动调整，以实现消费负载均衡及故障自动转移。

其实这一点从assign方法参数可以看出端倪，两种类型subscribe()都有ConsumerRebalanceListener类型参数的方法，而assign()方法却没有。

4.2.4取消订阅

既然有订阅，那么就有取消订阅；

可以使用KafkaConsumer中的unsubscribe()方法采取消主题的订阅，这个方法既可以取消通过 subscribe( Collection）方式实现的订阅；

也可以取消通过subscribe(Pattem）方式实现的订阅，还可以取消通过assign( Collection）方式实现的订阅。示例码如下：

Javaconsumer.unsubscribe();

如果将subscribe(Collection )或assign(Collection）集合参数设置为空集合，作用与unsubscribe（）方法相同，如下示例中三行代码的效果相同：

Javaconsumer.unsubscribe(); consumer.subscribe(new ArrayList()) ; consumer.assign(new ArrayList());

4.2.5消息的消费模式

Kafka中的消费是基于拉取模式的。

消息的消费一般有两种模式：推送模式和拉取模式。推模式是服务端主动将消息推送给消费者，而拉模式是消费者主动向服务端发起请求来拉取消息。

Javapublic class ConsumerRecord {     public static final long NO_TIMESTAMP = RecordBatch.NO_TIMESTAMP;     public static final int NULL_SIZE = -1;     public static final int NULL_CHECKSUM = -1;     private final String topic;     private final int partition;     private final long offset;     private final long timestamp;     private final TimestampType timestampType;     private final int serializedKeySize;     private final int serializedValueSize;     private final Headers headers;     private final K key;     private final V value;     private volatile Long checksum;

示例代码片段

Java/**  * 订阅与消费方式2  */ TopicPartition tp1 = new TopicPartition("x", 0); TopicPartition tp2 = new TopicPartition("y", 0); TopicPartition tp3 = new TopicPartition("z", 0); List tps = Arrays.asList(tp1, tp2, tp3); consumer.assign(tps); ​ while (true) {     ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));     for (TopicPartition tp : tps) {         List rList = records.records(tp);         for (ConsumerRecord r : rList) {             r.topic();             r.partition();             r.offset();             r.value();             //do something to process record.         }     } }

4.2.6指定位移消费

有些时候，我们需要一种更细粒度的掌控，可以让我们从特定的位移处开始拉取消息，而 KafkaConsumer 中的seek()方法正好提供了这个功能，让我们可以追前消费或回溯消费。

seek()方法的具体定义如下：

Javaseek都是和assign这个方法一起用 指定消费位置 public void seek(TopicPartiton partition,long offset)

代码示例：

Javapublic class ConsumerDemo3指定偏移量消费 {     public static void main(String[] args) {         Properties props = new Properties();         props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,"g002");         props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"doit01:9092");         props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());         props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"latest");         // 是否自动提交消费位移        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true");         // 限制一次poll拉取到的数据量的最大值        props.setProperty(ConsumerConfig.FETCH_MAX_BYTES_CONFIG,"10240000");          KafkaConsumer consumer = new KafkaConsumer(props);         // assign方式订阅doit27-1的两个分区        TopicPartition tp0 = new TopicPartition("doit27-1", 0);         TopicPartition tp1 = new TopicPartition("doit27-1", 1);                 consumer.assign(Arrays.asList(tp0,tp1));         // 指定分区0，从offset：800开始消费    ；  分区1，从offset：650开始消费        consumer.seek(tp0,200);         consumer.seek(tp1,250);         // 开始拉取消息         while(true){             ConsumerRecords poll = consumer.poll(Duration.ofMillis(3000));             for (ConsumerRecord rec : poll) {                 System.out.println(rec.partition()+","+rec.key()+","+rec.value()+","+rec.offset());             }         }     } }

4.2.7自动提交消费者偏移量

Kafka中默认的消费位移的提交方式是自动提交，这个由消费者客户端参数enable.auto.commit 配置，默认值为 true 。当然这个默认的自动提交不是每消费一条消息就提交一次，而是定期提交，这个定期的周期时间由客户端参数 auto.commit.interval.ms配置，默认值为5秒，此参数生效的前提是 enable. auto.commit 参数为 true。

在默认的方式下，消费者每隔5秒会将拉取到的每个分区中最大的消息位移进行提交。自动位移提交的动作是在 poll() 方法的逻辑里完成的，在每次真正向服务端发起拉取请求之前会检查是否可以进行位移提交，如果可以，那么就会提交上一次轮询的位移。

Kafka 消费的编程逻辑中位移提交是一大难点，自动提交消费位移的方式非常简便，它免去了复杂的位移提交逻辑，让编码更简洁。但随之而来的是重复消费和消息丢失的问题。

假设刚刚提交完一次消费位移，然后拉取一批消息进行消费，在下一次自动提交消费位移之前，消费者崩溃了，那么又得从上一次位移提交的地方重新开始消费，这样便发生了重复消费的现象（对于再均衡的情况同样适用）。我们可以通过减小位移提交的时间间隔来减小重复消息的窗口大小，但这样并不能避免重复消费的发送，而且也会使位移提交更加频繁。

按照一般思维逻辑而言，自动提交是延时提交，重复消费可以理解，那么消息丢失又是在什么情形下会发生的呢？我们来看下图中的情形：

拉取线程不断地拉取消息并存入本地缓存，比如在BlockingQueue 中，另一个处理线程从缓存中读取消息并进行相应的逻辑处理。设目前进行到了第 y+l 次拉取，以及第m次位移提交的时候，也就是 x+6 之前的位移己经确认提交了，处理线程却还正在处理x+3的消息；此时如果处理线程发生了异常，待其恢复之后会从第m次位移提交处，也就是 x+6 的位置开始拉取消息，那么 x+3至x+6 之间的消息就没有得到相应的处理，这样便发生消息丢失的现象。

4.2.8手动提交消费者偏移量（调用kafka api）

自动位移提交的方式在正常情况下不会发生消息丢失或重复消费的现象，但是在编程的世界里异常无可避免；同时，自动位移提交也无法做到精确的位移管理。在 Kafka中还提供了手动位移提交的方式，这样可以使得开发人员对消费位移的管理控制更加灵活。

很多时候并不是说拉取到消息就算消费完成，而是需要将消息写入数据库、写入本地缓存，或者是更加复杂的业务处理。在这些场景下，所有的业务处理完成才能认为消息被成功消费；

手动的提交方式可以让开发人员根据程序的逻辑在合适的时机进行位移提交。开启手动提交功能的前提是消费者客户端参数 enable.auto.commit 配置为false ，示例如下：

Javaprops.put(ConsumerConf.ENABLE_AUTO_COMMIT_CONFIG, false);

手动提交可以细分为同步提交和异步提交，对应于 KafkaConsumer 中的 commitSync()和

commitAsync()两种类型的方法。

commitSync()方法的定义如下：

Java/**  * 手动提交offset  */while (true) {     ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));     for (ConsumerRecord r : records) {         //do something to process record.    }     consumer.commitSync(); }

对于采用 commitSync()的无参方法，它提交消费位移的频率和拉取批次消息、处理批次消息的频率是一样的，如果想寻求更细粒度的、更精准的提交，那么就需要使用commitSync()的另一个有参方法，具体定义如下：

Javapublic void commitSync(final Map offsets)

示例代码如下：

Javawhile (true) {     ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));     for (ConsumerRecord r : records) {         long offset = r.offset();         //do something to process record.        TopicPartition topicPartition = new TopicPartition(r.topic(), r.partition());         consumer.commitSync(Collections.singletonMap(topicPartition,new OffsetAndMetadata(offset+1)));     } }

提交的偏移量  =  消费完的record的偏移量  +  1

因为，__consumer_offsets中记录的消费偏移量，代表的是，消费者下一次要读取的位置！！！

异步提交的方式（ commitAsync（））在执行的时候消费者线程不会被阻塞；可能在提交消费位移的结果还未返回之前就开始了新一次的拉取。异步提交可以让消费者的性能得到一定的增强。 commitAsync方法有一个不同的重载方法，具体定义如下

Java/**  * 异步提交offset  */while (true) {     ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));     for (ConsumerRecord r : records) {         long offset = r.offset();         //do something to process record.        TopicPartition topicPartition = new TopicPartition(r.topic(), r.partition());         consumer.commitSync(Collections.singletonMap(topicPartition,new OffsetAndMetadata(offset+1)));         consumer.commitAsync(Collections.singletonMap(topicPartition, new OffsetAndMetadata(offset + 1)), new OffsetCommitCallback() {      @Override      public void onComplete(Map map, Exception e) {                 if(e == null ){                     System.out.println(map);                 }else{                     System.out.println("error commit offset");                 }             }         });     } }

4.2.9手动提交位移（时机的选择）

可能会发生漏处理的现象（数据丢失）反过来说，这种方式实现了： at most once的数据处理（传递）语义

可能会发生重复处理的现象（数据重复）反过来说，这种方式实现了： at least once的数据处理（传递）语义当然，数据处理（传递）的理想语义是： exactly once（精确一次）Kafka也能做到exactly once（基于kafka的事务机制）

代码示例：

Javapackage com.doitedu;import org.apache.kafka.clients.consumer.*;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import java.sql.*;import java.time.Duration;import java.util.Arrays;import java.util.Collection;import java.util.Properties;public class CommitOffsetByMyself {     public static void main(String[] args) throws SQLException {         //获取mysql的连接对象        Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/football", "root", "123456");         connection.setAutoCommit(false);         PreparedStatement pps = connection.prepareStatement("insert into user values (?,?,?)");         PreparedStatement pps_offset = connection.prepareStatement("insert into offset values (?,?) on duplicate key update offset = ?");         Properties props = new Properties();         props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux01:9092,linux02:9092,linux03:9092");         props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());         //设置手动提交偏移量参数，需要将自动提交给关掉        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");         //设置从哪里开始消费//        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");        //设置组id        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group001");         KafkaConsumer consumer = new KafkaConsumer(props);         //订阅主题        consumer.subscribe(Arrays.asList("kafka2mysql"), new ConsumerRebalanceListener() {             @Override             public void onPartitionsRevoked(Collection collection) {             }             @Override             public void onPartitionsAssigned(Collection collection) {                 for (TopicPartition topicPartition : collection) {                     try {                         PreparedStatement get_offset = connection.prepareStatement("select offset from offset where topic_partition = ?");                         String topic = topicPartition.topic();                         int partition = topicPartition.partition();                         get_offset.setString(1, topic + "_" + partition);                         ResultSet resultSet = get_offset.executeQuery();                         if (resultSet.next()){                             int offset = resultSet.getInt(1);                             System.out.println("发生了再均衡，被分配了分区消费权，并且查到了目标分区的偏移量"+partition+" , "+offset);                             //拿到了offset后就可以定位消费了                            consumer.seek(new TopicPartition(topic, partition), offset);                         }                     } catch (SQLException e) {                         e.printStackTrace();                     }                 }             }         });         //拉去数据后写入到mysql        while (true) {             try {                 ConsumerRecords records = consumer.poll(Duration.ofMillis(Integer.MAX_VALUE));                 for (ConsumerRecord record : records) {                     String data = record.value();                     String[] arr = data.split(",");                     String id = arr[0];                     String name = arr[1];                     String age = arr[2];                    pps.setInt(1, Integer.parseInt(id));                    pps.setString(2, name);                    pps.setInt(3, Integer.parseInt(age));                    pps.execute();                    //埋个异常，看看是不是真的是这样//                    if (Integer.parseInt(id) == 5) {//                        throw new SQLException();//                    }                    long offset = record.offset();                    int partition = record.partition();                    String topic = record.topic();                    pps_offset.setString(1, topic + "_" + partition);                    pps_offset.setInt(2, (int) offset + 1);                    pps_offset.setInt(3, (int) offset + 1);                    pps_offset.execute();                    //提交jdbc事务                    connection.commit();                }             } catch (Exception e) {                 connection.rollback();             }         }     } }

4.2.10消费者提交偏移量方式的总结

consumer的消费位移提交方式：

4.2.11一些相对重要参数（了解）

Javafetch.min.bytes=1B   // 一次拉取的最小字节数 fetch.max.bytes=50M  //一次拉取的最大数据量   fetch.max.wait.ms=500ms  //拉取时的最大等待时长   max.partition.fetch.bytes = 1MB  //每个分区一次拉取的最大数据量   max.poll.records=500 //一次拉取的最大条数   connections.max.idle.ms=540000ms //网络连接的最大闲置时长   request.timeout.ms=30000ms  //一次请求等待响应的最大超时时间   metadata.max.age.ms=300000ms  //元数据在限定时间内没有进行更新，则会被强制更新   reconnect.backoff.ms=50ms  //尝试重新连接指定主机之前的退避时间  ==》 TODO   retry.backoff.ms=100ms  //尝试重新拉取数据的重试间隔    //事务的时候会出现这个isolation.level=read_uncommitted  //隔离级别！ 决定消费者能读到什么样的数据read_uncommitted：  //可以消费到LSO（LastStableOffset）位置；read_committed：  //可以消费到HW（High Watermark）位置   max.poll.interval.ms  //超过时限没有发起poll操作，则消费组认为该消费者已离开消费组   enable.auto.commit=true  //开启消费位移的自动提交 auto.offset.reset = latest   auto.commit.interval.ms=5000  //自动提交消费位移的时间间隔