kafka入门

淡淡的烟草味﹌ 2022-10-15 00:44 183阅读 0赞

# **Kafka** 概 述 #

## 定义 ##

Kafka 是一个分布式的基于发布/订阅模式的消息队列（Message Queue），主要应用于大数据实时处理领域。

## 消息队列 ##

### MQ传统应用场景之异步处理 ###

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-Hw5sMd9k-1621689176911)(C:\\Users\\22051\\Desktop\\kafka\\kafka.assets\\image-20210520105400536.png)\]][img-Hw5sMd9k-1621689176911_C_Users_22051_Desktop_kafka_kafka.assets_image-20210520105400536.png]

### 使用消息队列的好处 ###

**解耦**

允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。

**可恢复性**

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

**缓冲**

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

**灵活性** & **峰值处理能力**

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

**异步通信**

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

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

**点对点模式**（一对一，消费者主动拉取数据，消息收到后消息清除）

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

消息生产者生产消息发送到Queue 中，然后消息消费者从Queue 中取出并且消费消息。消息被消费以后，queue 中不再有存储，所以消息消费者不可能消费到已经被消费的消息。 Queue 支持存在多个消费者， 但是对一个消息而言， 只会有一个消费者可以消费。

**发布/订阅模式**（一对多，消费者消费数据之后不会清除消息）

消息生产者（发布）将消息发布到 topic 中，同时有多个消息消费者（订阅）消费该消息。和点对点方式不同，发布到 topic 的消息会被所有订阅者消费。

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

# Kafka架构 #

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

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。

## Kafka **高效读写数据** ##

**顺序写磁盘**

Kafka 的 producer 生产数据，要写入到log 文件中，写的过程是一直追加到文件末端，为顺序写。官网有数据表明，同样的磁盘，顺序写能到 600M/s，而随机写只有 100K/s。这与磁盘的机械机构有关，顺序写之所以快，是因为其省去了大量磁头寻址的时间。

**零复制技术**

# **kafka部署** #

## 前提环境 ##

安装好java，而且是必须有jre的java版本，比如jdk 1.8。**jdk11就没有jre，会导致Eagle安装失败**。

## 安装zookeeper ##

注意：提前安装好java1.8

#  新建工作目录
    cd  /opt
    mkdir zookeeper
    cd /opt/zookeeper
    mkdir data
    mkdir logs
    
    //下载安装包
    cd /opt/zookeeper
    wget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.5/zookeeper-3.4.5.tar.gz
    tar -xf zookeeper-3.4.5.tar.gz
    
    //修改配置文件
    cd /opt/zookeeper/zookeeper-3.4.5/conf/
    cp zoo_sample.cfg zoo.cfg
    vi zoo.cfg
    dataDir=/opt/zookeeper/data
    dataLogDir=/opt/zookeeper/logs
    
    //启动zk
    cd /opt/zookeeper/zookeeper-3.4.5/bin/
    zkServer.sh start
    
    
    //检查
    netstat -lntp |grep 2181
    tcp6       0      0 :::2181                 :::*                    LISTEN      31220/java
    ./zkServer.sh status
    JMX enabled by default
    Using config: /opt/zookeeper/zookeeper-3.4.5/bin/../conf/zoo.cfg
    Mode: standalone

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

## 安装kafka ##

我们试着在

//新增工作目录
    cd /opt
    mkdir kafka01
    cd  /opt/kafka01
    mkdir /opt/kafka01/logs
    
    //下载kafka文件
    cd  /opt/kafka01
    wget https://mirrors.bfsu.edu.cn/apache/kafka/2.8.0/kafka_2.13-2.8.0.tgz
    tar -xf kafka_2.13-2.8.0.tgz
    
    
    
    
    
    //复制kafka文件
    scp -r /opt/kafka01 /opt/kafka02
    scp -r /opt/kafka01 /opt/kafka03
    
    
    //修改配置文件  
    /opt/kafka01/kafka_2.13-2.8.0/config/server.properties修改
    broker.id=0
    port=9091
    log.dirs=/opt/kafka01/logs
    
    
    /opt/kafka01/kafka_2.13-2.8.0/config/server.properties修改
    broker.id=1
    port=9092
    log.dirs=/opt/kafka02/logs
    
    
    
    /opt/kafka01/kafka_2.13-2.8.0/config/server.properties修改
    broker.id=2
    port=9093
    log.dirs=/opt/kafka03/logs
    
    
    
    
    
    //启动kafka
    nohup  /opt/kafka01/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka01/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &
    nohup  /opt/kafka02/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka02/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &
    nohup  /opt/kafka03/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka03/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &

## 安装Eagle ##

重新启动kafka

//关闭kafka
    
    /opt/kafka01/kafka_2.13-2.8.0/bin/kafka-server-stop.sh
    /opt/kafka02/kafka_2.13-2.8.0/bin/kafka-server-stop.sh
    /opt/kafka03/kafka_2.13-2.8.0/bin/kafka-server-stop.sh
    
    
    
    //修改配置kafka-server-start.sh命令文件
    
    vi  /opt/kafka01/kafka_2.13-2.8.0/bin/kafka-server-start.sh
    
    if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
        export JMX_PORT="9991"
        export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    fi
    
    
    
    vi  /opt/kafka02/kafka_2.13-2.8.0/bin/kafka-server-start.sh
    
    if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
        export JMX_PORT="9992"
        export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    fi
    
    
    
    vi  /opt/kafka03/kafka_2.13-2.8.0/bin/kafka-server-start.sh
    
    if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
        export JMX_PORT="9993"
        export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G"
    fi
    
    
    
    
    //重新启动kafka
    nohup  /opt/kafka01/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka01/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &
    nohup  /opt/kafka02/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka02/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &
    nohup  /opt/kafka03/kafka_2.13-2.8.0/bin/kafka-server-start.sh     /opt/kafka03/kafka_2.13-2.8.0/config/server.properties  >/dev/null 2>&1 &

安装eagle

//新增工作目录
    cd /opt
    mkdir eagle
    
    //下载安装包并上传到/opt/eagle/目录
    https://codeload.github.com/smartloli/kafka-eagle-bin/tar.gz/v2.0.5
    
    
    //解压文件
    
    tar -zxvf   kafka-eagle-bin-2.0.5.tar.gz
    cd  /opt/eagle/kafka-eagle-bin-2.0.5/
    tar -zxvf  kafka-eagle-web-2.0.5-bin.tar.gz
    
    
    // 修改配置文件
    cd /opt/eagle/kafka-eagle-bin-2.0.5/kafka-eagle-web-2.0.5/conf/
    vi system-config.properties
    
    改成如下配置：
    
    ######################################
    # multi zookeeper & kafka cluster list
    ######################################
    kafka.eagle.zk.cluster.alias=cluster1
    cluster1.zk.list=m1:2181     # zookeeper的地址
    
    ######################################
    # zookeeper enable acl
    ######################################
    #cluster1.zk.acl.enable=false
    #cluster1.zk.acl.schema=digest
    #cluster1.zk.acl.username=test
    #cluster1.zk.acl.password=test123
    
    ######################################
    # broker size online list
    ######################################
    cluster1.kafka.eagle.broker.size=20
    
    ######################################
    # zk client thread limit
    ######################################
    kafka.zk.limit.size=32
    
    ######################################
    # kafka eagle webui port
    ######################################
    kafka.eagle.webui.port=8049
    
    ######################################
    # kafka jmx acl and ssl authenticate
    ######################################
    #cluster1.kafka.eagle.jmx.acl=false
    #cluster1.kafka.eagle.jmx.user=keadmin
    #cluster1.kafka.eagle.jmx.password=keadmin123
    #cluster1.kafka.eagle.jmx.ssl=false
    #cluster1.kafka.eagle.jmx.truststore.location=/Users/dengjie/workspace/ssl/certificates/kafka.truststore
    #cluster1.kafka.eagle.jmx.truststore.password=ke123456
    
    ######################################
    # kafka offset storage
    ######################################
    cluster1.kafka.eagle.offset.storage=kafka
    # cluster2.kafka.eagle.offset.storage=zk
    
    ######################################
    # kafka jmx uri
    ######################################
    cluster1.kafka.eagle.jmx.uri=service:jmx:rmi:///jndi/rmi://%s/jmxrmi
    
    ######################################
    # kafka metrics, 15 days by default
    ######################################
    kafka.eagle.metrics.charts=true
    kafka.eagle.metrics.retain=15
    
    ######################################
    # kafka sql topic records max
    ######################################
    kafka.eagle.sql.topic.records.max=5000
    
    ######################################
    # delete kafka topic token
    ######################################
    kafka.eagle.topic.token=keadmin
    
    ######################################
    # kafka sasl authenticate
    ######################################
    cluster1.kafka.eagle.sasl.enable=false
    cluster1.kafka.eagle.sasl.protocol=SASL_PLAINTEXT
    cluster1.kafka.eagle.sasl.mechanism=SCRAM-SHA-256
    cluster1.kafka.eagle.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="kafka" password="kafka-eagle";
    cluster1.kafka.eagle.sasl.client.id=
    cluster1.kafka.eagle.blacklist.topics=
    cluster1.kafka.eagle.sasl.cgroup.enable=false
    cluster1.kafka.eagle.sasl.cgroup.topics=
    
    
    
    ######################################
    # kafka sqlite jdbc driver address
    ######################################
    #kafka.eagle.driver=org.sqlite.JDBC
    #kafka.eagle.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
    #kafka.eagle.username=root
    #kafka.eagle.password=www.kafka-eagle.org
    
    ######################################
    # kafka mysql jdbc driver address
    ######################################
    kafka.eagle.driver=com.mysql.cj.jdbc.Driver
    kafka.eagle.url=jdbc:mysql://m1:3306/ke?useUnicode=true&characterEncoding=UTF-8&zeroDateTimeBehavior=convertToNull
    kafka.eagle.username=root
    kafka.eagle.password=mysql123456
    
    
    //如果可用机器内存小于2G，还需要修改ke.sh的启动参数
    cd  /opt/eagle/kafka-eagle-bin-2.0.5/kafka-eagle-web-2.0.5/bin/
    vi  ke.sh
    比如将 KE_JAVA_OPTS设置最大可用内存成1G
    export KE_JAVA_OPTS="-server -Xmx1g -Xms1g -XX:MaxGCPauseMillis=20 -XX:+UseG1GC -XX:MetaspaceSize=128m -XX:InitiatingHeapOccupancyPercent=35 -XX:G1HeapRegionSize=16M -XX:MinMetaspaceFreeRatio=50 -XX:MaxMetaspaceFreeRatio=80"
    
    // 设置KE_HOME
    vi   /etc/profile
    
    export KE_HOME=/opt/eagle/kafka-eagle-bin-2.0.5/kafka-eagle-web-2.0.5/
    
    source  /etc/profile
    
    
    //启动eagle
    cd  /opt/eagle/kafka-eagle-bin-2.0.5/kafka-eagle-web-2.0.5/bin/
    ./ke.sh start

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

# Kafka **命令行操作** #

## 新增topic ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    
    ./kafka-topics.sh --zookeeper m1:2181 --create --replication-factor 3 --partitions 1 --topic testTopicName

选项说明：

–zookeeper zookeeper ip 地址

–topic 定义topic名

–replication-factor 定义副本数

–partitions 定义分区数

## 查看当前服务器中的所有 topic ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    
    ./kafka-topics.sh --zookeeper m1:2181 --list

![在这里插入图片描述][20210522211700357.png]

## 查看某个topic详情 ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    ./kafka-topics.sh --zookeeper m1:2181 --describe --topic testTopicName

![在这里插入图片描述][20210522211646880.png]

## 发送消息 ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    
    ./kafka-console-producer.sh --broker-list m1:9091 --topic testTopicName

![在这里插入图片描述][20210522211727703.png]

## 消费消息 ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    ./kafka-console-consumer.sh  --bootstrap-server m1:9091 --from-beginning --topic testTopicName

–from-beginning：会把first主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。

–bootstrap-server：生产消息的服务器

![在这里插入图片描述][20210522211744725.png]

## 修改分区数 ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    ./kafka-topics.sh --zookeeper m1:2181 --alter --topic testTopicName --partitions 6

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-NWvop0EH-1621689176921)(C:\\Users\\22051\\Desktop\\kafka\\kafka.assets\\image-20210521164119464.png)\]][img-NWvop0EH-1621689176921_C_Users_22051_Desktop_kafka_kafka.assets_image-20210521164119464.png]

## 删除 topic ##

cd /opt/kafka01/kafka_2.13-2.8.0/bin/
    ./kafka-topics.sh --zookeeper m1:2181 --delete --topic testTopicName

![在这里插入图片描述][20210522211807608.png]

需要 server.properties 中设置 delete.topic.enable=true 否则只是标记删除。

# Kafka 架构深入 #

## Kafka **工作流程及文件存储机制** ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 5]

Kafka部分名词解释如下：

1、Broker：消息中间件处理结点，一个Kafka节点就是一个broker，多个broker可以组成一个Kafka集群。
    2、Topic：一类消息，例如page view日志、click日志等都可以以topic的形式存在，Kafka集群能够同时负责多个topic的分发。
    3、Partition：topic物理上的分组，一个topic可以分为多个partition，每个partition是一个有序的队列。
    4、Segment：partition物理上由多个segment组成，下面2.3和2.4有详细说明。
    5、offset：每个partition都由一系列有序的、不可变的消息组成，这些消息被连续的追加到partition中。partition中的每个消息都有一个连续的序列号叫做offset,用于partition唯一标识一条消息.

Kafka 中消息是以 **topic** 进行分类的，生产者生产消息，消费者消费消息，都是面向 topic的。

topic 是逻辑上的概念，而 partition 是物理上的概念，每个 partition 对应于一个 log 文件，该 log 文件中存储的就是 producer 生产的数据。Producer 生产的数据会被不断追加到该 log 文件末端，且每条数据都有自己的 offset。消费者组中的每个消费者，都会实时记录自己消费到了哪个offset，以便出错恢复时，从上次的位置继续消费。

由于生产者生产的消息会不断追加到 log 文件末尾，为防止 log 文件过大导致数据定位效率低下，Kafka 采取了**分片**和**索引**机制，将每个 partition 分为多个 segment。每个 segment对应两个文件——“.index”文件和“.log”文件。这些文件位于一个文件夹下，该文件夹的命名规则为：topic 名称+分区序号。例如：

![在这里插入图片描述][20210522211850137.png]

index 和 log 文件以当前 segment 的第一条消息的 offset 命名。“.index”文件存储大量的索引信息，“.log”文件存储大量的数据，索引文件中的元数据指向对应数据文件中message 的物理偏移地址。

## 消息发送流程 ##

Kafka 的 Producer 发送消息采用的是**异步发送**的方式。在消息发送的过程中，涉及到了两个线程——**main线程**和 Sender 线程，以及一个线程共享变量——**RecordAccumulator**。 main 线程将消息发送给RecordAccumulator，**Sender 线程**不断从 RecordAccumulator 中拉取消息发送到Kafka broker。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 6]

相关参数：

batch.size：只有数据积累到 batch.size 之后，sender 才会发送数据。
    
    linger.ms：如果数据迟迟未达到 batch.size，sender 等待 linger.time 之后就会发送数据。

## Kafka **生产者** ##

### 分区策略 ###

#### 分区的原因 ####

**方便在集群中扩展**，每个 Partition 可以通过调整以适应它所在的机器，而一个 topic又可以有多个 Partition 组成，因此整个集群就可以适应任意大小的数据了；

**可以提高并发**，因为可以以Partition 为单位读写了。

#### 分区的原则 ####

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

![在这里插入图片描述][20210522211936349.png]

1.  指明 partition 的情况下，直接将指明的值直接作为 partiton 值；
2.  没有指明 partition 值但有 key 的情况下，将 key 的 hash 值与 topic 的 partition数进行取余得到 partition 值；
3.  既没有 partition 值又没有 key 值的情况下，第一次调用时随机生成一个整数（后面每次调用在这个整数上自增），将这个值与 topic 可用的 partition 总数取余得到 partition值，也就是常说的 round-robin 算法。

### 数据可靠性保证 ###

**为保证 producer 发送的数据，能可靠的发送到指定的 topic，topic 的每个 partition 收到 producer 发送的数据后，都需要向 producer 发送 ack（acknowledgement 确认收到），如果 producer 收到 ack，就会进行下一轮的发送，否则重新发送数据。**

副本数据同步策略

<table> 
 <thead> 
  <tr> 
   <th>方案</th> 
   <th>优点</th> 
   <th>缺点</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>半数以上完成同步，就发送 <strong>ack</strong></td> 
   <td>延迟低</td> 
   <td>选举新的 leader 时，容忍 n 台节点的故障，需要 2n+1 个副本</td> 
  </tr> 
  <tr> 
   <td>全部完成同步，才发送ack</td> 
   <td>选举新的 leader 时，容忍 n 台节点的故障，需要 n+1 个副本</td> 
   <td>延迟高</td> 
  </tr> 
 </tbody> 
</table>

Kafka 选择了第二种方案，原因如下：

1.  同样为了容忍 n 台节点的故障，第一种方案需要 2n+1 个副本，而第二种方案只需要 n+1个副本，而Kafka 的每个分区都有大量的数据，第一种方案会造成大量数据的冗余。
2.  虽然第二种方案的网络延迟会比较高，但网络延迟对 Kafka 的影响较小。

#### ISR ####

采用第二种方案之后，设想以下情景：leader 收到数据，所有 follower 都开始同步数据，但有一个 follower，因为某种故障，迟迟不能与 leader 进行同步，那 leader 就要一直等下去，直到它完成同步，才能发送 ack。这个问题怎么解决呢？

Leader 维护了一个动态的 in-sync replica set (ISR)，意为和 leader 保持同步的 follower 集合。当 ISR 中的 follower 完成数据的同步之后，leader 就会给 follower 发送 ack。如果 follower长时间 未 向 leader 同 步 数 据 ， 则 该 follower 将 被 踢 出 ISR ， 该 时 间 阈 值 由**replica.lag.time.max.ms** 参数设定。Leader 发生故障之后，就会从 ISR 中选举新的 leader。

#### ack **应答机制** ####

对于某些不太重要的数据，对数据的可靠性要求不是很高，能够容忍数据的少量丢失，所以没必要等ISR 中的 follower 全部接收成功。

所以 Kafka 为用户提供了三种可靠性级别，用户根据对可靠性和延迟的要求进行权衡，选择以下的配置。

acks 参数配置：

acks：

0：producer 不等待 broker 的 ack，这一操作提供了一个最低的延迟，broker 一接收到还没有写入磁盘就已经返回，当 broker 故障时有可能**丢失数据**；

1：producer 等待broker 的 ack，partition 的 leader 落盘成功后返回ack，如果在 follower同步成功之前leader 故障，那么将会**丢失数据**；

\-1（all）：producer 等待 broker 的 ack，partition 的 leader 和 follower 全部落盘成功后才返回 ack。但是如果在 follower 同步完成后，broker 发送 ack 之前，leader 发生故障，那么会造成**数据重复**。

#### Log文件中的HW和LEO ####

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

HW：所有副本中最小的LEO

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 7]

#### follower **故障** ####

follower 发生故障后会被临时踢出 ISR，待该 follower 恢复后，follower 会读取本地磁盘记录的上次的HW，并将 log 文件高于 HW 的部分截取掉，从 HW 开始向 leader 进行同步。等该 follower **的** LEO **大于等于该** Partition **的** HW，即 follower 追上 leader 之后，就可以重新加入 ISR 了。

#### leader **故障** ####

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

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

#### Exactly Once **语义** ####

将服务器的 ACK 级别设置为-1，可以保证 Producer 到 Server 之间不会丢失数据，即 At Least Once 语义。相对的，将服务器 ACK 级别设置为 0，可以保证生产者每条消息只会被发送一次，即At Most Once 语义。

At Least Once 可以保证数据不丢失，但是不能保证数据不重复；相对的，At Least Once可以保证数据不重复，但是不能保证数据不丢失。但是，对于一些非常重要的信息，比如说交易数据，下游数据消费者要求数据既不重复也不丢失，即 Exactly Once 语义。在 0.11 版本以前的 Kafka，对此是无能为力的，只能保证数据不丢失，再在下游消费者对数据做全局去重。对于多个下游应用的情况，每个都需要单独做全局去重，这就对性能造成了很大影响。

0.11 版本的 Kafka，引入了一项重大特性：幂等性。所谓的幂等性就是指Producer 不论向 Server 发送多少次重复数据，Server 端都只会持久化一条。幂等性结合 At Least Once 语义，就构成了Kafka 的Exactly Once 语义。即：

At Least Once + 幂等性 = Exactly Once

要启用幂等性，只需要将 Producer 的参数中 enable.idompotence 设置为 true 即可。Kafka的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在初始化的时候会被分配一个 PID，发往同一 Partition 的消息会附带 Sequence Number。而 Broker 端会对<PID, Partition, SeqNumber>做缓存，当具有相同主键的消息提交时，Broker 只会持久化一条。

但是PID 重启就会变化，同时不同的 Partition 也具有不同主键，所以幂等性无法保证跨分区跨会话的Exactly Once。

基于事务发送官方案例 （原子性和幂等性）:

Properties props = new Properties();
    props.put("bootstrap.servers", "localhost:9092");
    props.put("transactional.id", "my-transactional-id");
    Producer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), 
    new StringSerializer());
    
    producer.initTransactions();
    
    try {
        producer.beginTransaction();
        for (int i = 0; i < 100; i++)
            producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
        // We can't recover from these exceptions, so our only option is to close the producer and exit.
        producer.close();
    } catch (KafkaException e) {
        // For all other exceptions, just abort the transaction and try again.
        producer.abortTransaction();
    }
    producer.close();

## Kafka **消费者** ##

### 消费方式 ###

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

push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由 broker 决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成 consumer 来不及处理消息，

典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。

\*\*pull 模式不足之处是，如果 kafka 没有数据，消费者可能会陷入循环中，一直返回空数据。\*\*针对这一点，Kafka 的消费者在消费数据时会传入一个时长参数 timeout，如果当前没有数据可供消费，consumer 会等待一段时间之后再返回，这段时长即为timeout。

### 分区分配策略 ###

一个 consumer group 中有多个 consumer，一个 topic 有多个 partition，所以必然会涉及到 partition 的分配问题，即确定那个 partition 由哪个 consumer 来消费。

Kafka 有两种分配策略，一是 RoundRobin，一是Range。

### offset **的维护** ###

由于 consumer 在消费过程中可能会出现断电宕机等故障，consumer 恢复后，需要从故障前的位置的继续消费，所以 consumer 需要实时记录自己消费到了哪个 offset，以便故障恢复后继续消费。

## Zookeeper **在** Kafka **中的作用** ##

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

Controller 的管理工作都是依赖于 Zookeeper 的。以下为 partition 的 leader 选举过程：

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

## Kafka **事务** ##

Kafka 从 0.11 版本开始引入了事务支持。事务可以保证 Kafka 在Exactly Once 语义的基础上，生产和消费可以跨分区和会话，要么全部成功，要么全部失败。

### Producer **事务** ###

为了实现跨分区跨会话的事务，需要引入一个全局唯一的 Transaction ID，并将 Producer获得的PID 和Transaction ID 绑定。这样当Producer 重启后就可以通过正在进行的Transaction ID 获得原来的PID。

为了管理Transaction，Kafka 引入了一个新的组件Transaction Coordinator。Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。Transaction Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic，这样即使整个服务重启，由于事务状态得到保存，进行中的事务状态可以得到恢复，从而继续进行。

### Consumer **事务** ###

上述事务机制主要是从Producer 方面考虑，对于 Consumer 而言，事务的保证就会相对较弱，尤其时无法保证 Commit 的信息被精确消费。这是由于 Consumer 可以通过 offset 访问任意信息，而且不同的 Segment File 生命周期不同，同一事务的消息可能会出现重启后被删除的情况。

## Kafka API ##

### Producer API ###

需要用到的类：

![\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-mlIWyZ06-1621689176925)(kafka.assets/image-20210522183143523.png)\]][img-mlIWyZ06-1621689176925_kafka.assets_image-20210522183143523.png]

1、不带回调函数的API
     public Future<RecordMetadata> send(ProducerRecord<K, V> record)
    2、带回调函数的API
    public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)

### Consumer API ###

Consumer 消费数据时的可靠性是很容易保证的，因为数据在 Kafka 中是持久化的，故不用担心数据丢失问题。

**所以 offset 的维护是 Consumer 消费数据是必须考虑的问题。**

需要用到的类：

KafkaConsumer：需要创建一个消费者对象，用来消费数据 
    
    ConsumerConfig：获取所需的一系列配置参数
    
    ConsuemrRecord：每条数据都要封装成一个 ConsumerRecord 对象

为了使我们能够专注于自己的业务逻辑，Kafka 提供了自动提交offset 的功能。自动提交 offset 的相关参数：

enable.auto.commit：是否开启自动提交 offset 功能
    auto.commit.interval.ms 自动提交 offset的时间间隔

#### 手动提交offset ####

虽然自动提交 offset 十分简介便利，但由于其是基于时间提交的，开发人员难以把握offset 提交的时机。因此 Kafka 还提供了手动提交 offset 的 API。

手动提交 offset 的方法有两种：分别是 commitSync（同步提交）和 commitAsync（异步提交）。两者的相同点是，都会将本次 poll 的一批数据最高的偏移量提交；不同点是， commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而 commitAsync 则没有失败重试机制，故有可能提交失败。

手动提交有两种调用方式：

//同步提交
    void commitSync();
    //异步提交
    void commitAsync();

## 数据漏消费和重复消费分析 ##

无论是同步提交还是异步提交 offset，都有可能会造成数据的漏消费或者重复消费。先提交 offset 后消费，有可能造成数据的漏消费；而先消费后提交 offset，有可能会造成数据的重复消费。

## 自定义存储 offset-ConsumerRebalanceListener ##

Kafka 0.9 版本之前，offset 存储在 zookeeper，0.9 版本及之后，默认将 offset 存储在 Kafka的一个内置的topic 中。除此之外，Kafka 还可以选择自定义存储 offset。 offset 的维护是相当繁琐的，**因为需要考虑到消费者的 Rebalace。**

当有新的消费者加入消费者组、已有的消费者推出消费者组或者所订阅的主题的分区发生变化，就会触发到分区的重新分配，重新分配的过程叫做Rebalance。

消费者发生 Rebalance 之后，每个消费者消费的分区就会发生变化。因此消费者要首先获取到自己被重新分配到的分区，并且**定位到每个分区最近提交的 offset 位置继续消费。要实现自定义存储 offset，需要借助 ConsumerRebalanceListener**，需要根据所选的 offset 存储系统自行实现。

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

自己实现ConsumerRebalanceListener，再绑定到KafkaConsumer的subscribe即可

public void subscribe(Collection<String> topics, ConsumerRebalanceListener listener)

## Interceptor ##

拦截器原理

Producer 拦截器(interceptor)是在 Kafka 0.10 版本被引入的，主要用于实现 clients 端的定制化控制逻辑。

对于 producer 而言，interceptor 使得用户在消息发送前以及 producer 回调逻辑前有机会对消息做一些定制化需求，比如修改消息等。同时，producer 允许用户指定多个 interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)。Intercetpor 的实现接口是 org.apache.kafka.clients.producer.ProducerInterceptor，其定义的方法包括：

![在这里插入图片描述][20210522212138310.png]

1、configure(configs) 获取配置信息和初始化数据时调用。

2、onSend(ProducerRecord) 该方法封装进 KafkaProducer.send 方法中，即它运行在用户主线程中。Producer 确保在消息被序列化以及计算分区前调用该方法。用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的 topic 和分区，否则会影响目标分区的计算。

3、onAcknowledgement(RecordMetadata, Exception)：

该方法会在消息从RecordAccumulator 成功发送到 Kafka Broker 之后，或者在发送过程中失败时调用。并且通常都是在 producer 回调逻辑触发之前。onAcknowledgement 运行在 producer 的 IO 线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢 producer 的消息发送效率。

4、close：

关闭 interceptor，主要用于执行一些资源清理工作

如前所述，interceptor 可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全。另外倘若指定了多个 interceptor，则 producer 将按照指定顺序调用它们，并仅仅是捕获每个interceptor 可能抛出的异常记录到错误日志中而非在向上传递。这在使用过程中要特别留意。

[img-Hw5sMd9k-1621689176911_C_Users_22051_Desktop_kafka_kafka.assets_image-20210520105400536.png]: /images/20221014/73e810d7fe374123bae74620fdfc4b02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70]: /images/20221014/f968ce088271490ea85686a434a6c8df.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 1]: /images/20221014/8fef01d260294be1956ea7afcb2f4dfa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 2]: /images/20221014/aef53dde81ce4124994f5e72b92c6d04.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 3]: /images/20221014/8f5022ad033a4b04858764a5128faf04.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 4]: /images/20221014/5f734fe3b1274dfa8bcb6bacf4c21e38.png
[20210522211700357.png]: /images/20221014/149f3a4a83814ccb968ec7ad129b1f1d.png
[20210522211646880.png]: /images/20221014/b10aa08920f2453cbf496fb0150d9189.png
[20210522211727703.png]: /images/20221014/f5e569ff59e849df9129285561212020.png
[20210522211744725.png]: /images/20221014/5594f851b13a4479a9f1cf59dde39bf1.png
[img-NWvop0EH-1621689176921_C_Users_22051_Desktop_kafka_kafka.assets_image-20210521164119464.png]: /images/20221014/369fed385adf4452bc393da265818d5f.png
[20210522211807608.png]: /images/20221014/a03b975037ef4e7291de2b783ffa3523.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 5]: /images/20221014/a4b529c53d8d4e5385d4a6572efc660c.png
[20210522211850137.png]: /images/20221014/b541572a1ae948d28152c654d387d6fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 6]: /images/20221014/4ac634e7414845bc846985656a31044e.png
[20210522211936349.png]: /images/20221014/e52caeb80d6c42029bbca7d31ea7717a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 7]: /images/20221014/10775c8c04f94e54bd5a4805f65f8d07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 8]: /images/20221014/ceb35f71331d4d7e9cb9af76e4e12719.png
[img-mlIWyZ06-1621689176925_kafka.assets_image-20210522183143523.png]: /images/20221014/bee08e81cee94c2689018192b971bffb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L1pIT05HWkVXRUk_size_16_color_FFFFFF_t_70 9]: /images/20221014/6d6e1138755444b3929547bce113a0b5.png
[20210522212138310.png]: /images/20221014/c49b0d7b92a145dfa9ea95423cdf6e91.png