KafKa的安装与使用

冷不防 2022-10-12 05:42 214阅读 0赞

# 一、 KafKa #

##    1.1 Kafka是什么 ##

在流式计算中，Kafka一般用来缓存数据，Storm通过消费Kafka的数据进行计算。  
1）Apache Kafka是一个开源消息系统，由Scala写成。是由Apache软件基金会开发的一个开源消息系统项目。  
2）Kafka最初是由LinkedIn开发，并于2011年初开源。2012年10月从Apache Incubator毕业。该项目的目标是为处理实时数据提供一个统一、高通量、低等待的平台。  
3）Kafka是一个分布式消息队列。Kafka对消息保存时根据Topic进行归类，发送消息者称为Producer，消息接受者称为Consumer，此外kafka集群有多个kafka实例组成，每个实例(server)成为broker。  
4）无论是kafka集群，还是producer和consumer都依赖于zookeeper集群保存一些meta信息，来保证系统可用性

##    1.2 Kafka内部实现原理 ##

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16][]

（1）点对点模式（一对一，消费者主动拉取数据，消息收到后消息清除）  
点对点模型通常是一个基于拉取或者轮询的消息传送模型，这种模型从队列中请求信息，而不是将消息推送到客户端。这个模型的特点是发送到队列的消息被一个且只有一个接收者接收处理，即使有多个消息监听者也是如此。  
（2）发布/订阅模式（一对多，数据生产后，推送给所有订阅者）  
发布订阅模型则是一个基于推送的消息传送模型。发布订阅模型可以有多种不同的订阅者，临时订阅者只在主动监听主题时才接收消息，而持久订阅者则监听主题的所有消息，即使当前订阅者不可用，处于离线状态。

##    1.3 为什么需要消息队列 ##

1）解耦：  
　　允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束。  
2）冗余：  
消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。  
3）扩展性：  
因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。  
4）灵活性 & 峰值处理能力：  
在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。  
5）可恢复性：  
系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。  
6）顺序保证：  
在大多使用场景下，数据处理的顺序都很重要。大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka保证一个Partition内的消息的有序性）  
7）缓冲：  
有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。  
8）异步通信：  
很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

##    1.4 Kafka架构 ##

1）Producer ：消息生产者，就是向kafka broker发消息的客户端。  
2）Consumer ：消息消费者，向kafka broker取消息的客户端  
3）Topic ：可以理解为一个队列。  
4） Consumer Group （CG）：这是kafka用来实现一个topic消息的广播（发给所有的consumer）和单播（发给任意一个consumer）的手段。一个topic可以有多个CG。topic的消息会复制（不是真的复制，是概念上的）到所有的CG，但每个partion只会把消息发给该CG中的一个consumer。如果需要实现广播，只要每个consumer有一个独立的CG就可以了。要实现单播只要所有的consumer在同一个CG。用CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic。  
5）Broker ：一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topic。  
6）Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。  
7）Offset：kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找。例如你想找位于2049的位置，只要找到2048.kafka的文件即可。当然the first offset就是00000000000.kafka

#    二 Kafka集群部署 #

##    2.1 环境准备 ##

###    2.1.1 集群规划 ###

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td style="text-align:center;">bigdata01</td> 
   <td style="text-align:center;">bigdata02</td> 
   <td style="text-align:center;">bigdata03</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;">zookeeperk</td> 
   <td style="text-align:center;">zookeeperk</td> 
   <td style="text-align:center;">zookeeperk</td> 
  </tr> 
  <tr> 
   <td style="text-align:center;">kafka</td> 
   <td style="text-align:center;">kafka</td> 
   <td style="text-align:center;">kafka</td> 
  </tr> 
 </tbody> 
</table>

###    2.1.2 kafka的gz包下载 ###

[Apache Kafka][]

##    2.2 Kafka集群部署 ##

修改配置文件：

/mine/kafka_2.13-2.8.0/config

vim server.properties

#broker的全局唯一编号，不能重复
    broker.id=1
    #删除topic功能使能
    delete.topic.enable=true
    #kafka运行日志存放的路径
    log.dirs=/mine/kafka_2.13-2.8.0/config/logs
    #配置连接Zookeeper集群地址
    zookeeper.connect=bigdata01:2181,bigdata02:2181,bigdata03:2181

mkdir logs

![在这里插入图片描述][2020121915454619.png]  
配置环境变量：

vim /etc/profile

export JAVA_HOME=/mine/jdk1.8.0_201
    export ZOOKEEPER_HOME=/mine/apache-zookeeper-3.7.0-bin
    export KAFKA_HOME=/mine/kafka_2.13-2.8.0
    export PATH=$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$KAFKA_HOME/bin:$PATH

分别在bigdata02和bigdata03拷贝kafka的安装包，并在机器上修改配置文件/mine/kafka\_2.13-2.8.0/config/server.properties中的broker.id=2、broker.id=3

###    注：broker.id不得重复 ###

###    启动KAFKA集群： ###

在bigdata01、bigdata02、bigdata03上启动kafka：

kafka-server-start.sh /mine/kafka_2.13-2.8.0/config/server.properties &

![20210704102322312.png][]

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

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

kafka-topics.sh --zookeeper bigdata01:2181 --list

### 2、创建 topic  ###

kafka-topics.sh --zookeeper bigdata01:2181 --create --replication-factor 3 --partitions 2 --topic first

选项说明：

\--topic  定义  topic  名

\--replication-factor  定义副本数

\--partitions  定义分区数

### 3、删除 topic  ###

kafka-topics.sh --zookeeper bigdata01:2181 --delete --topic first

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

### **4、**发送消息 ###

kafka-console-producer.sh --broker-list bigdata01:9092 --topic first

![20210704112150573.png][]

###  5、消费消息 ###

kafka-console-consumer.sh --bootstrap-server bigdata01:9092 --topic first

kafka-console-consumer.sh --bootstrap-server bigdata01:9092 --from-beginning  --topic first

\--from-beginning ：会把主题中以往所有的数据都读取出来。

### 6、查看某个 Topic 的详情 ###

kafka-topics.sh --zookeeper bigdata01:2181 --describe --topic first

![20210704144608609.png][]

### 7、修改分区数  ###

kafka-topics.sh --zookeeper bigdata01:2181 --alter --topic first --partitions 6

# **二、Kafka** **架构深入** #

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70][]

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

的。

topic  是逻辑上的概念，而  partition  是物理上的概念，每个  partition  对应于一个  log  文

件，该  log  文件中存储的就是  producer  生产的数据。 Producer  生产的数据会被不断追加到该

log  文件末端，且每条数据都有自己的  offset 。消费者组中的每个消费者，都会实时记录自己

消费到了哪个  offset ，以便出错恢复时，从上次的位置继续消费。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1][]

由于生产者生产的消息会不断追加到  log  文件末尾，为防止  log  文件过大导致数据定位

效率低下， Kafka  采取了 **分片** 和 **索引** 机制，将每个  partition  分为多个  segment 。每个  segment

对应两个文件—— “.index” 文件和 “.log” 文件。这些文件位于一个文件夹下，该文件夹的命名

规则为： topic  名称 \+ 分区序号。例如， first  这个  topic  有三个分区，则其对应的文件夹为  first-

0,first-1,first-2 。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 2][]

index  和  log  文件以当前  segment  的第一条消息的  offset  命名。下图为  index  文件和  log

文件的结构示意图。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 3][]

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

###  2.2、**Kafka** **生产者** ###

### **2.2.1、分区策略** ###

**1** **）分区的原因**

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

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

**2** **）分区的原则**

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

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

（ 2 ）没有指明  partition  值但有  key  的情况下，将  key  的  hash  值与  topic  的  partition

数进行取余得到  partition  值；

（ 3 ）既没有  partition  值又没有  key  值的情况下，第一次调用时随机生成一个整数（后

面每次调用在这个整数上自增），将这个值与  topic  可用的  partition  总数取余得到  partition

值，也就是常说的  round-robin  算法。

### **2.2.2、数据可靠性保证**  ###

为保证  producer  发送的数据，能可靠的发送到指定的  topic ， topic  的每个  partition  收到

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

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 4][]

## 2.3、**Kafka** **消费者** ##

### **2.3.1、消费方式**  ###

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

push （推）模式很难适应消费速率不同的消费者，因为消息发送速率是由  broker  决定的。

它的目标是尽可能以最快速度传递消息，但是这样很容易造成  consumer 来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而 pull 模式则可以根据 consumer 的消费能力以适当的速率消费消息。

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

### **2.3.2、分区分配策略**  ###

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

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

### **2.3.3、offset** **的维护**  ###

由于  consumer  在消费过程中可能会出现断电宕机等故障， consumer  恢复后，需要从故

障前的位置的继续消费，所以  consumer  需要实时记录自己消费到了哪个  offset ，以便故障恢

复后继续消费。

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

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

# 三、**Kafka API**  #

## 3.1、**Producer API**  ##

### 3.1.1、**消息发送流程** ###

Kafka  的  Producer  发送消息采用的是 **异步发送** 的方式。在消息发送的过程中，涉及到了

**两个线程——** **main**  **线程和**  **Sender**  **线程** ，以及 **一个线程共享变量——** **RecordAccumulator** 。

main  线程将消息发送给  RecordAccumulator ， Sender  线程不断从  RecordAccumulator  中拉取

消息发送到  Kafka broker 。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 5][]

**相关参数：**

**batch.size** **：** 只有数据积累到  batch.size  之后， sender  才会发送数据。

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

### 3.1.2、**异步发送** **API** ###

**1** **）导入依赖**

<dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.0</version>
    </dependency>

**2****）编写代码**

需要用到的类：

**KafkaProducer** ：需要创建一个生产者对象，用来发送数据

**ProducerConfig** ：获取所需的一系列配置参数

**ProducerRecord** ：每条数据都要封装成一个  ProducerRecord  对象

**1.** **不带回调函数的**  **API**

package wxjwtl.cn;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.Producer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ProducerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("acks", "all");
            //重试次数
            properties.put("retries", 1);
            //批次大小
            properties.put("batch.size", 16384);
            //等待时间
            properties.put("linger.ms", 1);
            //RecordAccumulator 缓冲区大小
            properties.put("buffer.memory", 33554432);
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            Producer<String, String> producer = new KafkaProducer<>(properties);
            for (int i=0;i<100;i++){
                producer.send(new ProducerRecord<String,String>("first",Integer.toString(i), UUID.randomUUID().toString()));
            }
    
            producer.close();
        }
    }

**2.** **带回调函数的**  **API**

回调函数会在  producer  收到  ack  时调用，为异步调用，该方法有两个参数，分别是 RecordMetadata 和  Exception ，如果  Exception  为  null ，说明消息发送成功，如果

Exception  不为  null ，说明消息发送失败。

注意：消息发送失败会自动重试，不需要我们在回调函数中手动重试。

package wxjwtl.cn;
    
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Properties;
    import java.util.UUID;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ProducerTest  {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("acks", "all");
            //重试次数
            properties.put("retries", 1);
            //批次大小
            properties.put("batch.size", 16384);
            //等待时间
            properties.put("linger.ms", 1);
            //RecordAccumulator 缓冲区大小
            properties.put("buffer.memory", 33554432);
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            Producer<String, String> producer = new KafkaProducer<>(properties);
            for (int i=0;i<100;i++){
                producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), UUID.randomUUID().toString()), new Callback() {
                    //回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用
                    @Override
                    public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                        System.out.println(recordMetadata.topic());
                        System.out.println(recordMetadata.partition());
                        System.out.println(recordMetadata.timestamp());
                        System.out.println(null != e);
                    }
                });
            }
    
            producer.close();
        }
    }

**3、同步发送**  **API**

同步发送的意思就是，一条消息发送之后，会阻塞当前线程，直至返回  ack 。

由于  send  方法返回的是一个  Future  对象，根据  Futrue  对象的特点，我们也可以实现同步发送的效果，只需在调用 Future  对象的  get  方发即可。

package wxjwtl.cn;
    
    import org.apache.kafka.clients.producer.*;
    
    import java.util.Properties;
    import java.util.UUID;
    import java.util.concurrent.Future;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ProducerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("acks", "all");
            //重试次数
            properties.put("retries", 1);
            //批次大小
            properties.put("batch.size", 16384);
            //等待时间
            properties.put("linger.ms", 1);
            //RecordAccumulator 缓冲区大小
            properties.put("buffer.memory", 33554432);
            properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            Producer<String, String> producer = new KafkaProducer<>(properties);
            for (int i=0;i<10;i++){
                Future<RecordMetadata> recordMetadataFuture = producer.send(new ProducerRecord<String, String>("first", Integer.toString(i), UUID.randomUUID().toString()));
    
                RecordMetadata recordMetadata = recordMetadataFuture.get();
                System.out.println(recordMetadata.topic());
                System.out.println(recordMetadata.partition());
                System.out.println(recordMetadata.timestamp());
                System.out.println(recordMetadata.offset());
            }
    
            producer.close();
        }
    }

## **4、Consumer API**  ##

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

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

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

提交策略如何选择

Kafka提供了3种提交offset的方式

1、自动提交

// 自动提交,默认true
    properties.put("enable.auto.commit", "true");
    // 设置自动每1s提交一次
    properties.put("auto.commit.interval.ms", "1000");

2、手动同步提交offset

consumer.commitSync();

3、手动异步提交offset

consumer.commitAsync();

### 4.1、**自动提交** **offset**  ###

需要用到的类：

**KafkaConsumer** ：需要创建一个消费者对象，用来消费数据

**ConsumerConfig** ：获取所需的一系列配置参数

**ConsuemrRecord** ：每条数据都要封装成一个  ConsumerRecord  对象

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

自动提交  offset  的相关参数：

**enable.auto.commit** **：** 是否开启自动提交  offset  功能

**auto.commit.interval.ms** **：** 自动提交  offset  的时间间隔

package wxjwtl.cn;
    
    import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ConsumerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("group.id", "firstGroup");
    
            // 可选设置属性
            properties.put("enable.auto.commit", "true");
            // 自动提交offset,每1s提交一次
            properties.put("auto.commit.interval.ms", "1000");
            properties.put("auto.offset.reset","earliest");
            properties.put("client.id", "wxjwtl_client_id");
    
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            Consumer<String,String> consumer = new KafkaConsumer<>(properties);
            consumer.subscribe(Collections.singletonList("first"));
    
            while (true){
                ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                consumerRecords.forEach(consumerRecord->{
                    System.out.printf("topic = %s ,partition = %d,offset = %d, key = %s, value = %s%n",
                            consumerRecord.topic(), consumerRecord.partition(),
                            consumerRecord.offset(), consumerRecord.key(), consumerRecord.value());
                });
            }
        }
    }

### **4.2、手动提交** **offset**  ###

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

手动提交 offset  的方法有两种：分别是  commitSync （同步提交） 和  commitAsync （异步

提交） 。两者的相同点是，都会将 **本次**  **poll**  **的一批数据最高的偏移量提交** ；不同点是，

commitSync  阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，

也会出现提交失败）；而  commitAsync  则没有失败重试机制，故有可能提交失败。

**1** **）同步提交**  **offset**

由于同步提交  offset  有失败重试机制，故更加可靠，以下为同步提交  offset  的示例。

package wxjwtl.cn;
    
    import org.apache.kafka.clients.consumer.Consumer;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ConsumerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("group.id", "first-Group");
    
            // 可选设置属性
    //        properties.put("enable.auto.commit", "true");
    //        // 自动提交offset,每1s提交一次
    //        properties.put("auto.commit.interval.ms", "1000");
    //        properties.put("auto.offset.reset","earliest");
    //        properties.put("client.id", "wxjwtl_client_id");
    
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            Consumer<String,String> consumer = new KafkaConsumer<>(properties);
            consumer.subscribe(Collections.singletonList("first"));
    
            while (true){
                ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                consumerRecords.forEach(consumerRecord->{
                    System.out.printf("topic = %s ,partition = %d,offset = %d, key = %s, value = %s%n",
                            consumerRecord.topic(), consumerRecord.partition(),
                            consumerRecord.offset(), consumerRecord.key(), consumerRecord.value());
                });
    
                consumer.commitSync();
            }
        }
    }

**2** **）异步提交**  **offset**

虽然同步提交 offset  更可靠一些，但是由于其会阻塞当前线程，直到提交成功。因此吞吐量会收到很大的影响。因此更多的情况下，会选用异步提交 offset  的方式。

以下为异步提交 offset  的示例：

package wxjwtl.cn;
    
    import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.TopicPartition;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Map;
    import java.util.Properties;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ConsumerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("group.id", "first-Group");
    
            // 可选设置属性
    //        properties.put("enable.auto.commit", "true");
    //        // 自动提交offset,每1s提交一次
    //        properties.put("auto.commit.interval.ms", "1000");
    //        properties.put("auto.offset.reset","earliest");
    //        properties.put("client.id", "wxjwtl_client_id");
    
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            Consumer<String,String> consumer = new KafkaConsumer<>(properties);
            consumer.subscribe(Collections.singletonList("first"));
    
            while (true){
                ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                consumerRecords.forEach(consumerRecord->{
                    System.out.printf("topic = %s ,partition = %d,offset = %d, key = %s, value = %s%n",
                            consumerRecord.topic(), consumerRecord.partition(),
                            consumerRecord.offset(), consumerRecord.key(), consumerRecord.value());
                });
    
                consumer.commitAsync(new OffsetCommitCallback() {
                    异步提交
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
                        System.out.println(null != e);
                    }
                });
            }
        }
    }

### **4.3、自定义存储** **offset**  ###

Kafka 0.9 版本之前， offset  存储在  zookeeper ， 0.9  版本及之后，默认将  offset  存储在  Kafka

的一个内置的  topic  中。除此之外， Kafka  还可以选择自定义存储  offset 。

offset  的维护是相当繁琐的，因为需要考虑到消费者的  Rebalace 。

当有新的消费者加入消费者组、已有的消费者推出消费者组或者所订阅的主题的分区发

生变化，就会触发到分区的重新分配，重新分配的过程叫做  Rebalance 。

消费者发生  Rebalance  之后，每个消费者消费的分区就会发生变化。 因此消费者要首先获取到自己被重新分配到的分区，并且定位到每个分区最近提交的 offset  位置继续消费。

要实现自定义存储 offset ，需要借助  **ConsumerRebalanceListener** ，以下为 示例代码 ，其

中提交和获取  offset  的方法，需要根据所选的  offset  存储系统自行实现。

package wxjwtl.cn;
    
    import org.apache.kafka.clients.consumer.*;
    import org.apache.kafka.common.TopicPartition;
    
    import java.time.Duration;
    import java.util.*;
    
    /**
     * @author: wtl
     * @License: (C) Copyright 2021, wtl Corporation Limited.
     * @Contact: 1050100468@qq.com
     * @Date: 2021/6/15 15:52
     * @Version: 1.0
     * @Description:
     */
    public class ConsumerTest {
    
        public static void main(String[] args) throws Exception {
            Properties properties = new Properties();
            //kafka 集群，broker-list
            properties.put("bootstrap.servers", "bigdata01:9092");
            properties.put("group.id", "first-Group");
    
            // 可选设置属性
    //        properties.put("enable.auto.commit", "true");
    //        // 自动提交offset,每1s提交一次
    //        properties.put("auto.commit.interval.ms", "1000");
    //        properties.put("auto.offset.reset","earliest");
    //        properties.put("client.id", "wxjwtl_client_id");
    
            properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            Consumer<String,String> consumer = new KafkaConsumer<>(properties);
    
            consumer.subscribe(Collections.singletonList("first"), new ConsumerRebalanceListener() {
    
                //该方法会在 Rebalance 之前调用
                @Override
                public void onPartitionsRevoked(Collection<TopicPartition> collection) {
                    for(TopicPartition topicPartition:collection){
                        int partition = topicPartition.partition();
                    }
                }
    
                //该方法会在 Rebalance 之后调用
                @Override
                public void onPartitionsAssigned(Collection<TopicPartition> collection) {
                    for(TopicPartition topicPartition : collection){
                        int partition = topicPartition.partition();
                    }
    
                    while (true) {
                        ConsumerRecords<String, String> records =
                                consumer.poll(Duration.ofMillis(1000));//消费者拉取数据
                        for (ConsumerRecord<String, String> record : records) {
                            System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                        }
                    }
                }
            });
    
    
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5aSp6b6Z6Iez5bCK_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20221005/2bf93b3c54534b958d5065101180619e.png
[Apache Kafka]: http://kafka.apache.org/downloads
[2020121915454619.png]: /images/20221005/752d9804cdeb48da99814937210b37cc.png
[20210704102322312.png]: /images/20221005/4af4137798ad4efcb0a5d5aee3629b3c.png
[20210704112150573.png]: /images/20221005/4c7b28f2566b4b6c9b22893270939290.png
[20210704144608609.png]: /images/20221005/29c7a424c637495081e2ea05ab6b95e6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70]: /images/20221005/9e4f0b14bd8a42388fd5480694d60e9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 1]: /images/20221005/550a85197c7f4003835870588b2e2dbc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 2]: /images/20221005/24eb6ffbd2fa44d79d1dfd439e2618fe.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 3]: /images/20221005/f2315ccd556f40418fcc1b01e7397e7a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 4]: /images/20221005/e80e293f143d46ae9a94afe4eaa20269.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3d0bDE5OTI_size_16_color_FFFFFF_t_70 5]: /images/20221005/87ce52f9febc468c9271ccaba29c9930.png