fetch.max.wait.ms默认500ms

生产者连接集群所需的broker地址清单。

2.4.2 默认分区规则

1）指定分区，按分区进行划分

2）不指定分区，指定key，按key的hashcode值%分区数

3）不指定分区，不指定key，粘性

        第一次随机，一旦粘上一个，直到该分区挂掉（批次大小到了，或者响应时间到了）再切换分区。

二、最多可以缓存五个请求。RabbitMQ、

——为什么Kafka不用Java的序列化器？

        Java的序列化传输的数据比较重。

注意：这种方式只能保证副本之间的数据一致性，并不能保证数据不丢失或者不重复。可能会丢失数据

 1：生产者发送过来的数据，Leader收到数据后应答。该时间阈值由 replica.lag.time.max.ms参数设定，默认 30s。数据传输线程数。Kafka2.8.0以后也可以配置不采用ZK

        主要功能分为三块：生产者Producer、

0.9版本之前，offset会存储在zk中。那么消费者组是如何形成的呢？由coordinator组件辅助实现消费者组的初始化和分区的分配。因此这不是一个绝对最大值。默认打开

2.7.3 生产者事务

说明：开启事务，必须开启幂等性。

（2）CNC前期进行一些准备工作，首先调用 sendFetchs方法，用来抓取数据的初始化。但如果所有的offset都存储在zk中，那么消费者customer会和zk进行大量的交互，导致网络上数据传输非常频繁，传输压力过大。

——发送数据的条件：

        1. 只有数据累积到batch.size之后，sender才会发送数据，默认16k

        2. 如果数据迟迟未达到batch.size，sender等待linger.ms设置的时间到了之后就会发送数据。发送完请求之后，会通过回调方法onSuccess将对应的结果拉取过来。ActiveMQ、

        通过Kafka可以进行缓冲，我们可以将海量的数据先放到Kafka中，Kafka集群按照Hadoop的上传速度进行文件的传输。

c. 流量削峰

        有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

Kafka消费者和coordinator之间的心跳时间，默认3s。-1与all等价。

（4）拉去过来的数据会放在消息队列queue中

（5）数据拉过来之后，消费者FetchedRecords从队列中抓取数据。Kafka 生产者只会把数据发往 Leader，

public static void main(String[] args) {// 1. 配置信息    Properties properties = new Properties();    // 连接    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFFIG, "hadoop02:9092,hadoop03:9092");    // 反序列化    properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class);    properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserizlizer.class);    // 组id    properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");// 2. 创建消费者    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);    // 指定位置进行消费    Set<TopicPartition> assignment = kafkaConsumer.assignment;    // 保证分区分配方案已经制定完毕    while(assignment.size() == 0){        kafkaConsumer.poll(Duration.ofSeconds(1));        assignment = kafkaConsumer.assignment();    }    // 指定消费的offset    for(TopicPartition topicPartition : assignment){        kafkaConsumer.seek(topicPartition, 100);    }// 3.订阅主题     ArrayList<String> topics = new ArrayList<>();    topics.add("first");    kafkaConsumer.subscribe(topics);// 4. 消费数据    while(true){        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));        for(ConsumerRecord<String, String> consumerRecord : consumerRecords){            System.out.println(consumerRecord);        }    }}

4.4.5 指定时间消费

        需求：在生产环境中，会遇到最近消费的几个小时数据异常，想重新按照时间消费。 

 4.1.3 消费者组的初始化流程

        生产者把数据发送到Kafka集群，由消费者组中的消费者进行数据消费。于是引入了Kafka。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。因此Kafka提供了手动提交offset的API。RoundRobin、

默认default:50M。

2）异步发送

        将外部数据发送到队列DQueue中，不管数据是否发送到Kafka集群中。

默认是1。

Kafka消费者和coordinate之间连接超时时间，默认45s。但是，如果Kafka中没有数据，消费者会陷入循环，一直返回空数据。

——应答成功，清除掉所有的请求request，同时清理分区的数据。

        Max.poll.records        一次拉去数据返回消息的最大条数，默认500条

（6）数据拉去过来之后，经过parseRecord(反序列化)、

一般不建议修改。解耦、

发布/订阅：消息的发布者不会将消息直接发送给特点的订阅者，而是将发布的消息（数据）分为不同的类型，订阅者只接收感兴趣的消息，根据需求选择性订阅。

生产者Producer：对接外部设备（外部数据）

消费者Consumer：处理数据

Topic：存储数据。

超过该值，该消费者被移除，消费者组执行再平衡。如消息的key是用户ID，value是用户的资料。为了查看该系统主题，将配置文件 config / consumer.properties 中添加配置 exclude. internal.topics=false 。

3.3 Kafka副本Follower

3.3.1 Kafka副本基本信息

// 是否自动提交offset    propertoes.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);// 提交offset的时间周期为1000ms，默认为5000ms    properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);

__consumer_offsets的分区数。单位ms，默认为0ms，表示没有延迟。合理控制分区的任务，可以实现负载均衡的效果。

4）controller决定leader选举：在isr中存活为前提，按照AR中（AR启动的时候会有固定的顺序）排在前面的优先。两者的相同点是，都会将本次提交的一批数据最高的偏移量提交；不同点是，同步提交阻塞当前线程，一直到提交成功，并且会失败重试；而异步提交则没有失败重试机制，可能提交失败。指定消息队列名字 ArrayList<String> topics = new ArrayList<>(); topics.add("first"); consumer.subscribe(topics);// 5. 消费数据 while(true){ ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord<String, String> record : records) { // 如果数据过程过程中失败，可以将相关位置记录下来 System.out.println( record.topic() + "\t" + record.offset() + "\t" + record.partition() + "\t" + "key:" + record.key() + "\t" + "value:" + record.value() + "\t" + record.timestamp() ); consumer.commitAsync(); } } }}

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

public class MyConsumer {    public static void main(String[] args) {// 1. 配置        Properties pro = new Properties();    // 连接        pro.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.153.139:9092");    // 反序列化        pro.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);        pro.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);    // 组id        pro.put(ConsumerConfig.GROUP_ID_CONFIG, "GROUP1");// 2. 创建一个消费者        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(pro);// 3. 订阅主题对应的分区        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();        topicPartitions.add(new TopicPartition("bigdata",0));        consumer.assign(topicPartitions);// 4. 消费数据        while(true){            ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));            for(ConsumerRecords<String, String> record : records){                System.out.println(record);            }        }    }}

4.3 生产经验——分区的分配以及再平衡

        一个consumer group中有多个consumer组成，一个topic有多个partition组成。自动Leader Partition平衡。 Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。每个segment包含：

        .log                  日志文件

        .index              偏移量索引文件

        .timeindex        时间戳索引文件

每个segment大小为1G

log文件和 index文件：

 1）index为稀疏索引。

4.3.2 RoundRobin + 再平衡

1）RoundRobin分区策略原理

        RoundRobin 针对集群中所有Topic而言。