Kafka超全精讲（一）

柔光的暖阳◎ 2023-01-03 16:19 278阅读 0赞

学习目标：

*  了解消息队列的应用场景
 *  能够搭建Kafka集群
 *  能够完成生产者、消费者Java代码编写
 *  理解Kafka的架构，以及Kafka的重要概念
 *  了解Kafka的事务

Kafka精讲（一）思维导图：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

### 目录 ###

*   *  1.简介
     *   *  1.1 消息队列简介
         *   *  1.1.1 什么是消息队列
             *  1.1.2 消息队列中间件
             *  1.1.3 消息队列的应用场景
             *   *  1.1.3.1 异步处理
                 *  1.1.3.2 系统解耦
                 *  1.1.3.3 流量削峰
                 *  1.1.3.4 日志处理（大数据领域常见）
             *  1.1.4 生产者、消费者模型
             *  1.1.5 消息队列的两种模式
             *   *  1.1.5.1 点对点模式
                 *  1.1.5.2 发布订阅模式
         *  1.2 Kafka简介
         *   *  1.2.1 什么是Kafka
             *  1.2.2 Kafka的应用场景
             *  1.2.3 Kafka诞生背景
         *  1.3 Kafka的优势
         *  1.4 Kafka生态圈介绍
         *  1.5 Kafka版本
     *  2. 环境搭建
     *   *  2.1 搭建Kafka集群
         *  2.2 目录结构分析
         *  2.3 Kafka一键启动/关闭脚本
     *  3. 基础操作
     *   *  3.1 创建topic
         *  3.2 生产消息到Kafka
         *  3.3 从Kafka消费消息
         *  3.4 使用Kafka Tools操作Kafka
         *   *  3.4.1 连接Kafka集群
             *  3.4.2 创建topic
     *  4. Kafka基准测试
     *   *  4.1 基准测试
         *   *  4.1.1 基于1个分区1个副本的基准测试
             *   *  4.1.1.1 创建topic
                 *  4.1.1.2 生产消息基准测试
                 *  4.1.1.3 消费消息基准测试
             *  4.1.2 基于3个分区1个副本的基准测试
             *   *  4.1.2.1 创建topic
                 *  4.1.2.2 生产消息基准测试
                 *  4.1.2.3 消费消息基准测试
             *  4.1.3 基于1个分区3个副本的基准测试
             *   *  4.1.3.1 创建topic
                 *  4.1.3.2 生产消息基准测试
                 *  4.1.3.3 消费消息基准测试
     *  5. Java编程操作Kafka
     *   *  5.1 同步生产消息到Kafka中
         *   *  5.1.1 需求
             *  5.1.2 准备工作
             *   *  5.1.2.1 导入Maven Kafka POM依赖
                 *  5.1.2.2 导入log4j.properties
                 *  5.1.2.3 创建包和类
             *  5.1.3 开发步骤
             *  5.1.4 代码开发
         *  5.2 从Kafka的topic中消费消息
         *   *  5.2.1 需求
             *  5.2.2 准备工作
             *  5.2.3 开发步骤
             *  5.2.4 代码开发
         *  5.3 异步使用带有回调函数方法生产消息
     *  6. Kafka架构
     *   *  6.1 Kafka重要概念
         *   *  6.1.1 broker
             *  6.1.2 zookeeper
             *  6.1.3 producer（生产者）
             *  6.1.4 consumer（消费者）
             *  6.1.5 consumer group（消费者组）
             *  6.1.6 分区（Partitions）
             *  6.1.7 副本（Replicas）
             *  6.1.8 主题（Topic）
             *  6.1.9 偏移量（offset）
         *  6.2 消费者组
     *  7. Kafka生产者幂等性与事务
     *   *  7.1 幂等性
         *   *  7.1.1 简介
             *  7.1.2 Kafka生产者幂等性
             *  7.1.3 配置幂等性
             *  7.1.4 幂等性原理
         *  7.2 Kafka事务
         *   *  7.2.1 简介
             *  7.2.2 事务操作API
         *  7.3 【理解】Kafka事务编程
         *   *  7.3.1 事务相关属性配置
             *   *  7.3.1.1 生产者
                 *  7.3.1.2 消费者
             *  7.3.2 Kafka事务编程
             *   *  7.3.2.1 需求
                 *  7.3.2.2 启动生产者控制台程序模拟数据
                 *  7.3.2.3 编写创建消费者代码
                 *  7.3.2.4 编写创建生产者代码
                 *  7.3.2.5 编写代码消费并生产数据
                 *  7.3.2.6 测试
                 *  7.3.2.7 模拟异常测试事务

## 1.简介 ##

### 1.1 消息队列简介 ###

#### 1.1.1 什么是消息队列 ####

消息队列，英文名：Message Queue，经常缩写为MQ。从字面上来理解，消息队列是一种用来存储消息的队列。来看一下下面的代码：

// 1. 创建一个保存字符串的队列
    Queue<String> stringQueue = new LinkedList<String>();
    
    // 2. 往消息队列中放入消息
    stringQueue.offer("hello");
    
    // 3. 从消息队列中取出消息并打印
    System.out.println(stringQueue.poll());

上述代码，创建了一个队列，先往队列中添加了一个消息，然后又从队列中取出了一个消息。这说明了队列是可以用来存取消息的。

我们可以简单理解消息队列就是**将需要传输的数据存放在队列中。**

#### 1.1.2 消息队列中间件 ####

消息队列中间件就是用来存储消息的软件（组件）。举个例子来理解，为了分析网站的用户行为，我们需要记录用户的访问日志。这些一条条的日志，可以看成是一条条的消息，我们可以将它们保存到消息队列中。将来有一些应用程序需要处理这些日志，就可以随时将这些消息取出来处理。

目前市面上的消息队列有很多，例如：Kafka、RabbitMQ、ActiveMQ、RocketMQ、ZeroMQ等。

#### 1.1.3 消息队列的应用场景 ####

##### 1.1.3.1 异步处理 #####

电商网站中，新的用户注册时，需要将用户的信息保存到数据库中，同时还需要额外发送注册的邮件通知、以及短信注册码给用户。但因为发送邮件、发送注册短信需要连接外部的服务器，需要额外等待一段时间，此时，就可以使用消息队列来进行异步处理，从而实现快速响应。

*  可以将一些比较耗时的操作放在其他系统中，通过消息队列将需要进行处理的消息进行存储，其他系统可以消费消息队列中的数据
 *  比较常见的：发送短信验证码、发送邮件

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

##### 1.1.3.2 系统解耦 #####

*  原先一个微服务是通过接口（HTTP）调用另一个微服务，这时候耦合很严重，只要接口发生变化就会导致系统不可用
 *  使用消息队列可以将系统进行解耦合，现在第一个微服务可以将消息放入到消息队列中，另一个微服务可以从消息队列中把消息取出来进行处理。进行系统解耦

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]

##### 1.1.3.3 流量削峰 #####

*  因为消息队列是低延迟、高可靠、高吞吐的，可以应对大量并发

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]

##### 1.1.3.4 日志处理（大数据领域常见） #####

大型电商网站（淘宝、京东、国美、苏宁…）、App（抖音、美团、滴滴等）等需要分析用户行为，要根据用户的访问行为来发现用户的喜好以及活跃情况，需要在页面上收集大量的用户访问信息。

*  可以使用消息队列作为临时存储，或者一种通信管道

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]

#### 1.1.4 生产者、消费者模型 ####

我们之前学习过Java的服务器开发，Java服务器端开发的交互模型是这样的：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]

我们之前也学习过使用Java JDBC来访问操作MySQL数据库，它的交互模型是这样的：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]  
它也是一种请求响应模型，只不过它不再是基于http协议，而是基于MySQL数据库的通信协议。而如果我们基于消息队列来编程，此时的交互模式成为：生产者、消费者模型。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]

#### 1.1.5 消息队列的两种模式 ####

##### 1.1.5.1 点对点模式 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]

消息发送者生产消息发送到消息队列中，然后消息接收者从消息队列中取出并且消费消息。消息被消费以后，消息队列中不再有存储，所以消息接收者不可能消费到已经被消费的消息。

点对点模式特点：

*  每个消息只有一个接收者（Consumer）(即一旦被消费，消息就不再在消息队列中)
 *  发送者和接收者间没有依赖性，发送者发送消息之后，不管有没有接收者在运行，都不会影响到发送者下次发送消息；
 *  接收者在成功接收消息之后需向队列应答成功，以便消息队列删除当前接收的消息；

##### 1.1.5.2 发布订阅模式 #####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]

发布/订阅模式特点：

*  每个消息可以有多个订阅者；
 *  发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息。
 *  为了消费消息，订阅者需要提前订阅该角色主题，并保持在线运行；

### 1.2 Kafka简介 ###

#### 1.2.1 什么是Kafka ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]

Kafka是由Apache软件基金会开发的一个开源流平台，由Scala和Java编写。Kafka的Apache官网是这样介绍Kafka的。

Apache Kafka是一个分布式流平台。一个分布式的流平台应该包含3点关键的能力：

1.  发布和订阅流数据流，类似于消息队列或者是企业消息传递系统
2.  以容错的持久化方式存储数据流
3.  处理数据流

英文原版

*  Publish and subscribe to streams of records, similar to a message queue or enterprise messaging system.
 *  Store streams of records in a fault-tolerant durable way.
 *  Process streams of records as they occur.

更多参考：[http://kafka.apache.org/documentation/\#introduction][http_kafka.apache.org_documentation_introduction]

我们重点关键三个部分的关键词：

1.  Publish and subscribe：发布与订阅
2.  Store：存储
3.  Process：处理

#### 1.2.2 Kafka的应用场景 ####

我们通常将Apache Kafka用在两类程序：

1.  建立实时数据管道，以可靠地在系统或应用程序之间获取数据
2.  构建实时流应用程序，以转换或响应数据流

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 11]  
上图，我们可以看到：  
3. Producers：可以有很多的应用程序，将消息数据放入到Kafka集群中。  
4. Consumers：可以有很多的应用程序，将消息数据从Kafka集群中拉取出来。  
5. Connectors：Kafka的连接器可以将数据库中的数据导入到Kafka，也可以将Kafka的数据导出到  
数据库中。  
6. Stream Processors：流处理器可以Kafka中拉取数据，也可以将数据写入到Kafka中。

#### 1.2.3 Kafka诞生背景 ####

kafka的诞生，是为了解决linkedin的数据管道问题，起初linkedin采用了ActiveMQ来进行数据交换，大约是在2010年前后，那时的ActiveMQ还远远无法满足linkedin对数据传递系统的要求，经常由于各种缺陷而导致消息阻塞或者服务无法正常访问，为了能够解决这个问题，linkedin决定研发自己的消息传递系统，当时linkedin的首席架构师jay kreps便开始组织团队进行消息传递系统的研发。

### 1.3 Kafka的优势 ###

前面我们了解到，消息队列中间件有很多，为什么我们要选择Kafka？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 12]  
在大数据技术领域，一些重要的组件、框架都支持Apache Kafka，不论成成熟度、社区、性能、可靠性，Kafka都是非常有竞争力的一款产品。

### 1.4 Kafka生态圈介绍 ###

Apache Kafka这么多年的发展，目前也有一个较庞大的生态圈。  
Kafka生态圈官网地址：[https://cwiki.apache.org/confluence/display/KAFKA/Ecosystem][https_cwiki.apache.org_confluence_display_KAFKA_Ecosystem]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 13]

### 1.5 Kafka版本 ###

本次学习使用的Kafka版本为2.4.1，是2020年3月12日发布的版本。  
可以注意到Kafka的版本号为：kafka\_2.12-2.4.1，因为kafka主要是使用scala语言开发的，2.12为scala的版本号。

## 2. 环境搭建 ##

### 2.1 搭建Kafka集群 ###

1.  将Kafka的安装包上传到虚拟机，并解压

cd /opt/softwares
    chmod u+x kafka_2.12-2.4.1.tgz 
    tar -zxvf kafka_2.12-2.4.1.tgz -C /opt/modules/
    ln -s kafka_2.12-2.4.1 kafka

1.  修改 server.properties

cd /opt/modules/kafka/config
    vim server.properties
    # 指定broker的id,在node1中修改编号为1，node2中修改编号为2，node3中修改编号为3，以此类推
    broker.id=0
    # 这里根据自己的所在的主机名修改
    listeners=PLAINTEXT://bigdata-pro-m09:9092
    advertised.listeners=PLAINTEXT://bigdata-pro-m09:9092
    # 指定Kafka数据的位置,这里需要在/opt/modules/kafka/目录下创建一个kafka-logs文件夹
    log.dirs=/opt/modules/kafka/kafka-logs
    # 配置zk的三个节点
    zookeeper.connect=bigdata-pro-m07:2181,bigdata-pro-m08:2181,bigdata-pro-m09:2181

1.  将安装好的kafka复制到另外两台服务器

cd /opt/modules
    scp -r kafka_2.12-2.4.1/ bigdata-pro-m08:/opt/modules
    scp -r kafka_2.12-2.4.1/ bigdata-pro-m09:/opt/modules
    
    修改另外两个节点的broker.id分别为2和3
    ---------bigdata-pro-m08--------------
    cd /opt/modules/kafka/config
    vim server.properties
    broker.id=2
    
    --------bigdata-pro-m09--------------
    cd /opt/modules/kafka/config
    vim server.properties
    broker.id=3

1.  配置KAFKA\_HOME环境变量

sudo vim /etc/profile
    
    # kafka
    export KAFKA_HOME=/opt/modules/kafka
    export PATH=${KAFKA_HOME}/bin:$PATH
    
    source /etc/profile

1.  启动服务器

# 启动ZooKeeper
    zkServer.sh start
    
    # 在测试之前，需要配置一下producer.properties文件，意思是配置生产者
    bootstrap.servers=bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092
    
    # 启动Kafka
    bin/kafka-server-start.sh -daemon config/server.properties

### 2.2 目录结构分析 ###

<table> 
 <thead> 
  <tr> 
   <th>目录名称</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>bin</td> 
   <td>Kafka的所有执行脚本都在这里。例如：启动Kafka服务器、创建Topic、生产者、消费者程序等等</td> 
  </tr> 
  <tr> 
   <td>config</td> 
   <td>Kafka的所有配置文件</td> 
  </tr> 
  <tr> 
   <td>libs</td> 
   <td>运行Kafka所需要的所有JAR包</td> 
  </tr> 
  <tr> 
   <td>logs</td> 
   <td>Kafka的所有日志文件，如果Kafka出现一些问题，需要到该目录中去查看异常信息</td> 
  </tr> 
  <tr> 
   <td>site-docs</td> 
   <td>Kafka的网站帮助文件</td> 
  </tr> 
 </tbody> 
</table>

### 2.3 Kafka一键启动/关闭脚本 ###

为了方便将来进行一键启动、关闭Kafka，我们可以编写一个shell脚本来操作。将来只要执行一次该脚本就可以快速启动/关闭Kafka。

1.  在节点1中创建 `/opt/modules/kafka/onekey`目录

cd /opt/modules/kafka/onekey

1.  准备slave配置文件，用于保存要启动哪几个节点上的kafka

bigdata-pro-m07
    bigdata-pro-m08
    bigdata-pro-m09

1.  编写start-kafka.sh脚本

# !/bin/bash
    
    cat /opt/modules/kafka/onekey/slaves | while read line
    do
    { 
     echo $line
     ssh $line "source /etc/profile;export JMX_PORT=9988;nohup ${KAFKA_HOME}/bin/kafka-server-start.sh ${KAFKA_HOME}/config/server.properties >/dev/nul* 2>&1 & "
    }&
    wait
    done

1.  编写stop-kafka.sh脚本

# !/bin/bash
    
    cat /opt/modules/kafka/onekey/slaves | while read line
    do
    { 
     echo $line
     ssh $line "source /etc/profile;jps |grep Kafka |cut -d' ' -f1 |xargs kill -s 9"
    }&
    wait
    done

1.  给start-kafka.sh、stop-kafka.sh配置执行权限

chmod u+x start-kafka.sh
    chmod u+x stop-kafka.sh

1.  执行一键启动、一键关闭

./start-kafka.sh
    ./stop-kafka.sh

## 3. 基础操作 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 14]

### 3.1 创建topic ###

创建一个topic（主题）。Kafka中所有的消息都是保存在主题中，要生产消息到Kafka，首先必须要有一个确定的主题。

# 创建名为test的主题
    bin/kafka-topics.sh --create --zookeeper bigdata-pro-m07:2181 --replication-factor 1 --partitions 1 --topic test
    # 查看目前Kafka中的主题
    bin/kafka-topics.sh --list --zookeeper bigdata-pro-m07:2181

### 3.2 生产消息到Kafka ###

使用Kafka内置的测试程序，生产一些消息到Kafka的test主题中。

bin/kafka-console-producer.sh --broker-list bigdata-pro-m07:9092 --topic test

### 3.3 从Kafka消费消息 ###

使用下面的命令来消费 test 主题中的消息。

bin/kafka-console-consumer.sh --bootstrap-server bigdata-pro-m07:9092 --topic test --from-beginning

### 3.4 使用Kafka Tools操作Kafka ###

#### 3.4.1 连接Kafka集群 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 15]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 16]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 17]

#### 3.4.2 创建topic ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 18]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 19]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 20]

## 4. Kafka基准测试 ##

### 4.1 基准测试 ###

基准测试（benchmark testing）是一种测量和评估软件性能指标的活动。我们可以通过基准测试，了解到软件、硬件的性能水平。主要测试负载的执行时间、传输速度、吞吐量、资源占用率等。

#### 4.1.1 基于1个分区1个副本的基准测试 ####

测试步骤：

1.  启动Kafka集群
2.  创建一个1个分区1个副本的topic: benchmark
3.  同时运行生产者、消费者基准测试程序
4.  观察结果

##### 4.1.1.1 创建topic #####

bin/kafka-topics.sh --zookeeper bigdata-pro-m07:2181 --create --topic benchmark --partitions 1 --replication-factor 1

##### 4.1.1.2 生产消息基准测试 #####

在生产环境中，推荐使用生产5000W消息，这样会性能数据会更准确些。为了方便测试，课程上演示测试500W的消息作为基准测试。

bin/kafka-producer-perf-test.sh --topic benchmark --num-records 5000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 acks=1

bin/kafka-producer-perf-test.sh 
    --topic topic的名字
    --num-records	总共指定生产数据量（默认5000W）
    --throughput	指定吞吐量——限流（-1不指定）
    --record-size   record数据大小（字节）
    --producer-props bootstrap.servers=bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 acks=1 指定Kafka集群地址，ACK模式

测试结果：

<table> 
 <thead> 
  <tr> 
   <th>吞吐量</th> 
   <th>93092.533979 records/sec 每秒9.3W条记录</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>吞吐速率</td> 
   <td>(88.78 MB/sec) 每秒约89MB数据</td> 
  </tr> 
  <tr> 
   <td>平均延迟时间</td> 
   <td>346.62 ms avg latency</td> 
  </tr> 
  <tr> 
   <td>最大延迟时间</td> 
   <td>1003.00 ms max latency</td> 
  </tr> 
 </tbody> 
</table>

##### 4.1.1.3 消费消息基准测试 #####

bin/kafka-consumer-perf-test.sh --broker-list bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 --topic benchmark --fetch-size 1048576 --messages 5000000

bin/kafka-consumer-perf-test.sh
    --broker-list 指定kafka集群地址
    --topic 指定topic的名称
    --fetch-size 每次拉取的数据大小
    --messages 总共要消费的消息个数

测试结果：

<table> 
 <thead> 
  <tr> 
   <th>data.consumed.in.MB 共计消费的数据</th> 
   <th>4768.3716MB</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>MB.sec 每秒消费的数量</td> 
   <td>445.6006 每秒445MB</td> 
  </tr> 
  <tr> 
   <td>data.consumed.in.nMsg 共计消费的数量</td> 
   <td>5000000</td> 
  </tr> 
  <tr> 
   <td>nMsg.sec 每秒的数量</td> 
   <td>467246.0518 每秒46.7W条</td> 
  </tr> 
 </tbody> 
</table>

#### 4.1.2 基于3个分区1个副本的基准测试 ####

被测虚拟机：

<table> 
 <thead> 
  <tr> 
   <th>bigdata-pro-m07</th> 
   <th>bigdata-pro-m08</th> 
   <th>bigdata-pro-m09</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>i9 9th 6G</td> 
   <td>i9 9th 3G</td> 
   <td>i9 9th 3G</td> 
  </tr> 
 </tbody> 
</table>

##### 4.1.2.1 创建topic #####

bin/kafka-topics.sh --zookeeper bigdata-pro-m07:2181 --create --topic benchmark1 --partitions 3 --replication-factor 1

##### 4.1.2.2 生产消息基准测试 #####

bin/kafka-producer-perf-test.sh --topic benchmark1 --num-records 5000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 acks=1

测试结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 21]  
在虚拟机上，因为都是共享笔记本上的CPU、内存、网络，所以分区越多，反而效率越低。但如果是真实的服务器，分区多效率是会有明显提升的。

##### 4.1.2.3 消费消息基准测试 #####

bin/kafka-consumer-perf-test.sh --broker-list bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 --topic benchmark1 --fetch-size 1048576 --messages 5000000

#### 4.1.3 基于1个分区3个副本的基准测试 ####

##### 4.1.3.1 创建topic #####

bin/kafka-topics.sh --zookeeper bigdata-pro-m07:2181 --create --topic benchmark2 --partitions 1 --replication-factor 3

##### 4.1.3.2 生产消息基准测试 #####

bin/kafka-producer-perf-test.sh --topic benchmark2 --num-records 5000000 --throughput -1 --record-size 1000 --producer-props bootstrap.servers=bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 acks=1

##### 4.1.3.3 消费消息基准测试 #####

bin/kafka-consumer-perf-test.sh --broker-list bigdata-pro-m07:9092,bigdata-pro-m08:9092,bigdata-pro-m09:9092 --topic benchmark2 --fetch-size 1048576 --messages 5000000

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 24]

## 5. Java编程操作Kafka ##

### 5.1 同步生产消息到Kafka中 ###

#### 5.1.1 需求 ####

接下来，我们将编写Java程序，将1-100的数字消息写入到Kafka中。

#### 5.1.2 准备工作 ####

##### 5.1.2.1 导入Maven Kafka POM依赖 #####

<repositories>
            <repository>
                <id>central</id>
                <url>http://maven.aliyun.com/nexus/content/groups/public//</url>
                <releases>
                    <enabled>true</enabled>
                </releases>
                <snapshots>
                    <enabled>true</enabled>
                    <updatePolicy>always</updatePolicy>
                    <checksumPolicy>fail</checksumPolicy>
                </snapshots>
            </repository>
        </repositories>
    
        <dependencies>
            
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>2.4.1</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.apache.commons</groupId>
                <artifactId>commons-io</artifactId>
                <version>1.3.2</version>
            </dependency>
    
            
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.7.6</version>
            </dependency>
    
            
            <dependency>
                <groupId>log4j</groupId>
                <artifactId>log4j</artifactId>
                <version>1.2.16</version>
            </dependency>
        </dependencies>

##### 5.1.2.2 导入log4j.properties #####

将log4j.properties配置文件放入到resources文件夹中

log4j.rootLogger=INFO,stdout
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender 
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout 
    log4j.appender.stdout.layout.ConversionPattern=%5p - %m%n

##### 5.1.2.3 创建包和类 #####

创建包com.kafka，并创建KafkaProducerTest类。

#### 5.1.3 开发步骤 ####

1.  创建用于连接Kafka的Properties配置

Properties props = new Properties();
    props.put("bootstrap.servers", "bigdata-pro-m07:9092");
    props.put("acks", "all");
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

1.  创建一个生产者对象KafkaProducer
2.  调用send发送1-100消息到指定Topic test，并获取返回值Future，该对象封装了返回值
3.  再调用一个Future.get()方法等待响应
4.  关闭生产者

#### 5.1.4 代码开发 ####

package com.kafka;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    
    import java.util.Properties;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.Future;
    
    /** * kafka生产者程序，会将消息创建出来，并发送到Kafka集群中 * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2021/1/9 * @time : 7:50 下午 */
    public class KafkaProducerTest { 
        public static void main(String[] args) { 
    
            // 1.创建kafka生产者配置
            Properties props = new Properties();
            props.put("bootstrap.servers", "bigdata-pro-m07:9092");
            props.put("acks", "all");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 2.创建一个生产者对象KafkaProducer
            KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(props);
    
            // 3.调用send发送1-100消息到指定Topic test，并获取返回值Future，该对象封装了返回值
            for (int i = 0;i < 100;i++){ 
                // 使用同步等待的方式发送消息
                // 构建一条消息，直接new ProducerRecord
                ProducerRecord<String,String> producerRecord =  new ProducerRecord<String,String>("test",null,i+"");
                Future<RecordMetadata> future = kafkaProducer.send(producerRecord);
    
                try { 
                    // 调用Future的get方法等待响应
                    future.get();
                    System.out.println("第" + i + "条消息写入成功！");
                } catch (InterruptedException | ExecutionException e) { 
                    e.printStackTrace();
                }
            }
    
            // 4.关闭生产者
            kafkaProducer.close();
        }
    }

### 5.2 从Kafka的topic中消费消息 ###

#### 5.2.1 需求 ####

从 test topic中，将消息都消费，并将记录的offset、key、value打印出来

#### 5.2.2 准备工作 ####

在com.kafka包下创建KafkaConsumerTest类

#### 5.2.3 开发步骤 ####

1.  创建Kafka消费者配置

Properties props = new Properties();
    props.setProperty("bootstrap.servers", "bigdata-pro-m07:9092");
    props.setProperty("group.id", "test");
    props.setProperty("enable.auto.commit", "true");
    props.setProperty("auto.commit.interval.ms", "1000");
    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

1.  创建Kafka消费者
2.  订阅要消费的主题
3.  使用一个while循环，不断从Kafka的topic中拉取消息
4.  将将记录（record）的offset、key、value都打印出来

*  group.id：消费者组的概念，可以在一个消费组中包含多个消费者。如果若干个消费者的group.id是一样的，表示它们就在一个组中，一个组中的消费者是共同消费Kafka中topic的数据。
 *  Kafka是一种拉消息模式的消息队列，在消费者中会有一个offset，表示从哪条消息开始拉取数据
 *  kafkaConsumer.poll：Kafka的消费者API是一批一批数据的拉取

#### 5.2.4 代码开发 ####

package com.kafka;
    
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Arrays;
    import java.util.Properties;
    
    /** * kafka消费者程序 * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2021/1/9 * @time : 7:47 下午 */
    public class KafkaConsumerTest { 
        public static void main(String[] args) { 
    
            // 1.创建Kafka消费者配置
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "bigdata-pro-m07:9092");
            // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据
            // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的
            props.setProperty("group.id", "test");
            // 自动提交offset
            props.setProperty("enable.auto.commit", "true");
            // 自动提交offset的时间间隔
            props.setProperty("auto.commit.interval.ms", "1000");
            // 拉取的key、value数据的
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            // 2.创建Kafka消费者
            KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String,String>(props);
    
            // 3.订阅要消费的主题,指定消费者从哪个topic中拉取数据
            kafkaConsumer.subscribe(Arrays.asList("test"));
    
            // 4.使用一个while循环，不断从Kafka的topic中拉取消息
            while (true){ 
                // Kafka的消费者一次拉取一批的数据
                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
                // 5.将将记录（record）的offset、key、value都打印出来
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) { 
                    // 主题
                    String topic = consumerRecord.topic();
                    // offset：这条消息处于Kafka分区中的哪个位置
                    long offset = consumerRecord.offset();
                    // key\value
                    String key = consumerRecord.key();
                    String value = consumerRecord.value();
    
                    System.out.println("topic: " + topic + "offset: " + offset + "key: " + key + "value: " + value);
                }
            }
        }
    }

### 5.3 异步使用带有回调函数方法生产消息 ###

如果我们想获取生产者消息是否成功，或者成功生产消息到Kafka中后，执行一些其他动作。此时，可以很方便地使用带有回调函数来发送消息。

需求：

1.  在发送消息出现异常时，能够及时打印出异常信息
2.  在发送消息成功时，打印Kafka的topic名字、分区id、offset

代码实现：

package com.kafka;
    
    import org.apache.kafka.clients.producer.Callback;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    
    import java.util.Properties;
    
    /** * kafka生产者程序，会将消息创建出来，并发送到Kafka集群中 * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2021/1/9 * @time : 7:50 下午 */
    public class KafkaProducerTest2 { 
        public static void main(String[] args) { 
    
            // 1.创建kafka生产者配置
            Properties props = new Properties();
            props.put("bootstrap.servers", "bigdata-pro-m07:9092");
            props.put("acks", "all");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 2.创建一个生产者对象KafkaProducer
            KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(props);
    
            // 3.调用send发送1-100消息到指定Topic test，并获取返回值Future，该对象封装了返回值
            for (int i = 0;i < 100;i++){ 
                // 使用异步等待的方式发送消息
                // 构建一条消息，直接new ProducerRecord
                ProducerRecord<String,String> producerRecord =  new ProducerRecord<String,String>("test",null,i+"");
                kafkaProducer.send(producerRecord, new Callback() { 
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception exception) { 
                        // 1. 判断发送消息是否成功
                        if (exception == null){ 
                            // 发送成功
                            // 主题
                            String topic = recordMetadata.topic();
                            // 分区id
                            int partition = recordMetadata.partition();
                            // 偏移量
                            long offset = recordMetadata.offset();
    
                            System.out.println("topic:" + topic + " 分区id：" + partition + " 偏移量：" + offset);
                        } else { 
                            // 发送出现错误
                            System.out.println("生产消息出现异常！");
                            // 打印异常消息
                            System.out.println(exception.getMessage());
                            // 打印调用栈
                            System.out.println(exception.getStackTrace());
                        }
                    }
                });
            }
    
            // 4.关闭生产者
            kafkaProducer.close();
        }
    }

*  使用匿名内部类实现Callback接口，该接口中表示Kafka服务器响应给客户端，会自动调用onCompletion方法
    
     *  metadata：消息的元数据（属于哪个topic、属于哪个partition、对应的offset是什么）
     *  exception：这个对象Kafka生产消息封装了出现的异常，如果为null，表示发送成功，如果不为null，表示出现异常。

## 6. Kafka架构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 25]

### 6.1 Kafka重要概念 ###

#### 6.1.1 broker ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 26]

*  一个Kafka的集群通常由多个broker组成，这样才能实现负载均衡、以及容错
 *  broker是无状态（Sateless）的，它们是通过ZooKeeper来维护集群状态
 *  一个Kafka的broker每秒可以处理数十万次读写，每个broker都可以处理TB消息而不影响性能

#### 6.1.2 zookeeper ####

*  ZK用来管理和协调broker，并且存储了Kafka的元数据（例如：有多少topic、partition、consumer）
 *  ZK服务主要用于通知生产者和消费者Kafka集群中有新的broker加入、或者Kafka集群中出现故障的broker。

PS：Kafka正在逐步想办法将ZooKeeper剥离，维护两套集群成本较高，社区提出KIP-500就是要替换掉ZooKeeper的依赖。“Kafka on Kafka”——Kafka自己来管理自己的元数据

#### 6.1.3 producer（生产者） ####

*  生产者负责将数据推送给broker的topic

#### 6.1.4 consumer（消费者） ####

*  消费者负责从broker的topic中拉取数据，并自己进行处理

#### 6.1.5 consumer group（消费者组） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 27]

*  consumer group是kafka提供的可扩展且具有容错性的消费者机制
 *  一个消费者组可以包含多个消费者
 *  一个消费者组有一个唯一的ID（group Id）
 *  组内的消费者一起消费主题的所有分区数据

#### 6.1.6 分区（Partitions） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 28]

*  在Kafka集群中，主题被分为多个分区

#### 6.1.7 副本（Replicas） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 29]

*  副本可以确保某个服务器出现故障时，确保数据依然可用
 *  在Kafka中，一般都会设计副本的个数＞1

#### 6.1.8 主题（Topic） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 30]

*  主题是一个逻辑概念，用于生产者发布数据，消费者拉取数据
 *  Kafka中的主题必须要有标识符，而且是唯一的，Kafka中可以有任意数量的主题，没有数量上的限制
 *  在主题中的消息是有结构的，一般一个主题包含某一类消息
 *  一旦生产者发送消息到主题中，这些消息就不能被更新（更改）

#### 6.1.9 偏移量（offset） ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 31]

*  offset记录着下一条将要发送给Consumer的消息的序号
 *  默认Kafka将offset存储在ZooKeeper中
 *  在一个分区中，消息是有顺序的方式存储着，每个在分区的消费都是有一个递增的id。这个就是偏移量offset
 *  偏移量在分区中才是有意义的。在分区之间，offset是没有任何意义的

### 6.2 消费者组 ###

Kafka支持有多个消费者同时消费一个主题中的数据。我下面给大家演示，启动两个消费者共同来消费 test 主题的数据。

1.  首先，修改生产者程序

// 3.调用send发送1-100消息到指定Topic test，并获取返回值Future，该对象封装了返回值
    int MAX = 1000000;
    for (int i = 1000;i < MAX;i++){ 
        // 使用异步等待的方式发送消息
        // 构建一条消息，直接new ProducerRecord
        ProducerRecord<String,String> producerRecord =  new ProducerRecord<String,String>("test",null,i+"");
        kafkaProducer.send(producerRecord, new Callback() { 
            @Override
            public void onCompletion(RecordMetadata recordMetadata, Exception exception) { 
                // 1. 判断发送消息是否成功
                if (exception == null){ 
                    // 发送成功
                    // 主题
                    String topic = recordMetadata.topic();
                    // 分区id
                    int partition = recordMetadata.partition();
                    // 偏移量
                    long offset = recordMetadata.offset();
    
                    System.out.println("topic:" + topic + " 分区id：" + partition + " 偏移量：" + offset);
                } else { 
                    // 发送出现错误
                    System.out.println("生产消息出现异常！");
                    // 打印异常消息
                    System.out.println(exception.getMessage());
                    // 打印调用栈
                    System.out.println(exception.getStackTrace());
                }
            }
        });
    
        try { 
            Thread.sleep(1000);
        } catch (InterruptedException e) { 
            e.printStackTrace();
        }
    }

1.  接下来，同时运行两个消费者。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 32]

1.  同时运行两个消费者，我们发现，只有一个消费者程序能够拉取到消息。想要让两个消费者同时消费消息，必须要给test主题，添加一个分区。

# 设置 test topic为2个分区
    bin/kafka-topics.sh --zookeeper bigdata-pro-m07:2181 -alter --partitions 2 --topic test

1.  重新运行生产者、两个消费者程序，就可以看到两个消费者都可以消费Kafka Topic的数据了

消费者1

topic: testoffset: 338key: nullvalue: 1041
    topic: testoffset: 339key: nullvalue: 1043
    topic: testoffset: 340key: nullvalue: 1045
    topic: testoffset: 341key: nullvalue: 1047
    topic: testoffset: 342key: nullvalue: 1049
    topic: testoffset: 343key: nullvalue: 1051
    topic: testoffset: 344key: nullvalue: 1053
    topic: testoffset: 345key: nullvalue: 1055

消费者2

topic: testoffset: 19key: nullvalue: 1010
    topic: testoffset: 20key: nullvalue: 1012
    topic: testoffset: 21key: nullvalue: 1014
    topic: testoffset: 22key: nullvalue: 1016
    topic: testoffset: 23key: nullvalue: 1018
    topic: testoffset: 24key: nullvalue: 1020
    topic: testoffset: 25key: nullvalue: 1022

*  一个消费者组中可以包含多个消费者，共同来消费topic中的数据
 *  一个topic中如果只有一个分区，那么这个分区只能被某个组中的一个消费者消费
 *  有多少个分区，那么就可以被同一个组内的多少个消费者消费

## 7. Kafka生产者幂等性与事务 ##

### 7.1 幂等性 ###

#### 7.1.1 简介 ####

拿http举例来说，一次或多次请求，得到地响应是一致的（网络超时等问题除外），换句话说，就是执行多次操作与执行一次操作的影响是一样的。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 33]  
如果，某个系统是不具备幂等性的，如果用户重复提交了某个表格，就可能会造成不良影响。例如：用户在浏览器上点击了多次提交订单按钮，会在后台生成多个一模一样的订单。

#### 7.1.2 Kafka生产者幂等性 ####

#### 7.1.3 配置幂等性 ####

props.put("enable.idempotence",true);

#### 7.1.4 幂等性原理 ####

为了实现生产者的幂等性，Kafka引入了 Producer ID（PID）和 Sequence Number的概念。

*  PID：每个Producer在初始化时，都会分配一个唯一的PID，这个PID对用户来说，是透明的。
 *  Sequence Number：针对每个生产者（对应PID）发送到指定主题分区的消息都对应一个从0开始递增的Sequence Number。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 35]

*  生产者消息重复问题
    
     *  Kafka生产者生产消息到partition，如果直接发送消息，kafka会将消息保存到分区中，但Kafka会返回一个ack给生产者，表示当前操作是否成功，是否已经保存了这条消息。如果ack响应的过程失败了，此时生产者会重试，继续发送没有发送成功的消息，Kafka又会保存一条一模一样的消息
 *  在Kafka中可以开启幂等性
    
     *  当Kafka的生产者生产消息时，会增加一个pid（生产者的唯一编号）和sequence number（针对消息的一个递增序列）
     *  发送消息，会连着pid和sequence number一块发送
     *  kafka接收到消息，会将消息和pid、sequence number一并保存下来
     *  如果ack响应失败，生产者重试，再次发送消息时，Kafka会根据pid、sequence number是否需要再保存一条消息
     *  判断条件：生产者发送过来的sequence number 是否小于等于 partition中消息对应的sequence

### 7.2 Kafka事务 ###

#### 7.2.1 简介 ####

Kafka事务是2017年Kafka 0.11.0.0引入的新特性。类似于数据库的事务。Kafka事务指的是生产者生产消息以及消费者提交offset的操作可以在一个原子操作中，要么都成功，要么都失败。尤其是在生产者、消费者并存时，事务的保障尤其重要。（consumer-transform-producer模式）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 36]

#### 7.2.2 事务操作API ####

Producer接口中定义了以下5个事务相关方法：

1.  initTransactions（初始化事务）：要使用Kafka事务，必须先进行初始化操作
2.  beginTransaction（开始事务）：启动一个Kafka事务
3.  sendOffsetsToTransaction（提交偏移量）：批量地将分区对应的offset发送到事务中，方便后续一块提交
4.  commitTransaction（提交事务）：提交事务
5.  abortTransaction（放弃事务）：取消事务

### 7.3 【理解】Kafka事务编程 ###

#### 7.3.1 事务相关属性配置 ####

##### 7.3.1.1 生产者 #####

// 配置事务的id，开启了事务会默认开启幂等性
    props.put("transactional.id", "first-transactional");

##### 7.3.1.2 消费者 #####

// 1. 消费者需要设置隔离级别
    props.put("isolation.level","read_committed");
    // 2. 关闭自动提交
     props.put("enable.auto.commit", "false");

#### 7.3.2 Kafka事务编程 ####

##### 7.3.2.1 需求 #####

在Kafka的topic 「ods\_user」中有一些用户数据，数据格式如下：

姓名,性别,出生日期
    张三,1,1980-10-09
    李四,0,1985-11-01

我们需要编写程序，将用户的性别转换为男、女（1-男，0-女），转换后将数据写入到topic 「dwd\_user」中。要求使用事务保障，要么消费了数据同时写入数据到 topic，提交offset。要么全部失败。

##### 7.3.2.2 启动生产者控制台程序模拟数据 #####

# 创建名为ods_user和dwd_user的主题
    bin/kafka-topics.sh --create --bootstrap-server bigdata-pro-m07:9092 --topic ods_user
    bin/kafka-topics.sh --create --bootstrap-server bigdata-pro-m07:9092 --topic dwd_user
    # 生产数据到 ods_user
    bin/kafka-console-producer.sh --broker-list bigdata-pro-m07:9092 --topic ods_user
    # 从dwd_user消费数据
    bin/kafka-console-consumer.sh --bootstrap-server bigdata-pro-m07:9092 --topic dwd_user --from-beginning  --isolation-level read_committed

##### 7.3.2.3 编写创建消费者代码 #####

编写一个方法 createConsumer，该方法中返回一个消费者，订阅「ods\_user」主题。注意：需要配置事务隔离级别、关闭自动提交。

实现步骤：

/** * 一、创建一个消费者来消费ods_user中的数据 * @return */
        public static KafkaConsumer<String,String> createConsumer(){ 
            // 1.创建Kafka消费者配置
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "bigdata-pro-m07:9092");
            // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据
            // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的
            props.setProperty("group.id", "ods_user");
            // 配置事务的隔离级别
            props.put("isolation.level","read_committed");
            // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset
            props.setProperty("enable.auto.commit", "false");
            // 反序列化器
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            // 2.创建Kafka消费者
            KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String,String>(props);
    
            // 3.订阅要消费的主题,指定消费者从哪个topic中拉取数据
            kafkaConsumer.subscribe(Arrays.asList("ods_user"));
    
            return kafkaConsumer;
        }

##### 7.3.2.4 编写创建生产者代码 #####

编写一个方法 createProducer，返回一个生产者对象。注意：需要配置事务的id，开启了事务会默认开启幂等性。

实现步骤：

/** * 二、编写createProducer方法，用来创建一个带有事务配置的生产者 */
        public static KafkaProducer<String,String> createProducer(){ 
            // 1. 配置生产者带有事务配置的属性
            Properties props = new Properties();
            props.put("bootstrap.servers", "bigdata-pro-m07:9092");
            props.put("acks", "all");
            // 开启事务必须要配置事务的ID
            props.put("transactional.id", "dwd_user");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 2.创建一个生产者对象KafkaProducer
            KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(props);
    
            return kafkaProducer;
        }

##### 7.3.2.5 编写代码消费并生产数据 #####

实现步骤：

1.  调用之前实现的方法，创建消费者、生产者对象
2.  生产者调用initTransactions初始化事务
3.  编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic  
    (1) 生产者开启事务  
    (2) 消费者拉取消息  
    (3) 遍历拉取到的消息，并进行预处理（将1转换为男，0转换为女）  
    (4) 生产消息到dwd\_user topic中  
    (5) 提交偏移量到事务中  
    (6) 提交事务  
    (7) 捕获异常，如果出现异常，则取消事务

package com.kafka;
    
    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.clients.consumer.OffsetAndMetadata;
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    import org.apache.kafka.clients.producer.RecordMetadata;
    import org.apache.kafka.common.TopicPartition;
    
    import java.time.Duration;
    import java.util.Arrays;
    import java.util.HashMap;
    import java.util.Map;
    import java.util.Properties;
    import java.util.concurrent.Future;
    
    /** * @author : 蔡政洁 * @email :caizhengjie888@icloud.com * @date : 2021/1/10 * @time : 2:56 下午 */
    public class TransactionProgram { 
        public static void main(String[] args) { 
            // 1. 调用之前实现的方法，创建消费者、生产者对象
            KafkaConsumer<String,String> consumer = createConsumer();
            KafkaProducer<String,String> producer = createProducer();
    
            // 2. 生产者调用initTransactions初始化事务
            producer.initTransactions();
    
            // 3. 编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic
            while (true){ 
                try { 
                    // (1) 生产者开启事务
                    producer.beginTransaction();
    
                    // 这个Map保存了topic对应的partition的偏移量
                    Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
    
                    // 从topic中拉取一批的数据
                    // (2) 消费者拉取消息
                    ConsumerRecords<String, String> concumserRecordArray = consumer.poll(Duration.ofSeconds(5));
    
                    // (3) 遍历拉取到的消息，并进行预处理
                    for (ConsumerRecord<String, String> consumerRecord : concumserRecordArray) { 
                        // 将1转换为男，0转为女
                        String msg = consumerRecord.value();
                        String[] fieldArray = msg.split(",");
    
                        // 将消息的偏移量保存
                        // 消费的是ods_user中的数据
                        String topic = consumerRecord.topic();
                        int partition = consumerRecord.partition();
                        long offset = consumerRecord.offset();
    
                        // offset + 1：offset是当前消费的记录（消息）对应在partition中的offset，而我们希望下一次能继续从下一个消息消息
                        // 必须要+1，从能消费下一条消息
                        offsetMap.put(new TopicPartition(topic,partition),new OffsetAndMetadata(offset+1));
    
                        // 将字段进行替换
                        if (fieldArray != null && fieldArray.length > 2){ 
                            String sexField = fieldArray[1];
                            if (sexField.equals("1")){ 
                                fieldArray[1] = "男";
                            } else if (sexField.equals("0")){ 
                                fieldArray[1] = "女";
                            }
                        }
    
                        // 重新拼接字段
                        msg = fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2];
    
                        // (4) 生产消息到dwd_user topic中
                        ProducerRecord<String,String> dwdMsg =  new ProducerRecord<String,String>("dwd_user",msg);
    
                        // 发送消息
                        Future<RecordMetadata> future = producer.send(dwdMsg);
                        try { 
                            future.get();
                        } catch (Exception e) { 
                            e.printStackTrace();
                            producer.abortTransaction();
                        }
                    }
                    producer.sendOffsetsToTransaction(offsetMap,"ods_user");
    
                    // int i = 1 / 0;
    
                    // (6) 提交事务
                    producer.commitTransaction();
                } catch (Exception e){ 
                    e.printStackTrace();
                    // (7) 捕获异常，如果出现异常，则取消事务
                    producer.abortTransaction();
                }
            }
    
        }
    
        /** * 一、创建一个消费者来消费ods_user中的数据 * @return */
        public static KafkaConsumer<String,String> createConsumer(){ 
            // 1.创建Kafka消费者配置
            Properties props = new Properties();
            props.setProperty("bootstrap.servers", "bigdata-pro-m07:9092");
            // 消费者组（可以使用消费者组将若干个消费者组织到一起），共同消费Kafka中topic的数据
            // 每一个消费者需要指定一个消费者组，如果消费者的组名是一样的，表示这几个消费者是一个组中的
            props.setProperty("group.id", "ods_user");
            // 配置事务的隔离级别
            props.put("isolation.level","read_committed");
            // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset
            props.setProperty("enable.auto.commit", "false");
            // 反序列化器
            props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            // 2.创建Kafka消费者
            KafkaConsumer<String,String> kafkaConsumer = new KafkaConsumer<String,String>(props);
    
            // 3.订阅要消费的主题,指定消费者从哪个topic中拉取数据
            kafkaConsumer.subscribe(Arrays.asList("ods_user"));
    
            return kafkaConsumer;
        }
    
        /** * 二、编写createProducer方法，用来创建一个带有事务配置的生产者 */
        public static KafkaProducer<String,String> createProducer(){ 
            // 1. 配置生产者带有事务配置的属性
            Properties props = new Properties();
            props.put("bootstrap.servers", "bigdata-pro-m07:9092");
            props.put("acks", "all");
            // 开启事务必须要配置事务的ID
            props.put("transactional.id", "dwd_user");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            // 2.创建一个生产者对象KafkaProducer
            KafkaProducer<String,String> kafkaProducer = new KafkaProducer<String,String>(props);
    
            return kafkaProducer;
        }
    
    }

##### 7.3.2.6 测试 #####

往之前启动的console-producer中写入消息进行测试，同时检查console-consumer是否能够接收到消息：

bin/kafka-console-producer.sh --broker-list bigdata-pro-m07:9092 --topic ods_user

张三,1,1980-10-09
    李四,0,1985-11-01

启动消费者：

bin/kafka-console-consumer.sh --bootstrap-server bigdata-pro-m07:9092 --topic dwd_user --from-beginning  --isolation-level read_committed

张三,男,1980-10-09
    李四,女,1985-11-01

##### 7.3.2.7 模拟异常测试事务 #####

// 3. 保存偏移量
    offsetCommits.put(new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1));
    // 4. 进行转换处理
    String[] fields = record.value().split(",");
    fields[1] = fields[1].equalsIgnoreCase("1") ? "男":"女";
    String message = fields[0] + "," + fields[1] + "," + fields[2];
    
    // 模拟异常
    int i = 1/0;
    
    // 5. 生产消息到dwd_user
    producer.send(new ProducerRecord<>("dwd_user", message));

启动程序一次，抛出异常。  
再启动程序一次，还是抛出异常。  
直到我们处理该异常为止。

我们发现，可以消费到消息，但如果中间出现异常的话，offset是不会被提交的，除非消费、生产消息都成功，才会提交事务。

--------------------

到这里Kafka精讲（一）的内容就全部结束了，后面还会更新Kafka精讲（二）！！！

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221119/8ac0ddc02b23419aaa328b257c76250f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221119/1289eb6ef54a41d0908beddcddb40ed6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20210109113558860.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20210109113648304.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20210109113759621.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20210109113856250.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/2021010911390560.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20210109113946881.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 8]: https://img-blog.csdnimg.cn/20210109114128160.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 9]: https://img-blog.csdnimg.cn/20210109114313906.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 10]: https://img-blog.csdnimg.cn/20210109120712956.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[http_kafka.apache.org_documentation_introduction]: http://kafka.apache.org/documentation/#introduction
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 11]: https://img-blog.csdnimg.cn/20210109120750921.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 12]: https://img-blog.csdnimg.cn/20210109120859790.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[https_cwiki.apache.org_confluence_display_KAFKA_Ecosystem]: https://cwiki.apache.org/confluence/display/KAFKA/Ecosystem
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 13]: https://img-blog.csdnimg.cn/20210109121023427.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 14]: https://img-blog.csdnimg.cn/20210109163042267.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 15]: https://img-blog.csdnimg.cn/20210109163530390.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 16]: https://img-blog.csdnimg.cn/20210109163543843.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 17]: https://img-blog.csdnimg.cn/20210109163602721.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 18]: https://img-blog.csdnimg.cn/2021010916364868.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 19]: https://img-blog.csdnimg.cn/20210109163659858.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 20]: https://img-blog.csdnimg.cn/20210109163712708.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 21]: https://img-blog.csdnimg.cn/20210109165612967.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 22]: https://img-blog.csdnimg.cn/20210109165719996.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 23]: https://img-blog.csdnimg.cn/20210109165916171.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 24]: https://img-blog.csdnimg.cn/2021010917000968.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 25]: https://img-blog.csdnimg.cn/20210109230239862.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 26]: https://img-blog.csdnimg.cn/20210109230910769.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 27]: https://img-blog.csdnimg.cn/20210109231234982.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 28]: https://img-blog.csdnimg.cn/20210109231546859.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 29]: https://img-blog.csdnimg.cn/20210109231623248.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 30]: https://img-blog.csdnimg.cn/20210109231718158.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 31]: https://img-blog.csdnimg.cn/20210109231834961.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 32]: https://img-blog.csdnimg.cn/20210109232536690.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 33]: https://img-blog.csdnimg.cn/20210110162016955.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 34]: https://img-blog.csdnimg.cn/20210110162115112.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 35]: https://img-blog.csdnimg.cn/20210110162436973.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 36]: https://img-blog.csdnimg.cn/20210110162837637.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ==,size_16,color_FFFFFF,t_70#pic_center