Kafka` 消费者

`Kafka`消费方式

pull（拉）模式：
- consumer采用从broker中主动拉取数据
Kafka采用这种方式
- pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据

`Kafka`消费者总体工作流程

每个分区的数据只能由消费者组中一个消费者消费
一个消费者可以消费多个分区数据
每个消费者的offset由消费者提交到系统主题保存

消费者组原理

Consumer Group(CG)：消费者组，由多个conumser组成。形成一个消费者组的条件，是所有消费者的groupid相同
- 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费
- 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者
- 如果向消费者组中添加更多的消费者，超过主题分区数量，则有一部分消费者就会闲置，不会接受任何消息
- 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者

消费者初始化流程

coordinator：辅助实现消费者组的初始化和分区的分配
- coordinator节点选择：groupid的hashcode值%50(_consumer_offsets的分区数量)
流程
1. 每个conumser都发送JoinGroup请求给coordinator
2. coordinator会选出一个consumer作为leader
3. coordinator把要消费的topic情况发送给leader消费者
4. leader会负责制定消费方案
5. leader把消费方案发送给coordinator
6. coordinator把消费方案发送给各个consumer
  - 每个消费者都会和coordinator保持心跳（默认3s），一旦超时(session.timeout.ms=45s)，该消费者会被移除，并触发再平衡；或者消费者处理消息的时间过长(max.poll.interval.ms=5min)，也会触发再平衡
7. 消费者组初始化完成后，会先初始化ConsumerNetworkClient
8. consumer向ConsumerNetworkClient发送消费请求(sendFetches)
  - Fetch.min.bytes每批次最小抓取大小，默认1字节
  - fetch.max.wait.ms：一批数据最小值未达到的超时时间，默认500ms
  - Fetch.max.bytes每批次最大抓取大小，默认50m
9. ConsumerNetworkClient将上述三个参数发送给(send)Broker的Leader
10. ConsumerNetworkClient会调用一个回调函数onSuccess拉取数据到completedFetches(queue)中
11. Conumser会从completedFetches(queue)中拉取数据，经过parseRecord(反序列化)，Interceptors(拦截器)，最后处理数据
  - Max.poll.records：一次拉取数据返回消息的最大条数，默认500条

消费者`API`

独立消费者案例（订阅主题）

注意：在消费者API代码中必须配置消费者组id。命令行启动消费者不填写消费者组id会被自动填写随机的消费者组id

public class CustomConsumer {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");


        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

独立消费者案例（订阅分区）

public class CustomConsumerPartition {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题对应的分区
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first",0));
        kafkaConsumer.assign(topicPartitions);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

消费者组案例

多个线程执行相同的代码，必须配置消费者组id

public class CustomConsumer {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test5");

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

`offset`位移

`offset`的默认维护位置

Kafka0.9版本之前，consumer默认将offset保存在Zookeeper中。从0.9版本开始，consumer默认将offset保存在Kafka的一个内置的topic中，该topic为__conumser_offsets
__conumser_offsets主题里面采用key和value的方式存储数据。key是group.id+topic+分区号，value就是当前的offset的值。每隔一段时间，kafka内部会对这个topic进行compact,也就是每个group.id+topic+分区号就保留最新数据

查看`offset`

在配置文件config/consumer.properties中添加配置exclude.internal.topics=false，默认是true,表示不能消费系统主题。为了查看该系统主题数据，所以该参数修改为false

查看消费者消费主题__consumer_offsets

bin/kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server hadoop102:9092 --consumer.config config/consumer.properties --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning

自动提交`offset`

为了使我们能够专注于自己的业务逻辑，Kafka提供了自动提交offset的功能
自动提交offset的相关参数
- enable.auto.commit：是否开启自动提交offset功能，默认是true
- auto.commit.interval.ms：自动提交offset的时间间隔，默认是5s

消费者自动提交offset

public class CustomConsumerAutoOffset {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");

        //自动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,true);

        //提交时间间隔
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000);

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

手动提交`offset`

手动提交offset的方法有两种：
- commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据
- commitAsync（异步提交）：发送完提交offset请求后，就开始消费下一批数据
  - 两者的相同点：
    - 都会将本次提交的一批数据最高的偏移量提交
  - 不同点
    - 同步提交阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）
    - 异步提交则没有失败重试机制，固有可能提交失败

同步提交`offset`

由于同步提交offset有失败重试机制，故更加可靠，但是由于一直等待提交结果，提交的效率比较低。

public class CustomConsumerByHandSync {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test4");

        //手动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

            //手动提交offset
            kafkaConsumer.commitSync();

        }

    }
}

异步提交`offset`

虽然同步提交offset更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会受到很大的影响。因此更多的情况下，会使用异步提交offset的方式

public class CustomConsumerByHandSync {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test4");

        //手动提交
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,false);

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
        properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

            //手动提交offset
            kafkaConsumer.commitAsync();

        }

    }
}

指定`Offset`消费

auto.offset.reset=earliest|lastest|none：默认是latest
当Kafka中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除）
1. earliest：自动将偏移量重置为最早的偏移量,--from-beginging
2. latest（默认值）：自动将偏移量重置为最新偏移量
3. none：如果未找到消费者组的先前偏移量，则向消费者抛出异常

任意指定`offset`位移开始消费

public class CustomConsumerSeek {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test10");

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer
        // .RoundRobinAssignor");
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer
        // .StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //指定位置进行消费
        Set<TopicPartition> assignment = kafkaConsumer.assignment();

        //保证分区分配方案已经制定完毕
        while (assignment.size() == 0) {

            kafkaConsumer.poll(Duration.ofSeconds(1));

            assignment = kafkaConsumer.assignment();
        }

        for (TopicPartition topicPartition : assignment) {

            kafkaConsumer.seek(topicPartition, 100);
        }


        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

指定时间消费

public class CustomConsumerSeek {
    public static void main(String[] args) {

        //0配置
        Properties properties = new Properties();

        // 连续
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092,hadoop103:9092");

        //反序列化
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        //配置消费者组id
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test10");

        //设置分区分配策略
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer
        // .RoundRobinAssignor");
        //properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer
        // .StickyAssignor");

        //1创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);


        //2定义主题first
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);

        //指定位置进行消费
        Set<TopicPartition> assignment = kafkaConsumer.assignment();

        //保证分区分配方案已经制定完毕
        while (assignment.size() == 0) {

            kafkaConsumer.poll(Duration.ofSeconds(1));

            assignment = kafkaConsumer.assignment();
        }


        //希望把时间转换为对应的offset
        HashMap<TopicPartition, Long> topicPartitionLongHashMap = new HashMap<>();

        //封装对应集合
        HashMap<TopicPartition, Long> partitionLongHashMap = new HashMap<>();

        for (TopicPartition topicPartition : assignment) {
            partitionLongHashMap.put(topicPartition, System.currentTimeMillis() - 30 * 60 * 1000);
        }

        Map<TopicPartition, OffsetAndTimestamp> topicPartitionOffsetAndTimestampMap =
                kafkaConsumer.offsetsForTimes(partitionLongHashMap);

        for (TopicPartition topicPartition : assignment) {
            OffsetAndTimestamp offsetAndTimestamp = topicPartitionOffsetAndTimestampMap.get(topicPartition);
//            kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
            if(offsetAndTimestamp != null){
                kafkaConsumer.seek(topicPartition,offsetAndTimestamp.offset());
            }
        }


        //3消费数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));

            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }

        }

    }
}

漏消费和重复消费

漏消费

先提交offset后消费，有可能会造成数据的漏消费
- 场景：
  - 设置offset为手动提交，当offset被提交时，数据还在内存中未落盘，此时刚好消费者线程被kill掉，那么offset已经提交，但是数据未处理，导致这部分内存中的数据丢失

重复消费

已经消费了数据，但是offset没提交
- 场景：
  - Consumer每5s提交offset
  - 如果提交offset后的2s，consumer挂了
  - 再次重启consumer，则从上一次提交的offset处继续消费，导致重复消费

消费者重要参数

参数名称	描述
`bootstrap.servers`	向`Kafka` 集群建立初始连接用到的host/port 列表。
`key.deserializer` 和 `value.deserializer`	指定接收消息的`key` 和`value` 的反序列化类型。一定要写全类名
`group.id`	标记消费者所属的消费者组。
`enable.auto.commit`	默认值为`true`，消费者会自动周期性地向服务器提交偏移量。
`auto.commit.interval.ms`	如果设置了 `enable.auto.commit` 的值为`true`，则该值定义了消费者偏移量向`Kafka` 提交的频率，默认`5s`。
`auto.offset.reset`	当`Kafka` 中没有初始偏移量或当前偏移量在服务器中不存在（如，数据被删除了），该如何处理？ `earliest`：自动重置偏，移量到最早的偏移量。 `latest`：默认，自动重置偏移量为最新的偏移量。 `none`：如果消费组原来的（`previous`）偏移量不存在，则向消费者抛异常。 `anything`：向消费者抛异常。
`offsets.topic.num.partitions`	`__consumer_offsets` 的分区数，默认是50 个分区。
`heartbeat.interval.ms`	`Kafka` 消费者和`coordinator` 之间的心跳时间，默认3s。该条目的值必须小于 `session.timeout.ms` ，也不应该高于 `session.timeout.ms` 的1/3。
`session.timeout.ms`	`Kafka` 消费者和`coordinator`之间连接超时时间，默认45s。超过该值，该消费者被移除，消费者组执行再平衡。
`max.poll.interval.ms`	消费者处理消息的最大时长，默认是 5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
`fetch.min.bytes`	默认 1 个字节。消费者获取服务器端一批消息最小的字节数。
`fetch.max.wait.ms`	默认 `5 00 ms` 。如果没有从服务器端获取到一批数据的最小字节数。该时间到，仍然会返回数据。
`fetch.max.bytes`	默认 `Default:` 52428800 5 0 m 。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（`50m` ）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 `message.max.bytes` (`broker.config`) or `max.message.bytes` (`topic.config`) 影响。
`max.poll.records`	一次 `poll` 拉取数据返回消息的最大条数，默认是 500 条。

生产经验

分区的分配以及再平衡

一个conumser group中由多个consumer组成，一个topic由多个partition组成

Kafka有四种主流的分区分配策略：Range、RoundRobin、Sticky、CooperativeSticky

可以通过配置参数partition.assignment.straegy，修改分区的分配策略。默认策略是Range+CooperativeSticky。Kafka可以同时使用多个分区分配策略

参数名称	描述
`heartbeat.interval.ms`	`Kafka` 消费者和`coordinator` 之间的心跳时间，默认`3s`。该条目的值必须小于 `ession.timeout.ms` ，也不应该高于 `session.timeout.ms` 的1/3。
`session.timeout.ms`	`Kafka` 消费者和`coordinator` 之间连接超时时间，默认`45s`。超过该值，该消费者被移除，消费者组执行再平衡。
`max.poll.interval.ms`	消费者处理消息的最大时长，默认是5 分钟。超过该值，该消费者被移除，消费者组执行再平衡。
`partition.assignment.strategy`	消费者分区分配策略，默认策略是`Range +CooperativeSticky`。`Kafka` 可以同时使用多个分区分配策略。可以选择的策略包括： `Range` 、`RoundRobin` 、`Sticky` 、`CooperativeSticky`

`Range`分区策略原理

Range是对每个topic而言的
- 首先对同一个topic里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序
- 通过partitions数/consumer数来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费1个分区
  - 注意：
    - 如果只是针对1个topic而言，C0消费者多消费1个分区影响不是很大。但是如果有N多个topic，那么针对每个topic，消费者C0都将多消费1个分区，topic越多，C0消费的分区会比其他消费者明显多消费N个分区
    - 容易产生数据倾斜

`RoundRobin`分区策略原理

RoundRobin针对集群中所有topic而言
RoundRobin轮询分区策略，是把所有的partition和所有的consumer都列出来，然后按照hashcode进行排序，最后通过轮询算法来分配partition给到各个消费者

RoundRobin分区分配策略案例
1. 依次在CustomConsumer、CustomConsumer1、CustomConsumer2三个消费者代码中修改分区分配策略为RoundRobin
```
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.RoundRobinAssignor");
```

`Sticky`以及再平衡

粘性分区定义：可以理解为分配的结果带有"粘性"。即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可以节省大量的开销
- 首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化
- 修改分区分配策略为粘性
```
properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,"org.apache.kafka.clients.consumer.StickyAssignor");
```

消费者事务

如果想完成Consummer端的精准一次性消费，那么需要Kafka消费端将消费过程和提交offset过程做原子绑定。此时我们需要将Kafka的offset保存到支持事务的自定义介质(比如MySQL)。

数据积压（消费者如何提高吞吐量）

如果是Kafka消费能力不足，则可以考虑增加Topic的分区数，并且同时提升消费组的消费者数量，消费者数=分区数（两者缺一不可）
如果是下游的数据处理不及时：提高每批次拉取的数量。批次拉取数据过少（拉取数据/处理时间 < 生产速度），使处理的数据小于生产的数据，也会造成数据积压

参数名称	描述
`fetch.max.bytes`	默认`Default`: `52428800`（50 m）。消费者获取服务器端一批消息最大的字节数。如果服务器端一批次的数据大于该值（50m）仍然可以拉取回来这批数据，因此，这不是一个绝对最大值。一批次的大小受 `message.max.bytes` （`broker config`）or `max.message.bytes` （`topic config`）影响。
`max.poll.records`	一次 poll 拉取数据返回消息的最大条数，默认是 500 条

Kafka` 消费者

Kafka消费方式

Kafka消费者总体工作流程

消费者组原理

消费者初始化流程

消费者API

独立消费者案例（订阅主题）

独立消费者案例（订阅分区）

消费者组案例

offset位移

offset的默认维护位置

查看offset

自动提交offset

手动提交offset

同步提交offset

异步提交offset

指定Offset消费

任意指定offset位移开始消费