学会Kafka这一篇就够了

Kafka简介

Kafka是 一个开源的分布式事件流平台 （Event Streaming Platform），被数千家公司用于高性能数据管道、流分析、数据集成和关键任务应用，是一个分布式的基于发布/订阅模式的消息队列（Message Queue）。

官方文档： https://kafka.apache.org/documentation/#quickstart

学习Kafka的过程会相对于RocketMQ、RabbitMQ等消息队列简单，因为它采用最为简单且常用的发布/订阅模式，采用的思想也都大同小异.

消息队列的两种模式

1）点对点模式

• 消费者主动拉取数据，消息收到后清除消息
  

2. 发布/订阅模式

• 可以有多个topic主题
• 消费者消费数据之后，不删除数据
• 每个消费者相互独立，都可以消费到数据

基础架构

核心概念

• Producer：消息生产者，就是向 Kafka broker 发消息的客户端。
• Consumer：消息消费者，向 Kafka broker 取消息的客户端。
• Consumer Group（CG）：消费者组，由多个 consumer 组成。消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。
• Broker：一台 Kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个broker 可以容纳多个 topic。
• Topic：可以理解为一个队列，生产者和消费者面向的都是一个 topic。
• Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，一个 topic 可以分为多个 partition(提高并行度），每个 partition 是一个有序的队列。
• Replica：副本。一个 topic 的每个分区都有若干个副本，一个 Leader 和若干个Follower。
• Leader：每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对象都是 Leader。
• Follower：每个分区多个副本中的“从”，实时从 Leader 中同步数据，保持和Leader 数据的同步。Leader 发生故障时，某个 Follower 会成为新的 Leader。

Kafka 集群搭建安装

安装教程

根目录启动:

bin/kafka-server-start.sh -daemon config/server.properties

根目录关闭:

bin/kafka-server-stop.sh

Kafka 基础命令行

主题命令行操作

• 查看操作主题命令参数

[root@root kafka_2.12-3.0.0]# bin/kafka-topics.sh
1

• list 查看当前服务器中的所有Topic

bin/kafka-topics.sh --bootstrap-server localhost:9092 --list

• create 创建topic 名为first

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic first

选项说明：
–topic 定义 topic 名
–replication-factor 定义副本数
–partitions 定义分区数

• describe 查看主题的详情

bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic first

• alter 修改分区数（注意：分区数只能增加，不能减少）

bin/kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic first --partitions 3

• delete 删除 topic

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic first

生产者命令行操作

• 查看操作生产者命令参数

[root@root kafka_2.12-3.0.0]# bin/kafka-console-producer.sh
1

• 发送消息

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic first

消费者命令行操作

• 查看操作消费者命令参数

[root@root kafka_2.12-3.0.0]# bin/kafka-console-consumer.sh
1

• 消费 first 主题中的数据 (默认latest)

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic first

• 把主题中所有的数据都读取出来–from-beginning（包括历史数据）。

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic first

以上就是命令行的基础操作,如果要学习各个命令行的进一步参数,可以–help 查看详情.

---

Kafka 生产者

发送原理

在消息发送的过程中，涉及到了两个线程——main 线程和 Sender 线程。在 main 线程
中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，
Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker。

---

pom依赖

    <dependencies>
        <dependency>
            <groupId>org.apache.kafkagroupId>
            <artifactId>kafka-clientsartifactId>
            <version>3.0.0version>
        dependency>
    dependencies>
1
2
3
4
5
6
7

---

消息发送API

kafka producer默认是异步发送：

在初始化producer实例时，会创建一个sender线程负责批量发送消息
producer将消息暂存在缓冲区，消息根据topic-partition分类缓存
消息达到batch.size或者时间达到了linger.ms，sender线程将该批量的消息发送到topic-partition所在的broker

• 异步发送

Java异步发送分两种,一种是普通异步发送,一种是带回调函数的异步发送.回调函数会在 producer 收到 ack 时调用，为异步调用，该方法有两个参数，分别是元数据信息（RecordMetadata）和异常信息（Exception），如果 Exception 为 null，说明消息发送成功，如果 Exception 不为 null，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。从代码书写上,区别在与send()方法加不加callback回调函数.

public class CustomProducer {
    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "120.xx.xxx.xxx:9092");
                
        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "消息" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主题： " +
                                metadata.topic() + "->" + "分区：" + metadata.partition());
                    } else {
                        // 出现异常打印
                        exception.printStackTrace();
                    }
                }
            });
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

• 同步发送

如果需要使用同步发送，可以在每次发送之后使用get方法，因为producer.send方法返回一个Future类型的结果，Future的get方法会一直阻塞直到该线程的任务得到返回值，也就是broker返回发送成功。

Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>("first", "消息" + i));
RecordMetadata recordMetadata = future.get();
System.out.println(" 主题： " + recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());
1
2
3

---

生产者分区

概念

通过核心概念的简述,我们知道一个Topic主题创建时可以分配多个Partition分区,Topic是抽象的,而Partition是具体的存储.它可以分配到多个Broker节点上. 且同一个Partition只能被同一个消费者组的一个消费者消费.

• Partition的作用及好处

（1）便于合理使用存储资源，每个Partition在一个Broker上存储，可以把海量的数据按照分区切割成一
块一块数据存储在多台Broker上。合理控制分区的任务，可以实现负载均衡的效果。
（2）提高并行度，生产者可以以分区为单位发送数据；消费者可以以分区为单位进行消费数据,同一主题下增加Partition与消费者的数量,变向的提高了消费能力。

分区策略

通过ProducerRecord类,查看在类中可以看到如下构造方法可知,我们可以在send()方法是可以通过构造方法传入partition的取值,采取的策略如下:

• 指明partition的情况下，直接将指明的值作为partition值；例如partition=0，所有数据写入分区0
• 没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；
  例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，那 么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。
• 既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。
  例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到， Kafka再随机一个分区进行使用（如果还是0会继续随机）

kafkaProducer.send(new ProducerRecord<>("first",1,"key", "消息"));
1

自定义分区器

研发人员可以根据企业需求，自己重新实现分区器,通过定义类实现 Partitioner 接口,重写 partition()方法

~编写自定义分区器

/**
 * 1. 实现接口 Partitioner
 * 2. 实现 3 个方法:partition,close,configure
 * 3. 编写 partition 方法,返回分区号
 */
public class MyPartitioner implements Partitioner {
    /**
     * 返回信息对应的分区
     * @param topic 主题
     * @param key 消息的 key
     * @param keyBytes 消息的 key 序列化后的字节数组
     * @param value 消息的 value
     * @param valueBytes 消息的 value 序列化后的字节数组
     * @param cluster 集群元数据可以查看分区信息
     * @return
         */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {

        if (value.toString().contains("hello")){
            return 1;
        }
        return 0;
    }

    @Override
    public void close() {
    }

    @Override
    public void configure(Map<String, ?> map) {
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

~ 使用自定义分区器

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "120.xx:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // *******使用自定义分区器***************
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"producer.MyPartitioner");
        
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 5; i++) {
            Future<RecordMetadata> future = kafkaProducer.send(new ProducerRecord<>("first", "hello" + i));
            RecordMetadata recordMetadata = future.get();
            System.out.println(" 主题： " + recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

---

Kafka 生产者-生产经验

生产者提高吞吐量

回顾生产者发送数据的架构,可以通过修改参数提高:

• batch.size：批次大小，默认16k
• linger.ms：等待时间，修改为5-100ms ,默认0
• compression.type：压缩snappy ,默认none
• RecordAccumulator：缓冲区大小，修改为64m 默认32m

 public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "120.xx:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // batch.size：批次大小，默认 16K
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        // linger.ms：等待时间，默认 0
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
        // RecordAccumulator：缓冲区大小，默认 32M：buffer.memory
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
        // compression.type：压缩，默认 none，可配置值 gzip、snappy、lz4 和 zstd
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"snappy");

        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
        // 4. 调用 send 方法,发送消息
        for (int i = 0; i < 50000; i++) {
            kafkaProducer.send(new ProducerRecord<>("first", "消息" + i), new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception == null) {
                        // 没有异常,输出信息到控制台
                        System.out.println(" 主题： " +
                                metadata.topic() + "->" + "分区：" + metadata.partition());
                    } else {
                        exception.printStackTrace();
                    }
                }
            });
        }
        // 5. 关闭资源
        kafkaProducer.close();
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

---

数据可靠性

数据完全可靠条件 = ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

ISR：

• Leader维护了一个动态的in-sync replica set（ISR），意为和Leader保持同步的Follower+Leader集合(leader：0，isr:0,1,2)。
• 如果Follower长时间未向Leader发送通信请求或同步数据，则该Follower将被踢出ISR。该时间阈值由replica.lag.time.max.ms参数设定，默认30s。例如2超时，(leader:0, isr:0,1)。

acks：

• 0：生产者发送过来的数据，不需要等数据落盘应答
  数据可靠性分析：丢数

• 1：生产者发送过来的数据，Leader收到数据后应答。
  数据可靠性分析：丢数

• -1（all）：生产者发送过来的数据，Leader和ISR队列里面的所有节点收齐数据后应答.
  数据可靠性分析：如果分区副本设置为1个，或者ISR里应答的最小副本数量（ min.insync.replicas 默认为1）设置为1，和ack=1的效果是一样的，仍然有丢数的风险（leader：0，isr:0）

注意: acks： -1（all）,Leader和ISR队列里的所有节点收到数据后,在返回ack时Leader挂了,会出现数据重复问题

可靠性总结：
acks=0，生产者发送过来数据就不管了，可靠性差，效率高；
acks=1，生产者发送过来数据Leader应答，可靠性中等，效率中等；
acks=-1，生产者发送过来数据Leader和ISR队列里面所有Follwer应答，可靠性高，效率低；
在生产环境中，acks=0很少使用；acks=1，一般用于传输普通日志，允许丢个别数据；acks=-1，一般用于传输和钱相关的数据，对可靠性要求比较高的场景。

---

数据去重

• 至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2
• 最多一次（At Most Once）= ACK级别设置为0
• 总结：
  At Least Once可以保证数据不丢失，但是不能保证数据不重复；
  At Most Once可以保证数据不重复，但是不能保证数据不丢失。
• 精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失。
  Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务。

---

幂等性

幂等性就是指Producer不论向Broker发送多少次重复数据，Broker端都只会持久化一条(认为重复了)，保证了不重复
精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数>=2 + ISR最小副本数量>=2） 。

幂等性判断重复数据的标准: 具有**相同主键的消息提交时，Broker只会持久化一条。其 中PID是Kafka每次重启都会分配一个新的(挂掉重启就是新的)；Partition 表示分区号；Sequence Number是单调自增的**。
所以幂等性只能保证的是在单分区单会话内不重复。 开启参数 enable.idempotence 默认为 true

---

事务

开启事务，必须开启幂等性(默认开启)。且指定全局唯一的 transactional.id

• 精确一次为什么要用事务?

因为幂等性只能保证单分区单会话, 也就是如果主机挂了,PID就变了, 此时虽然ack是-1,但是Leader挂的时机又恰巧是前面所说的所有节点都接收到数据,在leader ack响应的时候挂调, 生产者就会重复发送消息,等待新的leader响应,此时幂等性就无法保证事务,即无法保证数据重复.

• 新成员概念

Transaction Coordinator :事务协调器,每个broker解决都有一个,专门处理事务
__transaction_state-分区-Leader: 存储事务信息的特殊主题, 用于存储事务在提交过程中需要持久化到硬盘的信息

• 如何判断一条事务消息交由哪个broker节点的Transaction Coordinator 进行管理?

特殊主题默认有50个分区，这些分区分散到多个broker节点上,每个分区负责一部分事务。事务划分是根据transactional.id的hashcode值%50，计算出该事务属于哪个分区。 该分区Leader副本所在的broker节点即为这个transactional.id对应的Transaction Coordinator节点。

• 事务原理

1. 请求 producer id（PID幂等性需要）
2. 事务协调器返回producer id (PID)
3. 发送消息到Topic
4. 先事务协调器发送commit请求.
5. 提交commit请求的消息一定会持久化到 __transaction_state-分区-Leader
6. 持久化完成后,告诉producer 成功了
7. 后台向Partition发送commit请求, 判断消息是不是正在发到Partition了
8. 返回成功事务协调器
9. 事务协调器持久化成功的信息到 __transaction_state-分区-Leader

• Java 事务API

// 1 初始化事务
void initTransactions();
// 2 开启事务
void beginTransaction() throws ProducerFencedException;
// 3 在事务内提交已经消费的偏移量（主要用于消费者）
void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
 String consumerGroupId) throws 
ProducerFencedException;
// 4 提交事务
void commitTransaction() throws ProducerFencedException;
// 5 放弃事务（类似于回滚事务的操作）
void abortTransaction() throws ProducerFencedException;

1
2
3
4
5
6
7
8
9
10
11
12
13

    public static void main(String[] args) {

        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "120.77.39.136:9092");

        // key,value 序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        // 设置事务 id（必须），事务 id 任意起名
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG,
                "transaction_id_0");
        // 3. 创建 kafka 生产者对象
        KafkaProducer<String, String> kafkaProducer = new
                KafkaProducer<String, String>(properties);
        // 初始化事务
        kafkaProducer.initTransactions();
        // 开启事务
        kafkaProducer.beginTransaction();
        try {
            // 4. 调用 send 方法,发送消息
            for (int i = 0; i < 5; i++) {
                // 发送消息
                kafkaProducer.send(new ProducerRecord<>("first",
                        "事务消息 " + i));
            }
// int i = 1 / 0;
            // 提交事务
            kafkaProducer.commitTransaction();
        } catch (Exception e) {
            // 终止事务
            kafkaProducer.abortTransaction();
        } finally {
            // 5. 关闭资源
            kafkaProducer.close();
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

---

数据有序和乱序

Kafka保证 一个partition 单分区内有序,多分区，分区与分区间无序

• 1）kafka在1.x版本之前保证数据单分区有序，条件如下：
  max.in.flight.requests.per.connection=1（不需要考虑是否开启幂等性）。
• 2）kafka在1.x及以后版本保证数据单分区有序，条件如下：
• （1）未开启幂等性
  max.in.flight.requests.per.connection需要设置为1。
• （2）开启幂等性
  max.in.flight.requests.per.connection需要设置小于等于5。
  原因说明：因为在kafka1.x以后，启用幂等后，kafka服务端会缓存producer发来的最近5个request的元数据，
  故无论如何，都可以保证最近5个request的数据都是有序的(根据幂等性单调递增SeqNumber),如果无需,先不落盘。

---

Kafka Broker

zookeeper

在zookeeper的服务端存储的Kafka相关信息：
1）/kafka/brokers/ids [0,1,2] 记录有哪些服务器
2）/kafka/brokers/topics/first/partitions/0/state
{“leader”:1 ,“isr”:[1,0,2] } 记录谁是Leader，有哪些服务器可用
3）/kafka/controller
{“brokerid”:0} 辅助选举Leader

---

总体工作流程

1. 每台kafka节点启动后,都会往zookeeper进行注册
2. 开始抢占Controller节点,谁先注册谁说了算 (每个broker都有一个controller,用于负责日后的leader选举工作)
3. 上任后,由选举出来的Controller监听brokers节点变化
4. Controller开始正在选举Leader
   选举规则：在isr中存活为前提，按照AR中排在前面的优先。例如ar[1,0,2], isr [1，0，2]，那么leader就会按照1，0，2的顺序轮询 (AR：Kafka分区中的所有副本统称)
5. 成功选举出leader后, Controller将节点信息上传到ZK (/brokers/topics/first/partitions/0/state)
6. 其他contorller从zk同步相关信息

---

7. 假设Broker 中Leader挂了,此时Controller就能因为(3)步骤能够监控到节点变化
8. 重新拉取因(5)上传的信息 (/brokers/topics/first/partitions/0/state)
9. 重新根据(4)步骤,重新选举Leader
10. 更新Leader及ISR,Controller将节点信息上传到ZK
    

---

Broker 重要参数

---

Kafka 副本

• 基本信息

（1）Kafka 副本作用：提高数据可靠性。
（2）Kafka 默认副本 1 个，生产环境一般配置为 2 个，保证数据可靠性；太多副本会增加磁盘存储空间，增加网络上数据传输，降低效率。
（3）Kafka 中副本分为：Leader 和 Follower。Kafka 生产者只会把数据发往 Leader，然后 Follower 找 Leader 进行同步数据。
（4）Kafka 分区中的所有副本统称为 AR（Assigned Repllicas）。AR = ISR + OSR

ISR，表示和 Leader 保持同步的 Follower 集合。如果 Follower 长时间未向 Leader 发送
通信请求或同步数据，则该 Follower 将被踢出 ISR。该时间阈值由 replica.lag.time.max.ms
参数设定，默认 30s。Leader 发生故障之后，就会从 ISR 中选举新的 Leader。
OSR，表示 Follower 与 Leader 副本同步时，延迟过多的副本。

Leader Partition 负载平衡

正常情况下，Kafka本身会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量都是均匀的。但是如果某 些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡。

---

增加副本因子

在生产环境当中，由于某个主题的重要等级需要提升，我们考虑增加副本。副本数的
增加需要先制定计划，然后根据计划执行。

1）创建 topic 设置副本为1

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh --bootstrap-server xxxx:9092 --create --partitions 3 --replication-factor 1 --topic four
1

2）手动增加副本存储
（1）创建副本存储计划（所有副本都指定存储在 broker0、broker1、broker2 中）。

[root@root kafka]$ vim increase-replication-factor.json
1

输入如下内容：

{"version":1,"partitions":[{"topic":"four","partition":0,"replicas":[0,1,2]},{"topic":"four","partition":1,"replicas":[0,1,2]},{"topic":"four","partition":2,"replicas":[0,1,2]}]}
1

（2）执行副本存储计划。

[root@root kafka]$ bin/kafka-reassign-partitions.sh --bootstrap-server xxxx:9092 --reassignment-json-file increase-replication-factor.json --execute
1

---

文件存储机制

Topic是逻辑上的概念，而partition是物理上的概念，每个partition对应于一个log文件，该log文件中存储的就是Producer生产的数据。Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制， 将每个partition分为多个segment。每个segment包括：“.index”文件、“.log”文件和.timeindex等文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。

.log 日志文件
.index 偏移量索引文件
.timeindex 时间戳索引文件
其他文件
index和log文件以当前segment的第一条消息的offset命名

• 通过工具查看 index 和 log 信息

kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.index
kafka-run-class.sh kafka.tools.DumpLogSegments --files ./00000000000000000000.log
1
2

• 稀疏索引

1、index为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。参数log.index.interval.bytes默认4kb
2、Index文件中保存的offset为相对offset（相对文件名的偏移），这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小

---

文件清理策略

Kafka 中默认的日志保存时间为 7 天，可以通过调整如下参数修改保存时间。

• log.retention.hours，最低优先级小时，默认 7 天。
• log.retention.minutes，分钟。
• log.retention.ms，最高优先级毫秒。
• log.retention.check.interval.ms，负责设置检查周期，默认 5 分钟。

那么日志一旦超过了设置的时间，怎么处理呢？
Kafka 中提供的日志清理策略有 delete 和 compact 两种。

1）delete 日志删除：将过期数据删除

• log.cleanup.policy = delete 所有数据启用删除策略
  （1）基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。
  （2）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的 segment。
  log.retention.bytes，默认等于-1，表示无穷大

2）compact 日志压缩
compact日志压缩：对于相同key的不同value值，只保留最后一个版本。

• log.cleanup.policy = compact 所有数据启用压缩策略

压缩后的offset可能是不连续的，比如上图中没有6，当从这些offset消费消息时，将会拿到比这个offset大 的offset对应的消息，实际上会拿到offset为7的消息，并从这个位置开始消费。
这种策略只适合特殊场景，比如消息的key是用户ID，value是用户的资料，通过这种压缩策略，整个消息集里就保存了所有用户最新的资料。

---

高效读写数据

1）Kafka 本身是分布式集群，可以采用分区技术，并行度高
2）读数据采用稀疏索引，可以快速定位要消费的数据
3）顺序写磁盘 ，Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端
4）页缓存 + 零拷贝技术：
    零拷贝：Kafka的数据加工处理操作交由Kafka生产者和Kafka消费者处理。Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高。
    PageCache页缓存：Kafka重度依赖底层操作系统提供的PageCache功 能。当上层有写操作时，操作系统只是将数据写入PageCache。当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。

---

Kafka 消费者

基础概念

Kafka采用的consumer采用从broker中主动pull拉取数据，每个消费者的offset由消费者提交到系统主题保存

Consumer Group（CG）：消费者组，由多个consumer组成。形成一个消费者组的条件，是所有消费者的groupid相同。

• 每个消费者组都必须指定一个group.id
• 消费者组内每个消费者负责消费不同分区的数据，一个分区只能由一个组内消费者消费。
• 消费者组之间互不影响。所有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

---

消费者组初始化流程

coordinator：辅助实现消费者组的初始化和分区的分配。

coordinator节点选择 = groupid的hashcode值 % 50（ __consumer_offsets的分区数量）
例如： groupid的hashcode值 = 1，1% 50 = 1，那么__consumer_offsets 主题的1号分区，在哪个broker上，就选择这个节点的coordinator作为这个消费者组的老大。消费者组下的所有的消费者提交offset的时候就往这个分区去提offset。

1）消费者组下的所有consumer都会给改组%50后选出的对应Coordinator发送请求
2）Coordinator选出这组中的一个consumer作为leader
3）Coordinator将所有消费者要消费的topic情况发送给leader
4）leader会负责制定消费方案
5）leader将制定好的计划发送给Coordinator
6）Coordinator就把消费计划下发给各个consumer
7）每个消费者都会和coordinator保持心跳（默认3s），一旦超时（session.timeout.ms=45s），该消费者会被移除，并触发再平衡；或者消费者处理消息的时间过长（max.poll.interval.ms5分钟），也会触发再平衡

---

详细消费流程

---

消费消息ApI

• 普通消费

public class CustomConsumer {
    public static void main(String[] args) {
        // 1.创建消费者的配置对象
        Properties properties = new Properties();
        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "xxx:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        // 配置消费者组（组名任意起名） 必须
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 创建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
        // 注册要消费的主题（可以消费多个主题）
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        // 拉取数据打印
        while (true) {
            // 设置 1s 中消费一批数据
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            // 打印消费到的数据
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

• 指定分区消费

特别注意：即便消费者组名相同，但是也不参与Kafka Group 管理，也就是存在2次消费的情况

public class TopicPartitionConsumer {
    public static void main(String[] args) {
        Properties properties = new Properties();

        // 2.给消费者配置对象添加参数
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "xxx:9092");

        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        // 配置消费者组（必须），名字可以任意起
        properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        // 消费某个主题的某个分区数据
        ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
        topicPartitions.add(new TopicPartition("first", 1));
        kafkaConsumer.assign(topicPartitions);
        while (true){
            ConsumerRecords<String, String> consumerRecords =
                    kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord :
                    consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

---

分区分配策略

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

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

• Range 以及再平衡

首先对同一个 topic 里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。
通过 partitions数/consumer数 来决定每个消费者应该消费几个分区。如果除不尽，那么前面几个消费者将会多消费 1 个分区。
当C0消费多个Topic时，永远多消费一个分区，容易产生数据倾斜！

        假设 0 号消费者挂掉后，消费者组需要按照超时时间 45s 来判断它是否退出，所以需
要等待，时间到了 45s 后，判断它真的退出就会把任务分配给其他 broker 执行。
重新按照 range 方式分配：
    1 号消费者：消费到 0、1、2、3 号分区数据。
    2 号消费者：消费到 4、5、6 号分区数据。

• RoundRobin 以及再平衡

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

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.RoundRobinAssignor");
1

再平衡后：
1 号消费者：消费到 0、2、4、6 号分区数据
2 号消费者：消费到 1、3、5 号分区数据

• Sticky 以及再平衡

首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化。

properties.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients.consumer.StickyAssignor");
1

例如：
原来
0 号消费者：消费到 0、1 号分区数据。
1 号消费者：消费到 2、5、3 号分区数据。
2 号消费者：消费到 4、6 号分区数据。

0号消费者挂了再平衡（45s）
1 号消费者：消费到 2、3、5 号分区数据。
2 号消费者：消费到 0、1、4、6 号分区数据。

---

offset 位移

从0.9版本开始，consumer默认将offset保存在Kafka一个内置的topic中，该topic为__consumer_offsets
Kafka0.9版本之前，consumer默认将offset保存在Zookeeper中。

__consumer_offsets 主题里面采用 key 和 value 的方式存储数据。
key 是 group.id+topic+分区号，value 就是当前 offset 的值每隔一段时间，kafka 内部会对这个 topic 进行compact，也就是每个 group.id+topic+分区号就保留最新数据。

• 自动提交offset

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

• 手动提交 offset

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

是否自动提交 offset :false
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
同步提交 offset
consumer.commitSync();
异步提交 offset
consumer.commitAsync();

public class CommitSync {

    public static void main(String[] args) {
        // 1. 创建 kafka 消费者配置类
        Properties properties = new Properties();
        // 2. 添加配置参数
        // 添加连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "hadoop102:9092");
        // 配置序列化 必须
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        // 配置消费者组

        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 是否自动提交 offset
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //3. 创建 kafka 消费者
        KafkaConsumer<String, String> consumer = new
                KafkaConsumer<>(properties);
        //4. 设置消费主题 形参是列表
        consumer.subscribe(Arrays.asList("first"));
        //5. 消费数据
        while (true) {
            // 读取消息
            ConsumerRecords<String, String> consumerRecords =
                    consumer.poll(Duration.ofSeconds(1));
            // 输出消息
            for (ConsumerRecord<String, String> consumerRecord :
                    consumerRecords) {
                System.out.println(consumerRecord.value());
            }
            // 同步提交 offset
            consumer.commitSync();
            // 异步提交 offset
            consumer.commitAsync();
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

• 指定 Offset 消费

auto.offset.reset = earliest | latest | none 默认是 latest。
（1）earliest：自动将偏移量重置为最早的偏移量，–from-beginning。
（2）latest（默认值）：自动将偏移量重置为最新偏移量。
（3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

注意：每次执行完，需要修改消费者组名；

public class CustomConsumerSeek {

    public static void main(String[] args) {
        // 0 配置信息
        Properties properties = new Properties();
        // 连接
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "120.77.39.136:9092");
        // key value 反序列化

        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());

        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test2");
        // 1 创建一个消费者
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);

        // 2 订阅一个主题
        ArrayList<String> topics = new ArrayList<>();
        topics.add("first");
        kafkaConsumer.subscribe(topics);
        
        Set<TopicPartition> assignment= new HashSet<>();
        while (assignment.size() == 0) {
            kafkaConsumer.poll(Duration.ofSeconds(1));
            // 获取消费者分区分配信息（有了分区分配信息才能开始消费）
            assignment = kafkaConsumer.assignment();
        }
        // 遍历所有分区，并指定 offset 从 1700 的位置开始消费
        for (TopicPartition tp: assignment) { 
            kafkaConsumer.seek(tp, 1700);
        }
        // 3 消费该主题数据
        while (true) {
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

数据积压

消费者如何提高吞吐量
1）增 加Topic的分区数，让消费者数 = 分区数
2）提高每批次拉取的数量。

SpringBoot整合

依赖pom

	<dependency>
       <groupId>org.springframework.bootgroupId>
       <artifactId>spring-boot-starter-webartifactId>
   dependency>

   <dependency>
       <groupId>org.springframework.kafkagroupId>
       <artifactId>spring-kafkaartifactId>
   dependency>
1
2
3
4
5
6
7
8
9

application.yml配置文件

server:
  port: 8080
spring:
  application:
    name: kafka-demo

  kafka:
    bootstrap-servers: xxx:9092
    producer: # producer 生产者
      retries: 0 # 重试次数
      acks: -1 # 应答级别:多少个分区副本备份完成时向生产者发送ack确认(可选0、1、all/-1)
      batch-size: 16384 # 批量大小
      buffer-memory: 33554432 # 生产端缓冲区大小
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      #      value-serializer: com.itheima.demo.config.MySerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

    consumer: # consumer消费者
      group-id: test # 默认的消费组ID
      enable-auto-commit: true # 是否自动提交offset
      auto-commit-interval: 500  # 提交offset延时(接收到消息后多久提交offset) 默认 5000ms  5秒

      # earliest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
      # latest:当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
      # none:topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
      auto-offset-reset: latest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      #      value-deserializer: com.itheima.demo.config.MyDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

生产消费者实现

同步+异步+回调全整上

KafkaProducer

@RestController
public class KafkaProducer {
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    @GetMapping("/kafka/test/{msg}")
    public void sendMessage(@PathVariable("msg") String msg) throws InterruptedException, ExecutionException, TimeoutException {

        kafkaTemplate.send(new ProducerRecord<>("first", msg)).addCallback(success -> {
            // 消息发送到的topic
            RecordMetadata recordMetadata = success.getRecordMetadata();
            ProducerRecord<String, Object> producerRecord = success.getProducerRecord();

            System.out.println(" 单次监听主题： " +
                    recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());

        }, failure -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        });
		/**设置全局 **/
		/*
		kafkaTemplate.send(new ProducerRecord<>("first", msg)).addCallback( success ->{
            // 消息发送到的topic
            RecordMetadata recordMetadata = success.getRecordMetadata();
            ProducerRecord producerRecord = success.getProducerRecord();

            System.out.println(" 全局监听主题： " +
                    recordMetadata.topic() + "->" + "分区：" + recordMetadata.partition());

        },failure  -> {
            System.out.println("发送消息失败:" + failure.getMessage());
        }); 
        */

        /**同步阻塞 **/
//        ListenableFuture> future = kafkaTemplate.send(new ProducerRecord<>("first", msg));
//        //注意，可以设置等待时间，超出后，不再等候结果
//        SendResult result = future.get(3,TimeUnit.SECONDS);

    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

KafkaConsumer

@Component
public class KafkaConsumer {

    //不指定group，默认取yml里配置的
    @KafkaListener(topics = {"first"})
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord) {
        System.out.println(consumerRecord);
    }
}
1
2
3
4
5
6
7
8
9