kafka API详解

本是古典何须时尚 2022-01-07 02:47 510阅读 0赞

> ## 生产者api ##

Producer是Kafka三大组件中的一个，用于发送消息到kafka集群中

Producer提供了丰富的配置（见后面的配置项）用于控制它的行为

在编码之前先使用命令创建topic

./kafka-topics.sh --create --zookeeper hadoop01,hadoop02,hadoop03:2181 --partitions 2 --replication-factor 3 --topic mytopic1

## 简单模式 ##

Producer采用默认分区方式将消息散列的发送到各个分区当中

package com.jv;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class MyProducer {
        public static void main(String[] args) throws InterruptedException {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //ack模式，all是最慢但最安全的
            props.put("acks", "-1");
            //失败重试次数
            props.put("retries", 0);
            //每个分区未发送消息总字节大小（单位：字节），超过设置的值就会提交数据到服务端
            props.put("batch.size", 10);
            //props.put("max.request.size",10);
            //消息在缓冲区保留的时间，超过设置的值就会被提交到服务端
            props.put("linger.ms", 10000);
            //整个Producer用到总内存的大小，如果缓冲区满了会提交数据到服务端
            //buffer.memory要大于batch.size，否则会报申请内存不足的错误
            props.put("buffer.memory", 10240);
            //序列化器
            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++)
                producer.send(new ProducerRecord<String, String>("mytopic1", Integer.toString(i), "dd:"+i));
            //Thread.sleep(1000000);
            producer.close();
        }
    }

配置项

<table> 
 <tbody> 
  <tr> 
   <th>名称</th> 
   <th>说明</th> 
   <th>默认值</th> 
   <th>有效值</th> 
   <th>重要性</th> 
  </tr> 
 </tbody> 
 <tbody> 
  <tr> 
   <td>bootstrap.servers</td> 
   <td>kafka集群的broker-list，如：&lt;br&gt;hadoop01:9092,hadoop02:9092</td> 
   <td>无</td> 
   <td>&nbsp;</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>acks</td> 
   <td>确保生产者可靠性设置，有三个选项：&lt;br&gt;acks=0:不等待成功返回&lt;br&gt;acks=1:等Leader写成功返回&lt;br&gt;acks=all:等Leader和所有ISR中的Follower写成功返回,all也可以用-1代替</td> 
   <td>-1</td> 
   <td>0,1,-1,all</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>key.serializer</td> 
   <td>key的序列化器</td> 
   <td>&nbsp;</td> 
   <td>ByteArraySerializer&lt;br&gt;StringSerializer</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>value.serializer</td> 
   <td>value的序列化器</td> 
   <td>&nbsp;</td> 
   <td>ByteArraySerializer&lt;br&gt;StringSerializer</td> 
   <td>必选</td> 
  </tr> 
  <tr> 
   <td>buffer.memory</td> 
   <td>Producer总体内存大小</td> 
   <td>33554432</td> 
   <td>不要超过物理内存，根据实际情况调整</td> 
   <td>建议必选</td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>压缩类型&lt;br&gt;压缩最好用于批量处理，批量处理消息越多，压缩性能越好</td> 
   <td>无</td> 
   <td>none、gzip、snappy</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>retries</td> 
   <td>发送失败尝试重发次数</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>batch.size</td> 
   <td>每个partition的未发送消息大小</td> 
   <td>16384</td> 
   <td>根据实际情况调整</td> 
   <td>建议必选</td> 
  </tr> 
  <tr> 
   <td>client.id</td> 
   <td>附着在每个请求的后面，用于标识请求是从什么地方发送过来的</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>connections.max&lt;br&gt;.idle.ms</td> 
   <td>连接空闲时间超过过久自动关闭（单位毫秒）</td> 
   <td>540000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>linger.ms</td> 
   <td>数据在缓冲区中保留的时长,0表示立即发送&lt;br&gt;为了减少网络耗时，需要设置这个值&lt;br&gt;太大可能容易导致缓冲区满，阻塞消费者&lt;br&gt;太小容易频繁请求服务端</td> 
   <td>0</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.block.ms</td> 
   <td>最大阻塞时长</td> 
   <td>60000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.request.size</td> 
   <td>请求的最大字节数，该值要比batch.size大&lt;br&gt;不建议去更改这个值，如果设置不好会导致程序不报错，但消息又没有发送成功</td> 
   <td>1048576</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>partitioner.class</td> 
   <td>分区类，可以自定义分区类，实现partitioner接口</td> 
   <td>默认是哈希值%partitions</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>receive.buffer.bytes</td> 
   <td>socket的接收缓存空间大小,当阅读数据时使用</td> 
   <td>32768</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>request.timeout.ms</td> 
   <td>等待请求响应的最大时间,超时则重发请求,超过重试次数将抛异常</td> 
   <td>3000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>send.buffer.bytes</td> 
   <td>发送数据时的缓存空间大小</td> 
   <td>131072</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>timeout.ms</td> 
   <td>控制server等待来自followers的确认的最大时间</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>max.in.flight.&lt;br&gt;requests.per.&lt;br&gt;connection</td> 
   <td>kafka可以在一个connection中发送多个请求，叫作一个flight,这样可以减少开销，但是如果产生错误，可能会造成数据的发送顺序改变。</td> 
   <td>5</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metadata.fetch&lt;br&gt;.timeout.ms</td> 
   <td>从ZK中获取元数据超时时间&lt;br&gt;比如topic\host\partitions</td> 
   <td>60000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metadata.max.age.ms</td> 
   <td>即使没有任何partition leader 改变，强制更新metadata的时间间隔</td> 
   <td>300000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metric.reporters</td> 
   <td>类的列表，用于衡量指标。实现MetricReporter接口，将允许增加一些类，这些类在新的衡量指标产生时就会改变。JmxReporter总会包含用于注册JMX统计</td> 
   <td>none</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metrics.num.samples</td> 
   <td>用于维护metrics的样本数</td> 
   <td>2</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>metrics.sample.window.ms</td> 
   <td>metrics系统维护可配置的样本数量，在一个可修正的window size。这项配置配置了窗口大小，例如。我们可能在30s的期间维护两个样本。当一个窗口推出后，我们会擦除并重写最老的窗口</td> 
   <td>30000</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>reconnect.backoff.ms</td> 
   <td>连接失败时，当我们重新连接时的等待时间。这避免了客户端反复重连</td> 
   <td>10</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
  <tr> 
   <td>retry.backoff.ms</td> 
   <td>在试图重试失败的produce请求之前的等待时间。避免陷入发送-失败的死循环中</td> 
   <td>100</td> 
   <td>&nbsp;</td> 
   <td>&nbsp;</td> 
  </tr> 
 </tbody> 
</table>

更全的配置参考官方文档：[http://kafka.apache.org/documentation/\#producerconfigs][http_kafka.apache.org_documentation_producerconfigs]

## 幂等模式 ##

幂等性：客户端一次或多次操作，最终数据是一致的，比如购买火车票支付时可能显示网络异常，但其实已经扣款成功，用户再次发起扣款不会再触发真正的扣款Kafka只能保证在一个会话中的幂等性

**幂等模式只需要将enable.idempotence设置为true**，一旦设置了该属性，那么retries默认是Integer.MAX\_VALUE ，acks默认是all。代码的写法和前面例子没什么区别

## 事务模式 ##

事务模式要求数据发送必须包含在事务中，在事务中可以向多个topic发送数据，消费者端最好也使用事务模式读，保证一次能将整个事务的数据全部读取过来。当然消费者也可以不设置为事务读的模式。

@Test
        public void transactional(){
            Properties props = new Properties();
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            props.put("transactional.id", "my_transactional_id");
            Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer());
    
            producer.initTransactions();
    
            try {
                //数据发送必须在beginTransaction()和commitTransaction()中间，否则会报状态不对的异常
                producer.beginTransaction();
                for (int i = 0; i < 100; i++)
                    producer.send(new ProducerRecord<>("mytopic1", Integer.toString(i), Integer.toString(i)));
                producer.commitTransaction();
            } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
                // 这些异常不能被恢复，因此必须要关闭并退出Producer
                producer.close();
            } catch (KafkaException e) {
                // 出现其它异常，终止事务
                producer.abortTransaction();
            }
            producer.close();
        }

## 自定义分区类(Partitioner) ##

没什么特殊分区逻辑，把kafka默认的分区类拿出来读一下

package org.apache.kafka.clients.producer.internals;
    
    import java.util.List;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.ConcurrentMap;
    import java.util.concurrent.ThreadLocalRandom;
    import java.util.concurrent.atomic.AtomicInteger;
    import org.apache.kafka.clients.producer.Partitioner;
    import org.apache.kafka.common.Cluster;
    import org.apache.kafka.common.PartitionInfo;
    import org.apache.kafka.common.utils.Utils;
    
    public class DefaultPartitioner implements Partitioner {
        private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap();
        //需要覆盖的方法
        public DefaultPartitioner() {
        }
        //需要覆盖的方法，可以在这里添加配置信息
        public void configure(Map<String, ?> configs) {
        }
        //需要覆盖的方法，最重要的
        /*
        topic:主题
        key:动态绑定的，传的什么类型就是什么类型
        keyBytes:Ascii码数组
        value:动态绑定的，传的什么类型就是什么类型
        valueBytes:Ascii码数组
        cluster:kafka集群
        */
        public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
            //拿到所有分区
            List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);
            //分区数量
            int numPartitions = partitions.size();
            //如果key为空，则取消息作为分区依据
            if (keyBytes == null) {
                int nextValue = this.nextValue(topic);
                //可用分区，我在想应该是
                List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
                //可用分区数
                if (availablePartitions.size() > 0) {
                    //计算分区索引
                    int part = Utils.toPositive(nextValue) % availablePartitions.size();
                    //返回分区
                    return ((PartitionInfo)availablePartitions.get(part)).partition();
                } else {
                    //如果可用分区=0，则直接返回所有分区中的一个
                    return Utils.toPositive(nextValue) % numPartitions;
                }
            } else {
                //key有值，则返回所有分区中的一个
                return Utils.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
            }
        }
        //如果没有key，则调用该方法那消息来做分区依据
        private int nextValue(String topic) {
            AtomicInteger counter = (AtomicInteger)this.topicCounterMap.get(topic);
            if (null == counter) {
                counter = new AtomicInteger(ThreadLocalRandom.current().nextInt());
                AtomicInteger currentCounter = (AtomicInteger)this.topicCounterMap.putIfAbsent(topic, counter);
                if (currentCounter != null) {
                    counter = currentCounter;
                }
            }
    
            return counter.getAndIncrement();
        }
        //需要覆盖的方法
        public void close() {
        }
    }

> ## 消费者api ##

kafka的消费者API提供从kafka服务端拉取消息的能力，kafka引入了消费者组的概念，不同消费者组之间互不影响，独自拥有一份数据，而同一个消费者组内的消费者则有如下规律：

**分区数=消费者数：一个消费者拉取一个分区的数据**

**分区数>消费者数：同一个消费者可能拉取不同分区的数据**

**分区数<消费者数：一个消费者拉取一个分区的数据，多余的消费者不参与工作，当正在工作的消费者挂了之 后，这些闲着的消费者会顶替它干活，但会出现重复消费数据的情况**

## 偏移量由Kafka管理 ##

所有提交的offset都在kafka内建的一个消息队列中存在的，有50个分区，可以使用如下命令查看

查看所有topic

./kafka-topics.sh --zookeeper hadoop01:2181 --list

查看某个消费者组订阅的topic的当前offset和滞后进度

./kafka-consumer-groups.sh --bootstrap-server hadoop01:9092 --describe --group my\_group

### 1.偏移量-自动提交 ###

/*
            消费者拉取数据之后自动提交偏移量，不关心后续对消息的处理是否正确
            优点：消费快，适用于数据一致性弱的业务场景
            缺点：消息很容易丢失
         */
    
        @Test
        public void autoCommit() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            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("mytopic1"));
            //死循环不停的从broker中拿数据
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
        }

运行上面的程序输出结果：

![9d72746344219ee44b48f3f171f0b12da67.jpg][]

使用如下命令查看offset提交后当前位置

./kafka-consumer-groups.sh --bootstrap-server hadoop01:9092 --describe --group my\_group

![0fee32ddfd944e17e278b8d2c51e8b3c97d.jpg][]

比较上面两张图，最后一次消费的OFFSET=216493，下一个要消费的OFFSET=216494

### 2.偏移量-手动按消费者提交 ###

通常从Kafka拿到的消息是要做业务处理，而且业务处理完成才算真正消费成功，所以需要客户端控制offset提交时间

@Test
        public void munualCommit() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            props.put("enable.auto.commit", "false");
            //序列化器
            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("mytopic1"));
            final int minBatchSize = 50;
            List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records) {
                    buffer.add(record);
                }
                if (buffer.size() >= minBatchSize) {
                    //insertIntoDb(buffer);
                    for (ConsumerRecord bf : buffer) {
                        System.out.printf("offset = %d, key = %s, value = %s%n", bf.offset(), bf.key(), bf.value());
                    }
                    consumer.commitSync();
                    buffer.clear();
                }
            }
        }

### 3.偏移量-手动按分区提交 ###

在munualCommit的基础上更细粒度的提交数据，按照每个分区手动提交偏移量

这里实现了按照分区取数据，因此可以从分区入手，不同的分区可以做不同的操作，可以灵活实现一些功能

为了验证手动提交偏移量，有两种方式：

1.debug的时候，在poll数据之后，手动提交前偏移量之前终止程序，再次启动看数据是否重复被拉取

2.debug的时候，在poll数据之后，手动提交前偏移量之前终止程序，登录Linux 主机执行如下命令：

/kafka-consumer-groups.sh --bootstrap-server hadoop01:9092 --describe --group my_group

命令的输出结果可以看到当前topic每个区分被提交后的当前偏移量、还未被消费的最大偏移量、两者之间的差等信息

@Test
        public void munualCommitByPartition() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            props.put("enable.auto.commit", "false");
            //序列化器
            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("mytopic3"));
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("partition: " + partition.partition() + " , " + record.offset() + ": " + record.value());
                        }
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        /*
                            提交的偏移量应该始终是您的应用程序将要读取的下一条消息的偏移量。因此，在调用commitSync（）时，
                            offset应该是处理的最后一条消息的偏移量加1
                            为什么这里要加上面不加喃？因为上面Kafka能够自动帮我们维护所有分区的偏移量设置，有兴趣的同学可以看看SubscriptionState.allConsumed()就知道
                         */
                        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                    }
                }
            } finally {
                consumer.close();
            }
        }

### 4.消费者从指定分区拉取数据 ###

消费只读取特定分区数据，这种方式比上面的更加灵活，在实际应用场景中会经常使用

因为分区的数据是有序的，利用这个特性可以用于数据到达有先后顺序的业务，比如一个用户将订单提交，紧接着又取消订单，那么取消的订单一定要后于提交的订单到达某一个分区，这样保证业务处理的正确性

一旦指定了分区，要注意以下两点：

** a.kafka提供的消费者组内的协调功能就不再有效**

** b.这样的写法可能出现不同消费者分配了相同的分区，为了避免偏移量提交冲突，每个消费者实例的group\_id要不重复**

@Test
        public void munualPollByPartition() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            props.put("enable.auto.commit", "false");
            //序列化器
            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);
            //消费者订阅主题，并设置要拉取的分区
            TopicPartition partition0 = new TopicPartition("mytopic3", 0);
            //TopicPartition partition1 = new TopicPartition("mytopic2", 1);
            //consumer.assign(Arrays.asList(partition0, partition1));
            consumer.assign(Arrays.asList(partition0));
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("partition: " + partition.partition() + " , " + record.offset() + ": " + record.value());
                        }
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                    }
                }
            } finally {
                consumer.close();
            }
        }

## 5.偏移量由消费者管理 ##

**kafka Consumer Api还提供了自己存储offset的功能，将offset和data做到原子性，可以让消费具有Exactly Once 的语义，比kafka默认的At-least Once更强大**

### 消费者从指定分区拉取数据-手动更改偏移量 ###

设置消费者从自定义的位置开始拉取数据，比如从程序停止时已消费的下一Offset开始拉取数据，使用这个功能要求data和offset的update操作是原子的，否则可能会破坏数据一致性

/*
            手动设置指定分区的offset，只适用于使用Consumer.assign方法添加主题的分区，不适用于kafka自动管理消费者组中的消费者场景，
            后面这种场景可以使用ConsumerRebalanceListener做故障恢复使用
         */
        @Test
        public void controlsOffset() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            props.put("enable.auto.commit", "false");
            //序列化器
            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);
            //消费者订阅主题，并设置要拉取的分区
    
            //加一段代码将自己保存的分区和偏移量读取到内存
            //load partition and it's offset
            TopicPartition partition0 = new TopicPartition("mytopic3", 0);
            consumer.assign(Arrays.asList(partition0));
    
            //告知Consumer每个分区应该从什么位置开始拉取数据，offset从你加载的值或者集合中拿
            consumer.seek(partition0, 4140l);
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
    
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("partition: " + partition.partition() + " , " + record.offset() + ": " + record.value());
                        }
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                    }
                }
            } finally {
                consumer.close();
            }
        }

代码和上面的绝大多数都一样，就是要自己加载分区信息，给消费者设置每个分区的偏移量

### 添加因消费者改变导致kafka rebalance的监听 ###

kafka提供该监听来处理当某一个topic的消费者发生变化（加入、退出）时分区重新分配（先解除与消费者的绑定关系，再重新与消费者绑定）用户想做回调的情况，分区与消费者解除绑定时调用onPartitionsRevoked方法；重新绑定时调用onPartitionsAssigned。

监听代码

import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
    import org.apache.kafka.common.TopicPartition;
    
    import java.util.Collection;
    import java.util.Map;
    import java.util.concurrent.ConcurrentHashMap;
    
    /*
        kafka提供了这个监听来处理分区的变化，区分被取消时调用onPartitionsRevoked方法；分区被分配时调用onPartitionsAssigned
     */
    public class MyConsumerRebalanceListener implements ConsumerRebalanceListener {
        static Map<TopicPartition,Long> partitionMap = new ConcurrentHashMap<>();
        private Consumer<?,?> consumer;
        //实例化Listener的时候将Consumer传进来
        public MyConsumerRebalanceListener(Consumer<?,?> consumer) {
            this.consumer = consumer;
        }
    
        /*
            有新的消费者加入消费者组或者已有消费者从消费者组中移除会触发kafka的rebalance机制，rebalance被调用前会先调用下面的方法
            此时你可以将分区和它的偏移量记录到外部存储中，比如DBMS、文件、缓存数据库等，还可以在这里处理自己的业务逻辑
         */
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            for(TopicPartition partition: partitions){
                //记录分区和它的偏移量
                partitionMap.put(partition,consumer.position(partition));
                //清空缓存
    
                System.out.println("onPartitionsRevoked partition:" + partition.partition()+" - offset"+consumer.position(partition));
            }
        }
    
        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            //设置分区的偏移量
            for(TopicPartition partition: partitions){
                System.out.println("onPartitionsAssigned partition:" + partition.partition()+" - offset"+consumer.position(partition));
                if(partitionMap.get(partition)!=null){
                    consumer.seek(partition, partitionMap.get(partition));
                }else{
                    //自定义处理逻辑
                }
            }
        }
    }

测试代码

@Test
        public void autoCommitAddListner(){
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交 true-开启 false-关闭
            props.put("enable.auto.commit", "false");
            //序列化器
            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);
            MyConsumerRebalanceListener myListener = new MyConsumerRebalanceListener(consumer);
            //消费者订阅主题，可以订阅多个主题
            consumer.subscribe(Arrays.asList("mytopic3"),myListener);
            //consumer.subscribe(Arrays.asList("mytopic3"));
            try {
                while (true) {
                    ConsumerRecords<String, String> records = consumer.poll(Long.MAX_VALUE);
                    for (TopicPartition partition : records.partitions()) {
                        List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                        for (ConsumerRecord<String, String> record : partitionRecords) {
                            System.out.println("partition: " + partition.partition() + " , " + record.offset() + ": " + record.value());
                        }
                        long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                        /*
                            可以将这里的偏移量提交挪到监听的onPartitionsRevoked方法中，控制灵活，但是也很容易出问题
                         */
                        consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
                    }
                }
            } finally {
                consumer.close();
            }
        }

## 6.其他 ##

*  **使用pause和resume可以暂停和恢复一个分区的消费动作**
    
    consumer.**pause**(Arrays.asList(new TopicPartition("topic\_name",parition\_num)))
    
    consumer.**resume**(Arrays.asList(new TopicPartition("topic\_name",parition\_num)))
 *  **按事务读数据**
    
    该操作与Producer的按事务写相匹配，在Consumer代码的配置中增加一行：
    
        props.put("isolation.level","read_committed");
    
    **注意，按事务读，不能使用在按指定分区拉取数据的消费者中**

## 7.多线程 ##

KafkaConsumer是线程不安全，kafka官方提供了一种写法来避免线程安全问题

ConsumerRunner:

package com.jv.parallel;
    
    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.errors.WakeupException;
    
    import java.util.Arrays;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    public class ConsumerRunner implements Runnable{
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private final KafkaConsumer<String,String> consumer;
        private final CountDownLatch latch;
    
        public ConsumerRunner(KafkaConsumer<String,String> consumer, CountDownLatch latch){
            this.consumer = consumer;
            this.latch = latch;
        }
    
        @Override
        public void run() {
            System.out.println("threadName...."+Thread.currentThread().getName());
            try {
                consumer.subscribe(Arrays.asList("mytopic3"));
                while (!closed.get()) {
                    ConsumerRecords<String, String> records = consumer.poll(10000);
                    for (ConsumerRecord<String, String> record : records)
                        System.out.printf("threadName= %s, offset = %d, key = %s, value = %s%n", Thread.currentThread().getName(),record.offset(), record.key(), record.value());
                }
            } catch (WakeupException e) {
                if (!closed.get()) throw e;
            } finally {
                consumer.close();
                latch.countDown();
            }
        }
    
        public void shutdown() {
            System.out.println("close ConsumerRunner");
            closed.set(true);
            consumer.wakeup();
        }
    }

驱动方法：

@Test
        public void autoCommitParallelTest() {
            Properties props = new Properties();
            //设置kafka集群的地址
            props.put("bootstrap.servers", "hadoop01:9092,hadoop02:9092,hadoop03:9092");
            //设置消费者组，组名字自定义，组名字相同的消费者在一个组
            props.put("group.id", "my_group");
            //开启offset自动提交
            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");
            //实例化一个消费者
            final List<ConsumerRunner> consumers = new ArrayList<>();
            final List<KafkaConsumer<String, String>> kafkaConsumers = new ArrayList<>();
            for(int i = 0;i < 2;i++){
                kafkaConsumers.add(new KafkaConsumer<String, String>(props));
            }
            final CountDownLatch latch = new CountDownLatch(2);
            final ExecutorService executor = Executors.newFixedThreadPool(2);
            for(int i = 0;i < 2;i++){
                ConsumerRunner c = new ConsumerRunner(kafkaConsumers.get(i),latch);
                consumers.add(c);
                executor.submit(c);
            }
    
            /*
                这个方法的意思就是在jvm中增加一个关闭的钩子，当jvm关闭的时候，会执行系统中已经设置的所有通过方法addShutdownHook添加的钩子，当系统执行完这些钩子后，jvm才会关闭
                所以这些钩子可以在jvm关闭的时候进行内存清理、对象销毁、关闭连接等操作
             */
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    System.out.println("....................");
                    for (ConsumerRunner consumer : consumers) {
                        consumer.shutdown();
                    }
                    executor.shutdown();
                    try {
                        executor.awaitTermination(5000, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
    
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

但是Kafka官方任然不建议多个线程共用一个Consumer，否则会出现ConcurrentModificationException异常

Kafka提供如下两个方案实现并发：

1.一个线程一个Consumer

每个线程都拥有自己的Consumer

优点：

*  写代码容易
 *  因为不需要协调和调度线程，速度比较快
 *  实现分区的有序非常容易

缺点：

*  TCP连接更多，如果分区非常多，这种方案不可行了
 *  消费者多了，可能因为批处理少了，使IO吞吐量减少
 *  并发数严重依赖于分区数（消费者数只能小于等于分区数）

2.Consumer和Processer分离

使用一个或者多个Consumer从Kafka拉取数据，并将数据放到一个阻塞队列中，由Processor从阻塞队列中获取数据并做业务处理。

优点：

*  将消费和业务处理做垂直切分，然后在水平上可以独立的进行扩展

缺点：

*  分区顺序难以保障
 *  分区提交非常麻烦

针对这种方案的分区内数据顺序问题，可以使用让每个消费者都有自己的阻塞队列。因为Consumer和Processor已经分离了，如何让Consumer知道数据已经被Processor处理完是比较麻烦的事情，

[http_kafka.apache.org_documentation_producerconfigs]: http://kafka.apache.org/documentation/#producerconfigs
[9d72746344219ee44b48f3f171f0b12da67.jpg]: /images/20211227/4452b633b52a443aac441757c580f0ab.png
[0fee32ddfd944e17e278b8d2c51e8b3c97d.jpg]: /images/20211227/fd73b1d2c68c4892a18a3e30aad68cd5.png