本文章以小破某站的某硅谷资料为入门基础，再配合个人理解整理的资料，若涉及等问题，请及时联系我并删除整改。

借鉴资料：Spring-Boot整合kafka - CSDN资料

架构

1）Producer **：**消息生产者，就是向 kafka broker 发消息的客户端；

2）Consumer **：**消息消费者，向 kafka broker 取消息的客户端；

3）Consumer Group （CG）：**消费者组，由多个 consumer 组成。**消费者组内每个消费者负

**责消费不同分区的数据，一个分区只能由一个组内消费者消费；消费者组之间互不影响。**所

有的消费者都属于某个消费者组，即消费者组是逻辑上的一个订阅者。

4）Broker **：**一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker

可以容纳多个 topic。

5）Topic **：**可以理解为一个队列，生产者和消费者面向的都是一个 topic；

6）Partition：为了实现扩展性，一个非常大的 topic 可以分布到多个 broker（即服务器）上，

一个 topic 可以分为多个 partition，每个 partition 是一个有序的队列；

7）Replica：副本，为保证集群中的某个节点发生故 障时，该节点上的 partition 数据不丢失，

且 kafka 仍然能够继续工作，kafka 提供了副本机制，一个 topic 的每个分区都有若干个副本，

一个 leader 和若干个 follower。

8）leader：**每个分区多个副本的“主”，生产者发送数据的对象，以及消费者消费数据的对

象都是 leader。

9）follower：每个分区多个副本中的“从”，实时从 leader 中同步数据，保持和 leader 数据

的同步。leader 发生故障时，某个 follower 会成为新的 follower。

工作流程

topic 是逻辑上的概念，而 partition 是物理上的概念

每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据

Producer 生产的数据会被不断追加到该log 文件末端，且每条数据都有自己的 offset。

消费者组中的每个消费者，都会实时记录自己消费到了哪个 offset，以便出错恢复时，从上次的位置继续消费

“.index”文件存储大量的索引信息，

“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中 message 的物理偏移地址

命令行操作

查看操作主题topic命令参数

[atguigu@hadoop102 kafka]$ bin/kafka-topics.sh  其他操作 --topic 主题名 其他操作
1

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

生产者命令参数

发送消息

[atguigu@hadoop102 kafka]$ bin/kafka-console-producer.sh --bootstrap-server hadoop102:9092 --topic first
>hello world
>atguigu atguigu
1
2
3

消费者命令参数

消费信息

（1）消费 first 主题中的数据
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic first
（2）把主题中所有的数据都读取出来（包括历史数据）
[atguigu@hadoop102 kafka]$ bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --from-beginning --topic first
1
2
3
4

生产者

发生原理

两个线程—— main 线程和 Sender 线程

在 main 线程中创建了一个双端队列 RecordAccumulator。main 线程将消息发送给 RecordAccumulator，

Sender 线程不断从 RecordAccumulator 中拉取消息发送到 Kafka Broker

只需在异步发送的基础上，再调用一下 get()方法即可。

package com.atguigu.kafka.producer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class CustomProducerSync {

	public static void main(String[] args) throws
		InterruptedException, ExecutionException {
      // 1. 创建 kafka 生产者的配置对象
      Properties properties = new Properties();
      // 2. 给 kafka 配置对象添加配置信息

      properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102:9092");
      // key,value 序列化（必须）：key.serializer，value.serializer
      properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

      properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, 
      // 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");               
                    
      StringSerializer.class.getName());
      // 3. 创建 kafka 生产者对象
      KafkaProducer<String, String> kafkaProducer = new 
      KafkaProducer<String, String>(properties);
      // 4. 调用 send 方法,发送消息
      for (int i = 0; i < 10; i++) {
        // 异步发送 默认
        kafkaProducer.send(new ProducerRecord<>("first","kafka" + i), 
          // 1.添加回调
          new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
              if (exception == null) {
                // 没有异常,输出信息到控制台
                System.out.println(" 主题： " + metadata.topic() + "->" + "分区：" + metadata.partition());
              } else {
                // 出现异常打印
                exception.printStackTrace();
              }
            }
          }
        );
        // 2.同步发送
        kafkaProducer.send(new ProducerRecord<>("first","kafka" + i)).get();
      }
 		// 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
43
44
45
46
47
48
49
50
51
52
53
54

kafkaTemplate提供了一个回调方法addCallback，我们可以在回调方法中监控消息是否发送成功 或 失败时做补偿处理，有两种写法

生产者发送代码示例

@RestController
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    // 发送消息
    @GetMapping("/kafka/callbackOne/{message}")
		public void sendMessage2(@PathVariable("message") String callbackMessage) {
      kafkaTemplate.send("topic1", callbackMessage).addCallback(success -> {
          // 消息发送到的topic
          String topic = success.getRecordMetadata().topic();
          // 消息发送到的分区
          int partition = success.getRecordMetadata().partition();
          // 消息在分区内的offset
          long offset = success.getRecordMetadata().offset();
          System.out.println("发送消息成功:" + topic + "-" + partition + "-" + offset);
      }, failure -> {
          System.out.println("发送消息失败:" + failure.getMessage());
      });
	}
  
  @GetMapping("/kafka/callbackTwo/{message}")
  public void sendMessage3(@PathVariable("message") String callbackMessage) {
    kafkaTemplate.send("topic1", callbackMessage).addCallback(
      new ListenableFutureCallback<SendResult<String, Object>>() {
        @Override
        public void onFailure(Throwable ex) {
          System.out.println("发送消息失败："+ex.getMessage());
        }

        @Override
        public void onSuccess(SendResult<String, Object> result) {
          System.out.println("发送消息成功：" + result.getRecordMetadata().topic() + "-"
                             + result.getRecordMetadata().partition() + "-" + 
                             result.getRecordMetadata().offset());
        }
    });
  }

}
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

分区策略

方便在集群中扩展，提高并发

需要将 producer 发送的数据封装成一个 ProducerRecord 对象

public ProducerRecord(String topic, Integer partition, Long timestamp, K key, V value, Iterable<Header> headers) {	... ... }
1

1. 指明 partition 的情况下，直接将指明的值直接作为 partiton 值；

2. 没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition 数进行取余得到 partition 值；

3. 既没有 partition 值又没有 key 值的情况下，**采用Sticky Partition（黏性分区器）**第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition 值，也就是常说的 round-robin 算法。

   并尽可能一直使用该分区，待该分区的 batch 已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）

• 自定义分区器

定义类实现 Partitioner 接口，重写 partition()方法

package com.atguigu.kafka.producer;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;

/**
* 1. 实现接口 Partitioner
* 2. 实现 3 个方法:partition(返回信息对应的分区),close(关闭资源),configure(配置方法)
* 3. 编写 partition 方法,返回分区号
*/
public class MyPartitioner implements Partitioner {

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

使用分区器的方法，在生产者的配置中添加分区器参数

Properties properties = new Properties();
....
// 添加自定义分区器
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.atguigu.kafka.producer.MyPartitioner");
1
2
3
4

或者，在application.propertise中配置自定义分区器，配置的值就是分区器类的全路径名

# 自定义分区器
spring.kafka.producer.properties.partitioner.class=com.atguigu.kafka.producer.MyPartitioner
1
2

可靠性

**ack 应答机制 **

topic 的每个 partition 收到

producer 发送的数据后，都需要向 producer 发送 ack（acknowledgement 确认收到），如果

producer 收到 ack，就会进行下一轮的发送，否则重新发送数据

replica.lag.time.max.ms 参数设定。Leader 发生故障之后，就会从 ISR 中选举新的 leader。

Leader 维护了一个动态的 in-sync replica set (ISR)，意为和 leader 保持同步的 follower 集合。

当 ISR 中的 follower 完成数据的同步之后，leader 就会给 follower 发送 ack。

如果 follower长时间 未 向 leader 同 步 数 据 ， 则 该 follower 将 被 踢 出 ISR 。

生产者参数

对数据的可靠性要求不是很高，没必要等 ISR 中的 follower 全部接收成功，Kafka 为用户提供了三种可靠性级别

// 设置 acks
 properties.put(ProducerConfig.ACKS_CONFIG, "all");
 // 重试次数 retries，默认是 int 最大值，2147483647
 properties.put(ProducerConfig.RETRIES_CONFIG, 3);
1
2
3
4

幂等性

数据传递语义

至少一次（At Least Once）= ACK级别设置为-1 + 分区副本大于等于2 + ISR里应答的最小副本数量大于等于2

​ 保证数据不丢失，但是不能保证数据不重复

最多一次（At Most Once）= ACK级别设置为0，保证数据不重复，但是不能保证数据不丢失

精确一次（Exactly Once）：对于一些非常重要的信息，比如和钱相关的数据，要求数据既不能重复也不丢失

Kafka 0.11版本以后，引入了一项重大特性：幂等性和事务

精确一次（Exactly Once） = 幂等性 + 至少一次（ ack=-1 + 分区副本数 >=2 + ISR 最小副本数量 >=2 ）

重复数据的判断标准：具有相同主键的消息提交时，Broker只会持久化一条。

其 中PID是Kafka每次重启都会分配一个新的；Partition 表示分区号；Sequence Number是单调自增的

• 使用

开启参数 enable.idempotence 默认为 true，false 关闭

生产者事务

开启事务，必须开启幂等性

Producer 在使用事务功能前，必须先自定义一个唯一的 transactional.id。

有 了 transactional.id，即使客户端挂掉了，它重启后也能继续处理未完成的事务

// 1. 创建 kafka 生产者的配置对象
Properties properties = new Properties();
// 2. 给 kafka 配置对象添加配置信息
// 给 kafka 配置对象添加配置信息：bootstrap.servers
 properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102: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", "atguigu " + 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

• 整合spring boot

生产者事务示例代码，使用 KafkaTemplate 的 executeInTransaction 方法来声明事务

@GetMapping("/kafka/transaction")
public void sendMessage7(){
    // 声明事务：后面报错消息不会发出去
    kafkaTemplate.executeInTransaction(operations -> {
        operations.send("topic1","test executeInTransaction");
        throw new RuntimeException("fail");
    });
    // 不声明事务：后面报错但前面消息已经发送成功了
   kafkaTemplate.send("topic1","test executeInTransaction");
   throw new RuntimeException("fail");
}
1
2
3
4
5
6
7
8
9
10
11

Kafka 的事务一共有如下 5 个 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

数据有序

单分区内，有序

多分区，分区与分区间无序

负载平衡

正常下，Kafka会自动把Leader Partition均匀分散在各个机器上，来保证每台机器的读写吞吐量均匀

• 特殊情况

但是如果某 些broker宕机，会导致Leader Partition过于集中在其他少部分几台broker上，这会导致少数几台broker的读写请求压力过高，

其他宕机的broker重启之后都是follower partition，读写请求很低，造成集群负载不均衡

Kafka Broker

一个kafka运行实例叫做一个broker

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）启动 Zookeeper 客户端。

[atguigu@hadoop102 zookeeper-3.5.7]$ bin/zkCli.sh
1

（2）通过 ls 命令可以查看 kafka 相关信息。

[zk: localhost:2181(CONNECTED) 2] ls /kafka
1

broker总体工作流程

Broker参数

副本

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选举流程

Kafka 集群中有一个 broker 的 Controller 会被选举为**Controller Leader**，负责管理集群broker 的上下线，所有 topic 的分区副本分配和 Leader 选举等工作。

Controller 的信息同步工作是依赖于 **Zookeeper **的。

故障处理细节

Log文件中的HW和LEO

LEO：每个副本的最后一个offset

HW：指的是消费者能见到的最大的offset，ISR 队列中最小的 LEO

1）Follower 故障

1. Follower发生故障后会被临时踢出ISR
2. 这个期间Leader和Follower继续接收数据
3. 待该Follower恢复后，Follower会读取本地磁盘记录的上次的HW，并将log文件高于HW的部分截取，从HW开始向Leader进行同步
4. 等该Follower的LEO大于等于该Partition的HW，即Follower追上Leader之后，就可以重新加入ISR了

2）Leader 故障

1. Leader发生故障之后，会从ISR中选出一个新的Leader
2. 为保证多个副本之间的数据一致性，其余的Follower会先将各自的log文件高于HW的部分截掉，然后从新的Leader同步数据

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

文件存储

高效读写数据

1）Kafka本身是分布式集群，可以采用分区技术，并行度高

2）读数据采用稀疏索引，可以快速定位要消费的数据

3）顺序写磁盘

4）页缓存 + 零拷贝技术

分区

Topic是逻辑上的概念，而 partition 是物理上的概念，每个partition对应于一个 log 文件，该log文件中存储的就是Producer生产的数据。

Producer生产的数据会被不断追加到该log文件末端，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制

将每个partition分为多个segment。每个segment包括：“.index文件，.log文件和.timeindex等文件“。

这些文件位于一个文件夹下，该文件夹的命名规则为：topic名称+分区序号，例如：first-0。

.log 日志文件

.index 偏移量索引文件

.timeindex 时间戳索引文件

Log文件和Index文件详解

根据目标offset定位Segment文件

找到小于等于目标offset的最大offset对应的索引项

定位到log文件

向下遍历找到目标Record

1. index为稀疏索引，大约每往log文件写入4kb数据，会往index文件写入一条索引。参log.index.interval.bytes默认4kb

2. Index文件中保存的offset为相对offset，这样能确保offset的值所占空间不会过大，因此能将offset的值控制在固定大小

日志存储参数配置

文件清理策略

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

日志清理策略有 delete 和 compact 两种

• delete 日志删除

将过期数据删除，log.cleanup.policy = delete 所有数据启用删除策略

（1）基于时间：默认打开。以 segment 中所有记录中的最大时间戳作为该文件时间戳。

（2）基于大小：默认关闭。超过设置的所有日志总大小，删除最早的 segment。

​ log.retention.bytes，默认等于-1，表示无穷大。

• compact 日志压缩

对于相同key的不同value值，只保留最后一个版本，压缩后的offset可能是不连续的

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

顺序写磁盘

Kafka 的 producer 生产数据，要写入到 log 文件中，写的过程是一直追加到文件末端，为顺序写。

省去了大量磁头寻址的时间

页缓存 + 零拷贝

零拷贝：Kafka Broker应用层不关心存储的数据，所以就不用走应用层，传输效率高

PageCache页缓存

Kafka重度依赖底层操作系统提供的PageCache功 能。

当上层有写操作时，操作系统只是将数据写PageCache。

当读操作发生时，先从PageCache中查找，如果找不到，再去磁盘中读取。

实际上PageCache是把尽可能多的空闲内存都当做了磁盘缓存来使用。

消费者

消费方式

consumer 采用 pull（拉）模式从 broker 中读取数据。

pull模式不足之处是，如果Kafka没有数据，消费者可能会陷入循环中，一直返回空数据。

消费者组

Consumer Group（CG）：消费者组，由多个consumer组成。

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

消费者参数

注意：

在消费者 API 代码中必须配置消费者组 id。

命令行启动消费者不填写消费者组id 会被自动填写随机的消费者组 id。

package com.atguigu.kafka.consumer;
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 java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Properties;
public class CustomConsumerPartition {
  public static void main(String[] args) {
    // 创建消费者的配置对象
    Properties properties = new Properties();
 		// 给消费者配置对象添加参数
  	properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"hadoop102: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 topics = new ArrayList<>();
      topics.add("first");
      kafkaConsumer.subscribe(topics);
    */

    // 消费某个主题的某个分区数据
    ArrayList<TopicPartition> topicPartitions = new ArrayList<>();
    topicPartitions.add(new TopicPartition("first", 0));
    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
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

• 整合spring boot

批量消费

设置application.prpertise开启批量消费即可

# 设置批量消费
spring.kafka.listener.type=batch
# 批量消费每次最多消费多少条消息
spring.kafka.consumer.max-poll-records=50
1
2
3
4

接收消息时用 **List **来接收，监听代码如下

@KafkaListener(id = "consumer2",groupId = "felix-group", topics = "topic1")
public void onMessage3(List<ConsumerRecord<?, ?>> records) {
    System.out.println(">>>批量消费一次，records.size()="+records.size());
    for (ConsumerRecord<?, ?> record : records) {
        System.out.println(record.value());
    }
}
1
2
3
4
5
6
7

自定义异常处理器

通过ConsumerAwareListenerErrorHandler 异常处理器，可处理consumer在消费时发生的异常

新建一个 类型异常处理方法，用@Bean注入放到@KafkaListener注解的errorHandler属性里面，当监听抛出异常的时候，则会自动调用异常处理器

// 新建一个异常处理器，用@Bean注入
@Bean
public ConsumerAwareListenerErrorHandler consumerAwareErrorHandler() {
    return (message, exception, consumer) -> {
        System.out.println("消费异常："+message.getPayload());
        return null;
    };
}

// 将这个异常处理器的BeanName放到@KafkaListener注解的errorHandler属性里面
@KafkaListener(topics = {"topic1"},errorHandler = "consumerAwareErrorHandler")
public void onMessage4(ConsumerRecord<?, ?> record) throws Exception {
    throw new Exception("简单消费-模拟异常");
}

// 批量消费也一样，异常处理器的message.getPayload()也可以拿到各条消息的信息
@KafkaListener(topics = "topic1",errorHandler="consumerAwareErrorHandler")
public void onMessage5(List<ConsumerRecord<?, ?>> records) throws Exception {
    System.out.println("批量消费一次...");
    throw new Exception("批量消费-模拟异常");
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

分区分配以及平衡

一个consumer group中有多个consumer组成，一个 topic有多个partition组成，

现在的问题是，到底由哪个consumer来消费哪个partition的数据。

Range

对同一个 topic 里面的分区按照序号进行排序，并对消费者按照字母顺序进行排序。

通过 partitions数 / consumer数 来决定每个消费者应该消费几个分区。

如果除不尽，那么前面几个消费者将会多消费 1 个分区。

注意：如果只是针对 1 个 topic 而言，C0消费者多消费1个分区影响不是很大。

但是如果有 N 多个 topic，那么针对每个 topic，消费者 C0都将多消费 1 个分区，topic越多，C0消费的分区会比其他消费者明显多消费 N 个分区。

容易产生数据倾斜！

RoundRobi

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

Sticky

粘性分区：即在执行一次新的分配之前，考虑上一次分配的结果，尽量少的调整分配的变动，可节省开销

Kafka 从 0.11.x 版本开始引入首先会尽量均衡的放置分区到消费者上面，在出现同一消费者组内消费者出现问题的时候，会尽量保持原有分配的分区不变化

offset位移

Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，

consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为**__consumer_offsets**。

__consumer_offsets 主题里面采用 key 和 value 的方式存储数据。

key 是 group.id + topic + 分区号，value 就是当前 offset 的值。

每隔一段时间，kafka 内部会对这个 topic 进行compact，也就是每个 group.id+topic+分区号就保留最新数

• 参数配置

在配置文件 config/consumer.properties 中添加配置

自动提交

自动提交offset的相关参数：

手动提交

commitSync（同步提交）：必须等待offset提交完毕，再去消费下一批数据。

commitAsync（异步提交） ：发送完提交offset请求后，就开始消费下一批数据了

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

// 1. 创建 kafka 消费者配置类
Properties properties = new Properties();
// 2. 添加配置参数
	....
// 是否自动提交 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();
}


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

异步提交offset

// 其他配置与同步的
// 异步提交 offset
 consumer.commitAsync();
1
2
3

指定Offset 消费

auto.offset.reset = earliest | latest | none 默认是 latest

当 Kafka 中没有初始偏移量（消费者组第一次消费）或服务器上不再存在当前偏移量时（例如该数据已被删除），该怎么办？

（1）earliest：自动将偏移量重置为最早的偏移量，–from-beginning

（2）latest（默认值）：自动将偏移量重置为最新偏移量

（3）none：如果未找到消费者组的先前偏移量，则向消费者抛出异常。

（4）任意指定 offset 位移开始消费

指定时间消费

例如要求按照时间消费前一天的数据

package com.atguigu.kafka.consumer;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
public class CustomConsumerForTime {
	public static void main(String[] args) {
    // 0 配置信息
    Properties properties = new Properties();
    // 连接
    properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop102:9092");
    // key value 反序列化
    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");
    
    // 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();
 		}
		HashMap<TopicPartition, Long> timestampToSearch = new HashMap<>();
		// 封装集合存储，每个分区对应一天前的数据
		for (TopicPartition topicPartition : assignment) {
			timestampToSearch.put(topicPartition, 
				System.currentTimeMillis() - 1 * 24 * 3600 * 1000);
 		}
		// 获取从 1 天前开始消费的每个分区的 offset
  	Map<TopicPartition, OffsetAndTimestamp> offsets = 
      kafkaConsumer.offsetsForTimes(timestampToSearch);
		// 遍历每个分区，对每个分区设置消费时间。
 		for (TopicPartition topicPartition : assignment) {
 			OffsetAndTimestamp offsetAndTimestamp = offsets.get(topicPartition);
 			// 根据时间指定开始消费的位置
 			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);
      }
    }
  }
}
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
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

Kafka-Eagle监控

Kafka-Eagle 框架可以监控 Kafka 集群的整体运行情况，在生产环境中经常使用。

可以在浏览器打开 localhost:8084 登录页面查看监控数据

Kafka-Eagle 的安装依赖于 MySQL，MySQL 主要用来存储可视化展示的数据

官网：https://www.kafka-eagle.org/

上传压缩包 kafka-eagle-bin-2.0.8.tar.gz 到集群 **/opt/**software 目录

Kafka-Kraft 模式

Kafka 现有架构，元数据在 zookeeper 中，运行时动态选举 controller，由controller 进行 Kafka 集群管理。

kraft 模式架构（实验性），不再依赖 zookeeper 集群，而是用三台 controller 节点代替 zookeeper，元数据保存在 controller 中，由 controller 直接进行 Kafka 集群管理

好处有以下几个：

1. Kafka 不再依赖外部框架，而是能够独立运行；

2. controller 管理集群时，不再需要从 zookeeper 中先读取数据，集群性能上升；

3. 由于不依赖 zookeeper，集群扩展时不再受到 zookeeper 读写能力限制；

4. controller 不再动态选举，而是由配置文件规定。这样我们可以有针对性的加强

5. controller 节点的配置，而不是像以前一样对随机 controller 节点的高负载束手无策。

Spring Boot继承kafka

AdminClient API：允许管理和检测Topic、broker以及其它Kafka对象

Producer API：发布消息到1个或多个topic

Consumer API：订阅一个或多个topic,并处理产生的消息

Streams API：高效地将输入流转换到输出流

Connector API：从一些源系统或应用程序中拉取数据到kafka

@kafkaListener

直接依赖

		<dependency>
			<groupId>org.apache.kafkagroupId>
			<artifactId>kafka_2.13artifactId>
			<version>2.7.0version>
		dependency>
		
		<dependency>
			<groupId>org.apache.kafkagroupId>
			<artifactId>kafka-clientsartifactId>
			<version>2.7.0version>
		dependency>
		
		<dependency>
			<groupId>io.mosipgroupId>
			<artifactId>kafka-admin-clientartifactId>
			<version>1.2.0version>
			<exclusions>
				<exclusion>
					<groupId>org.apache.kafkagroupId>
					<artifactId>kafka-clientsartifactId>
				exclusion>
			exclusions>
		dependency>

		<dependency>
			<groupId>org.apache.kafkagroupId>
			<artifactId>kafka-server-commonartifactId>
			<version>3.1.0version>
			<exclusions>
				<exclusion>
					<groupId>org.apache.kafkagroupId>
					<artifactId>kafka-clientsartifactId>
				exclusion>
			exclusions>
		dependency>
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

adminClient

The AdminClient API 直接管理检查topic，brokers，acls等

创建Topic
createTopics(Collection<NewTopic> newTopics)
删除Topic：
deleteTopics(Collection<String> topics)
罗列所有Topic
listTopics()
查询Topic
describeTopics(Collection<String> topicNames)
查询集群信息
describeCluster()
查询ACL信息：
describeAcls(AclBindingFilter filter)
创建ACL信息：
createAcls(Collection<AclBinding> acls)
删除ACL信息：
deleteAcls(Collection<AclBindingFilter> filters)
查询配置信息：
describeConfigs(Collection<ConfigResource> resources)
修改配置信息：
alterConfigs(Map<ConfigResource, Config> configs)
修改副本的日志目录：
alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment)
查询节点的日志目录信息：
describeLogDirs(Collection<Integer> brokers)
查询副本的日志目录信息：
describeReplicaLogDirs(Collection<TopicPartitionReplica> replicas)
增加分区：
createPartitions(Map<String, NewPartitions> newPartitions)
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

简单使用adminclient创建topic实例

private static final String NEW_TOPIC = "topic-test2";
private static final String brokerUrl = "localhost:9092";
private static final Integer numPartition = 4;
private static final short replication = 1;
private static AdminClient adminClient;

@BeforeClass
public static void beforeClass(){
  Properties properties = new Properties();
  properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
  adminClient = AdminClient.create(properties);
}

@AfterClass
public static void afterClass(){
  log.info("Kafka-Admin-client开始被关闭");
  adminClient.close();
  log.info("Kafka-Admin-client已经被关闭了");
}

@Test
public void createTopics() {
  /*
  	NewTopic newTopic = new NewTopic(NEW_TOPIC, numPartition, replication);
    Collection newTopicList = new ArrayList<>();
    newTopicList.add(newTopic);
  	Arrays.asList(new NewTopic(topic, numPartition, replication))
  	CreateTopicsResult topics = adminClient.createTopics(newTopicList);
  	*/
  CreateTopicsResult topics = kafkaAdminClient.createTopics(Arrays.asList(
    new NewTopic(NEW_TOPIC, numPartition, replication)));
  /*
  KafkaFuture future = topics.all();
  while (!kafkaFuture.isDone()){
    try {
      Thread.sleep(1000L);
    } catch (InterruptedException e) {
      log.error("被线程中断信号干扰导致停止睡眠");
      e.printStackTrace();
    }
  }
  future.get();
  */
  log.debug("创建topic的请求成功");
}
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
44
45

springboot整合zookeeper

curator是Netflix公司开源的一个 zookeeper客户端，后捐献给 apache，curator框架在zookeeper原生API接口上进行了包装，解决了很多zooKeeper客户端非常底层的细节开发。

提供zooKeeper各种应用场景(比如:分布式锁服务、集群领导选举、共享计数器、缓存机制、分布式队列等的抽象封装，实现了Fluent风格的APl接口，是最好用，最流行的zookeeper的客户端

• 原生zookeeperAPI的不足

​ 连接对象异步创建，需要开发人员自行编码等待
​ 连接没有自动重连超时机制
​ watcher一次注册生效一次
​ 不支持递归创建树形节点

• curator特点

​ 解决session会话超时重连
​ watcher反复注册
​ 简化开发api
​ 遵循Fluent风格API

引入依赖

<dependency>
  <groupId>org.apache.zookeepergroupId>
  <artifactId>zookeeperartifactId>
  <version>3.6.2version>
dependency>
<dependency>
  <groupId>org.apache.curatorgroupId>
  <artifactId>curator-frameworkartifactId>
  <version>4.0.1version>
dependency>
<dependency>
  <groupId>org.apache.curatorgroupId>
  <artifactId>curator-recipesartifactId>
  <version>4.0.1version>
dependency>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16