最强中间件！Kafka快速入门（Kafka理论+SpringBoot集成Kafka实践）

布满荆棘的人生 2024-02-25 05:33 69阅读 0赞

#### 自媒体文章上下架 ####

需求分析

![image-20230218210826439][]

媒体端下架文章同时app端也下架文章的实现可以通过feign去调用，但这种实现**耦合度太高**，这里使用MQ进行解耦

自媒体端一旦上下架文章就发送消息给MQ，文章微服务在去读取消息根据消息内容上下架文章

MQ还可以流量削峰，比如文章的点赞量，评论短时间可能会很多，使用MQ就可以对流量进行很好的控制

![image-20230217222318314][]

#### kafka概述 ####

消息中间件对比

<table> 
 <thead> 
  <tr> 
   <th>特性</th> 
   <th>ActiveMQ</th> 
   <th>RabbitMQ</th> 
   <th>RocketMQ</th> 
   <th>Kafka</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>开发语言</td> 
   <td>java</td> 
   <td>erlang</td> 
   <td>java</td> 
   <td>scala</td> 
  </tr> 
  <tr> 
   <td>单机吞吐量</td> 
   <td>万级</td> 
   <td>万级</td> 
   <td>10万级</td> 
   <td>100万级</td> 
  </tr> 
  <tr> 
   <td>时效性</td> 
   <td>ms</td> 
   <td>us</td> 
   <td>ms</td> 
   <td>ms级以内</td> 
  </tr> 
  <tr> 
   <td>可用性</td> 
   <td>高（主从）</td> 
   <td>高（主从）</td> 
   <td>非常高（分布式）</td> 
   <td>非常高（分布式）</td> 
  </tr> 
  <tr> 
   <td>功能特性</td> 
   <td>成熟的产品、较全的文档、各种协议支持好</td> 
   <td>并发能力强、性能好、延迟低</td> 
   <td>MQ功能比较完善，扩展性佳</td> 
   <td>只支持主要的MQ功能，主要应用于大数据领域</td> 
  </tr> 
 </tbody> 
</table>

消息中间件对比-选择建议

<table> 
 <thead> 
  <tr> 
   <th><strong>消息中间件</strong></th> 
   <th><strong>建议</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Kafka</td> 
   <td>追求高吞吐量，适合产生大量数据的互联网服务的数据收集业务</td> 
  </tr> 
  <tr> 
   <td>RocketMQ</td> 
   <td>可靠性要求很高的金融互联网领域,稳定性高，经历了多次阿里双11考验</td> 
  </tr> 
  <tr> 
   <td>RabbitMQ</td> 
   <td>性能较好，社区活跃度高，数据量没有那么大，优先选择功能比较完备的RabbitMQ</td> 
  </tr> 
 </tbody> 
</table>

kafka介绍

Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。kafka官网：http://kafka.apache.org/

![image-20230218210955518][]

kafka介绍-名词解释

![image-20230218211009327][]

*  producer：发布消息的对象称之为主题生产者（Kafka topic producer）
 *  topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）
 *  consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumers）
 *  broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些已发布的消息。

#### kafka安装配置 ####

> Mac上安装也是一样，我对原笔记已更改为Mac使用版本

Kafka对于zookeeper是强依赖，保存kafka相关的节点数据，所以安装Kafka之前必须先安装zookeeper

*  Docker安装zookeeper

docker run -d --name zookeeper -p 2181:2181 wurstmeister/zookeeper

*  Docker安装kafka

docker run -d --name kafka -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.1.5:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.1.5:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -t wurstmeister/kafka

#### kafka入门 ####

（1）创建kafka-demo项目，导入依赖

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

（2）生产者发送消息

package com.heima.kafka.sample;
    
    import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerConfig;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    /**
     * 生产者
     */
    public class ProducerQuickStart {
        
    
        public static void main(String[] args) {
        
            //1.kafka的配置信息
            Properties properties = new Properties();
            //kafka的连接地址
            properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"172.20.10.2:9092");
            //发送失败，失败的重试次数
            properties.put(ProducerConfig.RETRIES_CONFIG,5);
            //消息key的序列化器
            properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
            //消息value的序列化器
            properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
    
            //2.生产者对象
            KafkaProducer<String,String> producer = new KafkaProducer<String, String>(properties);
    
            //封装发送的消息
            ProducerRecord<String,String> record = new ProducerRecord<String, String>("itheima-topic","100001","hello kafka");
    
            //3.发送消息
            producer.send(record);
    
            //4.关闭消息通道，必须关闭，否则消息发送不成功
            producer.close();
        }
    
    }

（3）消费者接收消息

package com.heima.kafka.sample;
    
    import org.apache.kafka.clients.consumer.ConsumerConfig;
    import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    
    import java.time.Duration;
    import java.util.Collections;
    import java.util.Properties;
    
    /**
     * 消费者
     */
    public class ConsumerQuickStart {
        
    
        public static void main(String[] args) {
        
            //1.添加kafka的配置信息
            Properties properties = new Properties();
            //kafka的连接地址
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "172.20.10.2:9092");
            //消费者组
            properties.put(ConsumerConfig.GROUP_ID_CONFIG, "group2");
            //消息的反序列化器
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
    
            //2.消费者对象
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(properties);
    
            //3.订阅主题
            consumer.subscribe(Collections.singletonList("itheima-topic"));
    
            //当前线程一直处于监听状态
            while (true) {
        
                //4.获取消息
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
        
                    System.out.println(consumerRecord.key());
                    System.out.println(consumerRecord.value());
                }
            }
    
        }
    }

*  生产者发送消息，多个消费者只能有一个消费者接收到消息（多个消费者订阅同一个主题，只能有一个消费者收到消息）

![image-20230218211553093][]

*  生产者发送消息，多个消费者都可以接收到消息(消费者组)

![image-20230218211601163][]

#### kafka高可用设计 ####

##### 集群 #####

![image-20230218211703302][]

*  Kafka 的服务器端由被称为 Broker 的服务进程构成，即一个 Kafka 集群由多个 Broker 组成
 *  这样如果集群中某一台机器宕机，其他机器上的 Broker 也依然能够对外提供服务。这其实就是 Kafka 提供高可用的手段之一

##### 备份机制(Replication） #####

![image-20230218211730646][]

Kafka 中消息的备份又叫做 副本（Replica）

Kafka 定义了两类副本：

*  领导者副本（Leader Replica）
 *  追随者副本（Follower Replica）

**同步方式**

![image-20230218211843076][]

ISR（in-sync replica）需要同步复制保存的follower

如果leader失效后，需要选出新的leader，选举的原则如下：

第一：选举时优先从ISR中选定，因为这个列表中follower的数据是与leader同步的

第二：如果ISR列表中的follower都不行了，就只能从其他follower中选取

极端情况，就是所有副本都失效了，这时有两种方案

第一：等待ISR中的一个活过来，选为Leader，数据可靠，但活过来的时间不确定

第二：选择第一个活过来的Replication，不一定是ISR中的，选为leader，以最快速度恢复可用性，但数据不一定完整

#### kafka生产者详解 ####

##### 发送类型 #####

*  同步发送
    
    使用send()方法发送，它会返回一个Future对象，调用get()方法进行等待，就可以知道消息是否发送成功

RecordMetadata recordMetadata = producer.send(kvProducerRecord).get();
    System.out.println(recordMetadata.offset());

*  异步发送
    
    调用send()方法，并指定一个回调函数，服务器在返回响应时调用函数

//异步消息发送
    producer.send(kvProducerRecord, new Callback() {
        
        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
        
            if(e != null){
        
                System.out.println("记录异常信息到日志表中");
            }
            System.out.println(recordMetadata.offset());
        }
    });

##### 参数详解 #####

*  ack

![image-20230218212046652][]

代码的配置方式：

//ack配置  消息确认机制
    prop.put(ProducerConfig.ACKS_CONFIG,"all");

参数的选择说明

<table> 
 <thead> 
  <tr> 
   <th><strong>确认机制</strong></th> 
   <th><strong>说明</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>acks=0</td> 
   <td>生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险，但是速度最快</td> 
  </tr> 
  <tr> 
   <td>acks=1（默认值）</td> 
   <td>只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应</td> 
  </tr> 
  <tr> 
   <td>acks=all</td> 
   <td>只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应</td> 
  </tr> 
 </tbody> 
</table>

*  retries

![image-20230218212057424][]

生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可以重发消息的次数，如果达到这个次数，生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms

代码中配置方式：

//重试次数
    prop.put(ProducerConfig.RETRIES_CONFIG,10);

*  消息压缩

默认情况下， 消息发送时不会被压缩。

代码中配置方式：

//数据压缩
    prop.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"lz4");

<table> 
 <thead> 
  <tr> 
   <th><strong>压缩算法</strong></th> 
   <th><strong>说明</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>snappy</td> 
   <td>占用较少的 CPU， 却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用</td> 
  </tr> 
  <tr> 
   <td>lz4</td> 
   <td>占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观</td> 
  </tr> 
  <tr> 
   <td>gzip</td> 
   <td>占用较多的 CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法</td> 
  </tr> 
 </tbody> 
</table>

使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。

#### kafka消费者详解 ####

##### 消费者组 #####

![image-20230218212202507][]

*  消费者组（Consumer Group） ：指的就是由一个或多个消费者组成的群体
 *  一个发布在Topic上消息被分发给此消费者组中的一个消费者
    
     *  **所有的消费者都在一个组中，那么这就变成了queue模型**
     *  所有的消费者都在不同的组中，那么就完全变成了**发布-订阅模型**

##### 消息有序性 #####

应用场景：

*  即时消息中的单对单聊天和群聊，保证发送方消息发送顺序与接收方的顺序一致
 *  充值转账两个渠道在同一个时间进行余额变更，短信通知必须要有顺序

![image-20230218212257551][]

topic分区中消息只能由消费者组中的唯一一个消费者处理，所以消息肯定是按照先后顺序进行处理的。但是它也仅仅是保证Topic的一个分区顺序处理，不能保证跨分区的消息先后处理顺序。 所以，如果你想要顺序的处理Topic的所有消息，那就只提供一个分区。

##### 提交和偏移量 #####

kafka不会像其他JMS队列那样需要得到消费者的确认，消费者可以使用kafka来追踪消息在分区的位置（偏移量）

消费者会往一个叫做\_consumer\_offset的特殊主题发送消息，消息里包含了每个分区的偏移量。如果消费者发生崩溃或有新的消费者加入群组，就会触发再均衡

![image-20230218212311057][]

正常的情况

![image-20230218212323297][]

如果消费者2挂掉以后，会发生再均衡，消费者2负责的分区会被其他消费者进行消费

再均衡后不可避免会出现一些问题

问题一：

![image-20230218212334471][]

如果提交偏移量小于客户端处理的最后一个消息的偏移量，那么处于两个偏移量之间的消息就会被重复处理。

问题二：

![image-20230218212347844][]

如果提交的偏移量大于客户端的最后一个消息的偏移量，那么处于两个偏移量之间的消息将会丢失。

如果想要解决这些问题，还要知道目前kafka提交偏移量的方式：

提交偏移量的方式有两种，分别是自动提交偏移量和手动提交

*  自动提交偏移量

当enable.auto.commit被设置为true，提交方式就是让消费者自动提交偏移量，每隔5秒消费者会自动把从poll()方法接收的最大偏移量提交上去

*  手动提交 ，当enable.auto.commit被设置为false可以有以下三种提交方式
    
     *  提交当前偏移量（同步提交）
     *  异步提交
     *  同步和异步组合提交

1.提交当前偏移量（同步提交）

把`enable.auto.commit`设置为false,让应用程序决定何时提交偏移量。使用commitSync()提交偏移量，commitSync()将会提交poll返回的最新的偏移量，所以在处理完所有记录后要确保调用了commitSync()方法。否则还是会有消息丢失的风险。

只要没有发生不可恢复的错误，commitSync()方法会一直尝试直至提交成功，如果提交失败也可以记录到错误日志里。

while (true){
        
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
        
            System.out.println(record.value());
            System.out.println(record.key());
            try {
        
                consumer.commitSync();//同步提交当前最新的偏移量
            }catch (CommitFailedException e){
        
                System.out.println("记录提交失败的异常："+e);
            }
    
        }
    }

2.异步提交

手动提交有一个缺点，那就是当发起提交调用时应用会阻塞。当然我们可以减少手动提交的频率，但这个会增加消息重复的概率（和自动提交一样）。另外一个解决办法是，使用异步提交的API。

while (true){
        
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
        for (ConsumerRecord<String, String> record : records) {
        
            System.out.println(record.value());
            System.out.println(record.key());
        }
        consumer.commitAsync(new OffsetCommitCallback() {
        
            @Override
            public void onComplete(Map<TopicPartition, OffsetAndMetadata> map, Exception e) {
        
                if(e!=null){
        
                    System.out.println("记录错误的提交偏移量："+ map+",异常信息"+e);
                }
            }
        });
    }

3.同步和异步组合提交

异步提交也有个缺点，那就是如果服务器返回提交失败，异步提交不会进行重试。相比较起来，同步提交会进行重试直到成功或者最后抛出异常给应用。异步提交没有实现重试是因为，如果同时存在多个异步提交，进行重试可能会导致位移覆盖。

举个例子，假如我们发起了一个异步提交commitA，此时的提交位移为2000，随后又发起了一个异步提交commitB且位移为3000；commitA提交失败但commitB提交成功，此时commitA进行重试并成功的话，会将实际上将已经提交的位移从3000回滚到2000，导致消息重复消费。

try {
        
        while (true){
        
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
        
                System.out.println(record.value());
                System.out.println(record.key());
            }
            consumer.commitAsync();
        }
    }catch (Exception e){
        +
        e.printStackTrace();
        System.out.println("记录错误信息："+e);
    }finally {
        
        try {
        
            consumer.commitSync();
        }finally {
        
            consumer.close();
        }
    }

#### springboot集成kafka ####

##### 入门 #####

1.导入spring-kafka依赖信息

<dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.springframework.kafka</groupId>
            <artifactId>spring-kafka</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.kafka</groupId>
                    <artifactId>kafka-clients</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
        </dependency>
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
        </dependency>
    </dependencies>

2.在resources下创建文件application.yml

server:
      port: 9991
    spring:
      application:
        name: kafka-demo
      kafka:
        bootstrap-servers: 172.20.10.2:9092
        producer:
          retries: 10
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.apache.kafka.common.serialization.StringSerializer
        consumer:
          group-id: ${
        spring.application.name}-test
          key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
          value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

3.消息生产者

package com.heima.kafka.controller;
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.kafka.core.KafkaTemplate;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    @RestController
    public class HelloController {
        
    
        @Autowired
        private KafkaTemplate<String,String> kafkaTemplate;
    
        @GetMapping("/hello")
        public String hello(){
        
            kafkaTemplate.send("itcast-topic","黑马程序员");
            return "ok";
        }
    }

4.消息消费者

package com.heima.kafka.listener;
    
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    import org.springframework.util.StringUtils;
    
    @Component
    public class HelloListener {
        
    
        @KafkaListener(topics = "itcast-topic")
        public void onMessage(String message){
        
            if(!StringUtils.isEmpty(message)){
        
                System.out.println(message);
            }
    
        }
    }

##### 传递消息为对象 #####

目前springboot整合后的kafka，因为序列化器是StringSerializer，这个时候如果需要传递对象可以有两种方式

方式一：可以自定义序列化器，对象类型众多，这种方式通用性不强，本章节不介绍

方式二：可以把要传递的对象进行转json字符串，接收消息后再转为对象即可，本项目采用这种方式

*  发送消息

@GetMapping("/hello")
    public String hello(){
        
        User user = new User();
        user.setUsername("xiaowang");
        user.setAge(18);
    
        kafkaTemplate.send("user-topic", JSON.toJSONString(user));
    
        return "ok";
    }

*  接收消息

package com.heima.kafka.listener;
    
    import com.alibaba.fastjson.JSON;
    import com.heima.kafka.pojo.User;
    import org.springframework.kafka.annotation.KafkaListener;
    import org.springframework.stereotype.Component;
    import org.springframework.util.StringUtils;
    
    @Component
    public class HelloListener {
        
    
        @KafkaListener(topics = "user-topic")
        public void onMessage(String message){
        
            if(!StringUtils.isEmpty(message)){
        
                User user = JSON.parseObject(message, User.class);
                System.out.println(user);
            }
    
        }
    }

####  ####

[image-20230218210826439]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/645737b8b1274c2bb91333cda2a28469.png
[image-20230217222318314]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/df81b720e8664e9aa25360058f2fecfe.png
[image-20230218210955518]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/d4162861312548d3b6f025e9bac67aad.png
[image-20230218211009327]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/0dce7f8c64a74f4cbce4b666c76d482e.png
[image-20230218211553093]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/b30506e1833b4659ae242b61dbaab4d4.png
[image-20230218211601163]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/b946bcf5c5954fe2a7bb9e60c67612b8.png
[image-20230218211703302]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/2b726fa3ea6941ceab47f6e1e0654307.png
[image-20230218211730646]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/6eb50a4f45ca4cad9b1f1ba7d9af263c.png
[image-20230218211843076]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/7202d76298504835bd37ede8602ee0cf.png
[image-20230218212046652]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/3c2b04044fb246a9ba01d353f32d9509.png
[image-20230218212057424]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/c38c7cc9e263423a826b92f234373d2f.png
[image-20230218212202507]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/c8203f1761384090be5b7fb35766c607.png
[image-20230218212257551]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/5f2a15f8673c4c749f794d737409cb65.png
[image-20230218212311057]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/129e555c8cdb43f18e214fb9e39ffcfd.png
[image-20230218212323297]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/aabb6072d5c148f2897f90a1392ca1c4.png
[image-20230218212334471]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/faf1174671c949e39b475fc3e265a082.png
[image-20230218212347844]: https://image.dandelioncloud.cn/pgy_files/images/2024/02/25/94b7201182284402b4f830310da6d616.png