大数据必学框架-kafka

- 日理万妓 2023-10-04 16:42 54阅读 0赞

## 大数据必学框架-kafka ##

### Kafka概述 ###

#### 定义： ####

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

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

1）解耦

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

2）可恢复性

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

3）缓冲

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

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

5）异步通信

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

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

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

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

#### Kafka 基础架构 ####

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

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是依赖于ZooKeeper的，所以在安装Kafka之前要确保ZooKeeper的安装

#### 集群规划 ####

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>hadoop102</th> 
   <th>hadoop103</th> 
   <th>hadoop104</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zookeeper</td> 
   <td>zk</td> 
   <td>zk</td> 
   <td>zk</td> 
  </tr> 
  <tr> 
   <td>kafka</td> 
   <td>kafka</td> 
   <td>kafka</td> 
   <td>kafka</td> 
  </tr> 
 </tbody> 
</table>

#### Kafka安装 ####

官网地址：http://kafka.Apache.org

下载地址：http://kafka.Apache.org/downloads.html

下载kafka的tar包，然后解压到相应的文件目录，可以在kafka的目录下新建一个文件夹data，用于等会存放kafka的数据使用

**修改配置文件：**

在conf文件夹下，有配置文件server.properties

配置内容详情：

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

**环境变量可以选择配置，配置了就可以在任何位置启动kafka，如果不配置就需要到kafka目录下启动**

配置完成配置文件之后，就把配置文件分发到集群，注意，**分发之后需要修改broker的id号**

#### 启动Kafka ####

依次在 hadoop102、hadoop103、hadoop104 节点上启动 kafka

**启动命令：**

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

\-\*\*daemon：\*\*守护进程，如果不加启动之后就是阻塞状态，占用窗口，使用daemon可以变成守护进程

**关闭集命令：**

bin/kafka-server-stop.sh stop

**Kafka的群起脚本**

for i in hadoop102 hadoop103 hadoop104 
    do 
    echo "========== $i =========="  
    ssh  $i  '/opt/module/kafka/bin/kafka-server-start.sh  -daemon 			  /opt/module/kafka/config/server.properties' 
    done

#### Kafka的命令行操作 ####

1）**查看当前服务器中的所有 topic**

bin/kafka-topics.sh  --zookeeper hadoop102:2181 --list

2）**创建 topic**

bin/kafka-topics.sh  --zookeeper hadoop102:2181 --create --replication-factor 3 --partitions 1 --topic first

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

3）**删除 topic**

bin/kafka-topics.sh  --zookeeper hadoop102:2181 --delete --topic first

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

4）**发送消息** ：即是启动生产者

bin/kafka-console-producer.sh --broker-list hadoop102:9092 --topic first

5）**消费消息** ：即是启动消费者

bin/kafka-console-consumer.sh --zookeeper hadoop102:2181 --topic first

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

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

6）**查看某个 Topic 的详情**

bin/kafka-topics.sh  --zookeeper hadoop102:2181 --describe --topic first

7）**修改分区数**

bin/kafka-topics.sh  --zookeeper hadoop102:2181 --alter --topic first --partitions 6

### Kafka 架构深入 ###

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

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

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

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

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

index 和 log 文件以当前 segment 的第一条消息的 offset 命名。下图为 index 文件和 log文件的结构示意图。  
![在这里插入图片描述][ec1977293d3e491097ebd4e42e91e7b3.png]

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

#### Kafka的生产者 ####

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

**分区的原因：**

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

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

**分区的原则：**

（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，就会进行下一轮的发送，否则重新发送数据。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 4]

##### ISR ：in-sync replica set #####

对于第二种策略，如果有一个Follower一直没有与Leader同步，那么就要一直等待，所以针对于这种状况，Kafka提供了ISR。

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 发生故障，那么会造成 数据重复。

##### 故障 处理：LEO和HW #####

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

**LEO** ：指的是每个副本最大的 offset ；

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

**1 ）follower 故障**

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

**（2 ）leader 故障**

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

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

#### 消费者Consumer ####

##### 消费 方式 #####

**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。**

**RoundRobin：循环，轮流消费**

**Range：范围，一个消费者负责一个范围(主题)—>默认**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 6]

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

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

Kafka 0.9 版本之前，consumer 默认将 offset 保存在 Zookeeper 中，从 0.9 版本开始，consumer 默认将 offset 保存在 Kafka 一个内置的 topic 中，该 topic 为\_\_consumer\_offsets。

**修改配置文件 consumer.properties**

exclude.internal.topics=false

0.11.0.0 之后版本(含):

bin/kafka-console-consumer.sh  --topic  __consumer_offsets  --zookeeper  hadoop102:2181  --formatter "kafka.coordinator.group.GroupMetadataManager $OffsetsMessageFormatter"  --consumer.config  config/consumer.properties  --from-
    beginning

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

**1 ）顺序写磁盘**

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

**2 ）零复制技术**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 7]

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

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

Controller 的管理工作都是依赖于 Zookeeper 的。

### kafka API ###

#### Producer API ####

##### 消息发送流程 #####

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

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

##### 异步发送API #####

**导入依赖**

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

**编写代码**

需要用到的类：  
\*\*KafkaProducer：\*\*需要创建一个生产者对象，用来发送数据  
\*\*ProducerConfig：\*\*获取所需的一系列配置参数  
\*\*ProducerRecord：\*\*每条数据都要封装成一个 ProducerRecord 对象

**不带回调函数**

import java.util.concurrent.ExecutionException; 
     
    public class CustomProducer {
         
     
        public static void main(String[] args) throws ExecutionException, 
    InterruptedException {
         
            Properties props = new Properties(); 
     
            //kafka 集群，broker-list 
            props.put("bootstrap.servers", "hadoop102:9092"); 
     
            props.put("acks", "all"); 
     
            //重试次数 
            props.put("retries", 1);  
     
            //批次大小 
            props.put("batch.size", 16384);  
     
            //等待时间 
            props.put("linger.ms", 1);  
     
            //RecordAccumulator 缓冲区大小 
            props.put("buffer.memory", 33554432); 
     
            props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
            props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
     
            Producer<String,  String>  producer  =  new 
    KafkaProducer<>(props); 
     
            for (int i = 0; i < 100; i++) {
         
                producer.send(new ProducerRecord<String, String>("first", 
    Integer.toString(i), Integer.toString(i))); 
            } 
     
            producer.close(); 
        } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 8]

**带回调函数**

package com.gis.kafka; 
     
    import org.apache.kafka.clients.producer.*; 
     
    import java.util.Properties; 
    import java.util.concurrent.ExecutionException; 
     
    public class CustomProducer {
         
     
    public static void main(String[] args) throws ExecutionException, 
    InterruptedException {
        
    
            Properties props = new Properties(); 
     
            props.put("bootstrap.servers", "hadoop102:9092");//kafka 集
    群，broker-list 
     
            props.put("acks", "all"); 
     
            props.put("retries", 1);//重试次数 
     
            props.put("batch.size", 16384);//批次大小 
     
            props.put("linger.ms", 1);//等待时间 
     
            props.put("buffer.memory", 33554432);//RecordAccumulator 缓
    冲区大小 
     
            props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
            props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
     
            Producer<String,  String>  producer  =  new 
    KafkaProducer<>(props); 
     
            for (int i = 0; i < 100; i++) {
         
                producer.send(new ProducerRecord<String, String>("first", 
    Integer.toString(i), Integer.toString(i)), new Callback() {
         
     
                    //回调函数，该方法会在 Producer 收到 ack 时调用，为异步调用 
                    @Override 
                    public void onCompletion(RecordMetadata metadata, 
    Exception exception) {
         
                        if (exception == null) {
         
                            System.out.println("success->"  + 
    metadata.offset()); 
                        } else {
         
                            exception.printStackTrace(); 
                        } 
                    } 
                }); 
            } 
            producer.close(); 
        } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 9]

**带回调函数和不带回调函数的区别点：**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 10]

##### 同步发送API #####

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

package com.gis.kafka; 
     
    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.concurrent.ExecutionException; 
     
    public class CustomProducer {
         
     
    public static void main(String[] args) throws ExecutionException, 
    InterruptedException {
         
     
            Properties props = new Properties(); 
     
            props.put("bootstrap.servers", "hadoop102:9092");//kafka 集
    群，broker-list 
     
            props.put("acks", "all"); 
     
            props.put("retries", 1);//重试次数 
     
            props.put("batch.size", 16384);//批次大小 
     
            props.put("linger.ms", 1);//等待时间 
     
            props.put("buffer.memory", 33554432);//RecordAccumulator 缓
    冲区大小 
     
            props.put("key.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
            props.put("value.serializer", 
    "org.apache.kafka.common.serialization.StringSerializer"); 
     
            Producer<String,  String>  producer  =  new 
    KafkaProducer<>(props); 
            for (int i = 0; i < 100; i++) {
         
                producer.send(new ProducerRecord<String, String>("first", 
    Integer.toString(i), Integer.toString(i))).get(); 
            } 
            producer.close(); 
        } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 11]

#### Consumer API ####

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

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

##### 自动提交offset #####

**导入依赖**

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

**编写代码**

需要用到的类

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

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

\*\*ConsuemrRecord：\*\*每条数据都要封装成一个 ConsumerRecord 对象 ，为了使我们能够专注于自己的业务逻辑，Kafka 提供了自动提交 offset 的功能。

自动提交 offset 的相关参数：

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

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

package com.gis.kafka; 
     
    import org.apache.kafka.clients.consumer.ConsumerRecord; 
    import org.apache.kafka.clients.consumer.ConsumerRecords; 
    import org.apache.kafka.clients.consumer.KafkaConsumer; 
     
    import java.util.Arrays; 
    import java.util.Properties; 
     
    public class CustomConsumer {
         
     
    public static void main(String[] args) {
         
     
            Properties props = new Properties(); 
     
            props.put("bootstrap.servers", "hadoop102:9092"); 
     
            props.put("group.id", "test"); 
     
            props.put("enable.auto.commit", "true"); 
     
            props.put("auto.commit.interval.ms", "1000"); 
     
            props.put("key.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
            props.put("value.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
     
            KafkaConsumer<String,  String>  consumer  =  new 
    KafkaConsumer<>(props); 
     
            consumer.subscribe(Arrays.asList("first")); 
     
            while (true) {
         
     
                ConsumerRecords<String,  String>  records  = 
    consumer.poll(100); 
     
                for (ConsumerRecord<String, String> record : records) 
     
                    System.out.printf("offset = %d, key = %s, value 
    = %s%n", record.offset(), record.key(), record.value()); 
            } 
        } 
    }

**代码分析和过程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 12]

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

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

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

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

不同点是，commitSync 阻塞当前线程，一直到提交成功，并且会自动失败重试（由不可控因素导致，也会出现提交失败）；而commitAsync 则没有失败重试机制，故有可能提交失败。

**同步提交offset**

package com.atguigu.kafka.consumer; 
     
    import org.apache.kafka.clients.consumer.ConsumerRecord; 
    import org.apache.kafka.clients.consumer.ConsumerRecords; 
    import org.apache.kafka.clients.consumer.KafkaConsumer; 
     
    import java.util.Arrays; 
    import java.util.Properties; 
     
    public class CustomComsumer {
         
     
        public static void main(String[] args) {
         
     
            Properties props = new Properties(); 
     
    //Kafka 集群 
            props.put("bootstrap.servers", "hadoop102:9092");  
     
    //消费者组，只要 group.id 相同，就属于同一个消费者组 
            props.put("group.id", "test");  
     
            props.put("enable.auto.commit", "false");//关闭自动提交 offset 
    		
            props.put("key.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
            props.put("value.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
     
            KafkaConsumer<String,  String>  consumer  =  new 
    KafkaConsumer<>(props); 
     
            consumer.subscribe(Arrays.asList("first"));//消费者订阅主题 
     
            while (true) {
         
     
    //消费者拉取数据 
                ConsumerRecords<String,  String>  records  = 
    consumer.poll(100);  
     
                for (ConsumerRecord<String, String> record : records) {
         
     
                    System.out.printf("offset = %d, key = %s, value 
    = %s%n", record.offset(), record.key(), record.value()); 
     
                } 
     
    //同步提交，当前线程会阻塞直到 offset 提交成功 
                consumer.commitSync(); 
            } 
        } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 13]

**异步提交offset**

package com.atguigu.kafka.consumer; 
     
    import org.apache.kafka.clients.consumer.*; 
    import org.apache.kafka.common.TopicPartition; 
     
    import java.util.Arrays; 
    import java.util.Map; 
    import java.util.Properties; 
     
     
    public class CustomConsumer {
         
     
        public static void main(String[] args) {
         
     
            Properties props = new Properties(); 
     
            //Kafka 集群 
            props.put("bootstrap.servers", "hadoop102:9092");  
     
            //消费者组，只要 group.id 相同，就属于同一个消费者组 
            props.put("group.id", "test");  
     
            //关闭自动提交 offset 
            props.put("enable.auto.commit", "false"); 
     
            props.put("key.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
            props.put("value.deserializer", 
    "org.apache.kafka.common.serialization.StringDeserializer"); 
     
            KafkaConsumer<String,  String>  consumer  =  new 
    KafkaConsumer<>(props); 
            consumer.subscribe(Arrays.asList("first"));//消费者订阅主题 
     
            while (true) {
         
                ConsumerRecords<String,  String>  records  = 
    consumer.poll(100);//消费者拉取数据 
                for (ConsumerRecord<String, String> record : records) {
         
                    System.out.printf("offset = %d, key = %s, value 
    = %s%n", record.offset(), record.key(), record.value()); 
                } 
     
    //异步提交 
                consumer.commitAsync(new OffsetCommitCallback() {
         
                    @Override 
                    public  void  onComplete(Map<TopicPartition, 
    OffsetAndMetadata> offsets, Exception exception) {
         
                        if (exception != null) {
         
                            System.err.println("Commit  failed  for"  + 
    offsets); 
                        } 
                    } 
                });  
            } 
        } 
    }

**代码细节和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 14]

#### 自定义拦截器Interceptor ####

package com.atguigu.kafka.interceptor; 
    import java.util.Map; 
    import org.apache.kafka.clients.producer.ProducerInterceptor; 
    import org.apache.kafka.clients.producer.ProducerRecord; 
    import org.apache.kafka.clients.producer.RecordMetadata; 
     
    public  class  TimeInterceptor  implements ProducerInterceptor<String, String> {
         
     
      @Override 
      public void configure(Map<String, ?> configs) {
         
     
      } 
     
      @Override 
      public  ProducerRecord<String,  String> onSend(ProducerRecord<String, String> record) {
         
     
       // 创建一个新的 record，把时间戳写入消息体的最前部 
       return  new  ProducerRecord(record.topic(), record.partition(), record.timestamp(), record.key(), 
         System.currentTimeMillis()  +  ","  + record.value().toString()); 
      } 
     
      @Override 
      public  void  onAcknowledgement(RecordMetadata  metadata, Exception exception) {
         
     
      } 
     
      @Override 
      public void close() {
         
     
      } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 15]

package com.atguigu.kafka.interceptor; 
    import java.util.Map; 
    import org.apache.kafka.clients.producer.ProducerInterceptor; 
    import org.apache.kafka.clients.producer.ProducerRecord; 
    import org.apache.kafka.clients.producer.RecordMetadata; 
     
    public  class  CounterInterceptor  implements ProducerInterceptor<String, String>{
         
        private int errorCounter = 0; 
        private int successCounter = 0; 
     
      @Override 
      public void configure(Map<String, ?> configs) {
         
       
      } 
     
      @Override 
      public  ProducerRecord<String,  String> onSend(ProducerRecord<String, String> record) {
         
       return record; 
      } 
     
      @Override 
      public  void  onAcknowledgement(RecordMetadata  metadata, Exception exception) {
         
       // 统计成功和失败的次数 
            if (exception == null) {
         
                successCounter++; 
            } else {
         
                errorCounter++; 
            } 
      } 
     
      @Override 
      public void close() {
         
            // 保存结果 
            System.out.println("Successful sent: " + successCounter); 
            System.out.println("Failed sent: " + errorCounter); 
      } 
    }

**代码分析和流程**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 16]

### Kafka对接Flume ###

**只需要更改sink类型**

# define 
    a1.sources = r1 
    a1.sinks = k1
    a1.channels = c1 
     
    # source 
    a1.sources.r1.type = exec 
    a1.sources.r1.command = tail -F -c +0 /opt/module/data/flume.log 
    a1.sources.r1.shell = /bin/bash -c 
     
    # sink 
    a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink 
    a1.sinks.k1.kafka.bootstrap.servers  = hadoop102:9092,hadoop103:9092,hadoop104:9092 
    a1.sinks.k1.kafka.topic = first 
    a1.sinks.k1.kafka.flumeBatchSize = 20 
    a1.sinks.k1.kafka.producer.acks = 1 
    a1.sinks.k1.kafka.producer.linger.ms = 1 
     
    # channel 
    a1.channels.c1.type = memory 
    a1.channels.c1.capacity = 1000 
    a1.channels.c1.transactionCapacity = 100 
     
    # bind 
    a1.sources.r1.channels = c1 
    a1.sinks.k1.channel = c1 
    # define 
    a1.sources = r1 
    a1.sinks = k1

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

**也可以自定义拦截器，只需要像Flume一样增加一个拦截器即可**

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/fa7436f072ee41e9b967ef15759b26d4.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/6fa3e09f62034a808d3ebfc02e6fe04c.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/70febd5c711546ed8fc05b0e101c428c.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/2a4c75199cad408482f558f614f121dd.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[ec1977293d3e491097ebd4e42e91e7b3.png]: https://img-blog.csdnimg.cn/ec1977293d3e491097ebd4e42e91e7b3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 4]: https://img-blog.csdnimg.cn/9f7f5fdcf2b4442eb6c0154d4443431b.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 5]: https://img-blog.csdnimg.cn/f896628d53fe487f90b67ef45c371ca2.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 6]: https://img-blog.csdnimg.cn/c9ba5506675c4fdab2ee9317572a30a7.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 7]: https://img-blog.csdnimg.cn/622aefe958864a5fb0e48328ddb87646.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 8]: https://img-blog.csdnimg.cn/d3a518f606f94d9bbde40c8cb4a04f1a.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 9]: https://img-blog.csdnimg.cn/d917af11bc214b809d2255f82a10e025.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 10]: https://img-blog.csdnimg.cn/ddc13ecc915f48beb9304647646ee8f5.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 11]: https://img-blog.csdnimg.cn/5e95aacc57e944b8bc053eb31e95c552.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 12]: https://img-blog.csdnimg.cn/c3a23e4eee6940a4acc317c6f5267203.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 13]: https://img-blog.csdnimg.cn/80d40a6fdee04dca9556b7fcc60d7c4d.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 14]: https://img-blog.csdnimg.cn/ca9955cbf1bb43fcaafb90479077c187.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 15]: https://img-blog.csdnimg.cn/8a43193a6e8b4cf391f1ba1c411c83f9.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 16]: https://img-blog.csdnimg.cn/432350fd7e994320b9a1e73d29c6023c.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA_size_16_color_FFFFFF_t_70 17]: https://img-blog.csdnimg.cn/a9df7a6c33e740f3bb3b701923ed7375.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NjQyOTI5MA==,size_16,color_FFFFFF,t_70