一、为什么要使用消息队列

1、生活中收快递

2、下单功能：同步

①问题1：耦合度高

②问题2：响应时间长

③问题3：并发压力传递

④问题4：系统结构弹性不足

3、下单功能：异步

①好处1：功能解耦

②好处2：快速响应

③好处3：异步削峰限流

削峰填谷：

④好处4：系统结构弹性大，易于扩展

二、什么是消息队列

1、概念

消息队列是实现应用程序和应用程序之间通信的中间件产品

2、消息队列底层实现的两大主流方式

由于消息队列执行的是跨应用的信息传递，所以制定底层通信标准非常必要目前主流的

消息队列通信协议标准包括：

• AMQP (Advanced Message Queuing Protocol)：通用协议，IBM公司研发
• JMS (Java Message Service)：专门为Java语言服务，SUN公司研发，一组由Java接口组成的Java标准

对比：

3、主流消息队列产品

三、Kafka介绍

1、Kafka是什么

Kafka是Apache开源的一款基于zookeeper协调的分布式消息系统，具有高吞吐率、高性能、实时、高可靠等特点，可实时处理流式数据。它最初由LinkedIn公司开发，使用Scala语言编写。

Kafka历经数年的发展，从最初纯粹的消息引擎，到近几年开始在流处理平台生态圈发力，多个组织或公司发布了各种不同特性的产品。常见产品如下：

• Apache Kafka ：最“正统”的Kafka也是开源版，它是后面其他所有发行版的基础。
• Cloudera/Hortonworks Kafka ：集成了目前主流的大数据框架，能够帮助用户实现从分布式存储、集群调度、流处理到机器学习、实时数据库等全方位的数据处理。
• Confluent Kafka ：主要提供基于Kafka的企业级流处理解决方案。

Apache Kafka，它现在依然是开发人数最多、版本迭代速度最快的Kafka。我们使用此产品学习。Apache 目前为止总共演进了8个大版本，分别是0.7、0.8、0.9、0.11、1.0、2.0和3.0，我们选择3.5.1版本讲解(截止2023.8)。

2、Kafka的特点

• 高吞吐量、低延迟：即使是非常普通的硬件Kafka也可以支持每秒数百万的消息，它的延迟最低只有几毫秒

• 持久性：支持消息持久化，即使数TB级别的消息也能够保持长时间的稳定性能。

• 可靠性：支持数据备份防止丢失

• 容错性：支持通过Kafka服务器和消费机集群来分区消息，允许集群中的节点失败（若分区副本数量为n，则允许n-1个节点失败）

• 高并发：单机可支持数千个客户端同时读写，支持在线水平扩展。可无缝对接hadoop、strom、spark等，支持Hadoop并行数据加载，

3、Kafka官网地址

4、Kafka应用场景

四、Kafka内部结构

一、Producer

生产者：消息发送端

二、Consumer

消费者：消息接收端

三、broker

一个Kafka服务器实例，在Kafka集群中会有多个broker实例

四、Topic

Topic中文意思是主题，在Kafka中只是一个逻辑概念，代表某一类消息。

结合具体项目中的业务功能，我们可以为每一个具体功能创建一个Topic。

五、Partition

Partition就是分区，为什么要分区？

有了分区就可以把消息数据分散到不同broker上保存。

六、Replication

数据分区之后有一个问题：每个broker上保存一部分数据，如果某个broker宕机，那么数据就会不完整。

所以Kafka允许分区创建副本

七、主从

当分区存在副本时，就会区分Leader、Follower：

• Leader：主分片，负责接收生产者端发送过来的消息，对接消费者端消费消息
• Follower：不和生产者、消费者交互，仅负责和Leader同步数据

• 创建Topic时通过“分区数”指定Partition的数量，通过“复制因子”指定副本数量
• 分区数和复制因子都不能为0
• 分区数为1，复制因子为1表示：1个Partition内有1个Leader（此时数据只有一份，没有冗余的副本，生产环境不建议）
• 复制因子为2表示每个Partition中包含1个Leader和1个Follower

八、注册

Kafka工作过程中，broker、Partition……信息都需要在Zookeeper中注册

五、图形化界面软件Eagle

一、创建Docker容器

# 搜索镜像
docker search efak

# 创建容器
docker run -d --name kafka-eagle \
-p 8048:8048 \
-e EFAK_CLUSTER_ZK_LIST="192.168.200.100:2181" \
nickzurich/efak:latest

提示：如果无法启动，往往是因为Zookeeper所需内存不足，可以试着把Zookeeper内存调整为更大的值再试。

二、使用

1、访问地址

http://192.168.200.100:8048

默认登录信息：

• 账号：admin
• 密码：123456

2、查看broker列表

3、主题相关操作

①新建

注意：Kafka集群中broker实例的数量需要大于等于复制因子（Replication factor），如果复制因子大于broker实例数量，那么就会看到下面保存信息——

②查看主题列表

③查看主题详情

点击主题名称查看详情：

4、查看分区中的消息

六、客户端原生API

一、生产者

1、创建主题

kafka-topics.sh --bootstrap-server 192.168.200.100:9092 --create --topic topic-java-client

2、启动消费者监听主题

kafka-console-consumer.sh --bootstrap-server 192.168.200.100:9092 --topic topic-java-client

3、引入依赖

<!-- kafka-clients 2023.10-->
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

4、Java程序

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 MyProducerDemo
{
    public static final String TOPIC_NAME = "topic-java-client";

    public static void main(String[] args)
    {
        // 1. 创建Kafka生产者的配置对象
        Properties properties = new Properties();
        
        // 2. 给Kafka配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.100:9092");
        
        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        
        // 3. 创建Kafka生产者对象
        KafkaProducer kafkaProducer = new KafkaProducer(properties);
        
        // 4. 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            kafkaProducer.send(new ProducerRecord<>(TOPIC_NAME, "hello-kafka-from-java-client~" + i));
        }
        
        System.out.println("----MyProducerDemo发送完毕");
        // 5. 关闭资源
        kafkaProducer.close();
    }
}

ProducerRecord参数说明：

public class ProducerRecord<K, V> {
    //主题名称，必选参数
    private final String topic;
    
    //分区号，大于等于0的整数，可选参数。
    private final Integer partition;
    
    //消息的头信息，类型是RecordHeaders，可选属性。
    private final Headers headers;
    
    //键，可选参数。
    private final K key;
    
    //消息内容，必选参数。
    private final V value;
    
    //每条消息都有一个时间戳，可选参数
    private final Long timestamp;
}

5、send()方法返回值

KafkaProducer的send()方法返回Future类型的对象，可以调用Future的get()方法同步获取任务执行结果。

此时程序就成了前一个消息发送完成再发送后一个的同步模式。

也就是说不调用get()方法就是异步模式。

// 同步
for (int i = 0; i < 5; i++) {
    // 发送消息的任务交给子线程去做
    Future future = kafkaProducer.send(new ProducerRecord<>(TOPIC_NAME, "hello-kafka-from-java-client~~~" + i));

    TimeUnit.SECONDS.sleep(1);

    // 但是因为调用了 get() 方法，就变成子线程必须执行完发送消息的任务
    // for 循环的本次循环体才算执行完，才能继续执行下一次循环
    // 下一次循环就是发送下一条消息
    future.get();
}

6、获取消息发送结果

给KafkaProducer的send()方法再传入一个CallBack类型的参数，以异步回调的方式获取消息发送结果，从而得知消息发送是成功还是失败。

①Java代码

kafkaProducer.send(new ProducerRecord<>(TOPIC_NAME, "hello-kafka-from-java-client*******"), new Callback() {

    // onCompletion() 方法在发送消息操作完成时被调用
    // 参数 RecordMetadata recordMetadata：发送消息相关的元数据
    // 参数 Exception e：发送消息失败时，失败原因封装的异常信息
    @Override
    public void onCompletion(RecordMetadata recordMetadata, Exception e) {

        if (e == null) {
            long offset = recordMetadata.offset();
            System.out.println("offset = " + offset);

            int partition = recordMetadata.partition();
            System.out.println("partition = " + partition);

            long timestamp = recordMetadata.timestamp();
            System.out.println("timestamp = " + timestamp);

            String topic = recordMetadata.topic();
            System.out.println("topic = " + topic);

        } else {
            System.out.println("e = " + e);
        }
    }
});

②失败情况举例

把broker地址改成错的：

e = org.apache.kafka.common.errors.TimeoutException: Topic topic-java-client not present in metadata after 60000 ms.

二、消费者

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class MyConsumerDemo
{
    public static final String TOPIC_NAME = "topic-java-client";
    public static void main(String[] args) throws InterruptedException
    {
        // 1、创建Kafka消费者的配置对象
        Properties properties = new Properties();
        
		// 2、给Kafka配置对象添加配置信息：bootstrap.servers
        properties.put("bootstrap.servers", "192.168.200.100:9092");
        properties.setProperty("group.id", "test");
        properties.setProperty("enable.auto.commit", "true");
        properties.setProperty("auto.commit.interval.ms", "1000");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        // 3、创建消费者对象
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties);
        
        // 4、订阅指定主题
        consumer.subscribe(Arrays.asList(TOPIC_NAME));

        while (true) {
            // 5、从broker拉取信息
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            
            // 6、每隔 1 秒做一次打印，让消费端程序持续运行
            TimeUnit.SECONDS.sleep(1);
            System.out.println("....进行中");
        }
    }
}

七、Kafka集群

一、集群搭建

1、重要原则

• Kafka节点只要注册到同一个Zookeeper上就代表它们是同一个集群的
• Kafka通过broker.id来区分集群中的不同节点

2、规划

• 简单起见，我们只使用一个VMWare虚拟机，所以各个broker实例需要设定不同端口号
• Kafka程序不需要复制，对应各自不同的配置文件启动多个进程就能组成集群
• Zookeeper还是使用原来的2181即可

3、具体操作

①创建目录

mkdir -p /opt/k-cluster/log7000
mkdir -p /opt/k-cluster/log8000
mkdir -p /opt/k-cluster/log9000

②复制配置文件

cp /opt/kafka_2.13-3.6.0/config/server.properties /opt/k-cluster/server7000.properties
cp /opt/kafka_2.13-3.6.0/config/server.properties /opt/k-cluster/server8000.properties
cp /opt/kafka_2.13-3.6.0/config/server.properties /opt/k-cluster/server9000.properties

③修改配置文件

[1]7000

broker.id=1
listeners=PLAINTEXT://192.168.200.100:7000
advertised.listeners=PLAINTEXT://192.168.200.100:7000
log.dirs=/opt/k-cluster/log7000

[2]8000

broker.id=2
listeners=PLAINTEXT://192.168.200.100:8000
advertised.listeners=PLAINTEXT://192.168.200.100:8000
log.dirs=/opt/k-cluster/log8000

[3]9000

broker.id=3
listeners=PLAINTEXT://192.168.200.100:9000
advertised.listeners=PLAINTEXT://192.168.200.100:9000
log.dirs=/opt/k-cluster/log9000

4、启动集群各实例

注意：此前需要先启动Zookeeper

kafka-server-start.sh -daemon /opt/k-cluster/server7000.properties
kafka-server-start.sh -daemon /opt/k-cluster/server8000.properties
kafka-server-start.sh -daemon /opt/k-cluster/server9000.properties

验证各个端口号：

lsof -i:2181
lsof -i:7000
lsof -i:8000
lsof -i:9000

如果因为内存不足而启动失败，可以修改对应启动脚本程序中的内存大小：

• Zookeeper启动脚本程序：zookeeper-server-start.sh
• Zookeeper中Kafka堆内存大小变量名称：KAFKA_HEAP_OPTS
• Kafka启动脚本程序：kafka-server-start.sh
• Kafka堆内存大小变量名称：KAFKA_HEAP_OPTS

5、停止集群

# 停止Kafka，无需指定端口号就能停止各个实例：
kafka-server-stop.sh
# 停止zk
zookeeper-server-stop.sh

二、使用集群

1、在集群上创建主题

kafka-topics.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000 \
--create \
--partitions 3 \
--replication-factor 3 \
--topic my-cluster-topic

2、查看集群主题

kafka-topics.sh \
--bootstrap-server 192.168.200.100:7000 \
--describe --topic my-cluster-topic

3、集群消息发送

kafka-console-producer.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000 \
--topic my-cluster-topic

4、集群消息消费

kafka-console-consumer.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000  \
--from-beginning \
--topic my-cluster-topic

5、集群消息消费相关问题

①问题描述

通过集群接收消息时，接收不到

②问题产生原因

多个broker实例部署在同一个虚拟机上

• 192.168.200.100:7000
• 192.168.200.100:8000
• 192.168.200.100:9000

这只是我们在测试环境下，非正式的这么安排，实际开发中不会把集群的所有实例放在一个机器上

③问题解决方案一

消费端接收消息时指定分区

kafka-console-consumer.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000  \
--from-beginning \
--partition 0 \
--topic my-cluster-topic

kafka-console-consumer.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000  \
--from-beginning \
--partition 1 \
--topic my-cluster-topic

kafka-console-consumer.sh \
--bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000  \
--from-beginning \
--partition 2 \
--topic my-cluster-topic

④问题解决方案二

• 第一步：把apache-zookeeper-3.9.1-bin.tar.gz上传到Linux系统/opt目录下
• 第二步：解压apache-zookeeper-3.9.1-bin.tar.gz文件

cd /opt
tar -zxvf apache-zookeeper-3.9.1-bin.tar.gz

• 第三步：运行zkCli.sh脚本文件，登录到Zookeeper服务器

/opt/apache-zookeeper-3.9.1-bin/bin/zkCli.sh

• 第四步：删除__consumer_offsets主题

deleteall /brokers/topics/__consumer_offsets

• 第五步：退出Zookeeper

quit

• 第六步：重启
  • 先关闭然后重新启动Zookeeper
  • 先关闭然后重新启动集群各实例

八 客户端SpringBoot

一、生产者

1、配置POM

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.1.3</version>
    <relativePath/>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <!--spring-kafka-->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!--hutool-->
    <dependency>
        <groupId>cn.hutool</groupId>
        <artifactId>hutool-all</artifactId>
        <version>5.8.19</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
                <excludes>
                    <exclude>
                        <groupId>org.projectlombok</groupId>
                        <artifactId>lombok</artifactId>
                    </exclude>
                </excludes>
            </configuration>
        </plugin>
    </plugins>
</build>

2、配置YAML

spring:
  kafka:
    bootstrap-servers: 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3、主启动类

package com.atguigu.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaMainType {

    public static void main(String[] args) {
        SpringApplication.run(KafkaMainType.class, args);
    }
    
}

4、配置类创建主题

package com.atguigu.kafka.config;

import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;

@Configuration
public class KafkaConfig {
    @Bean
    public NewTopic springTestTopic() {
        return TopicBuilder.name("topic-spring-boot") // 主题名称
                .partitions(3) // 分区数量
                .replicas(3) // 副本数量
                .build();
    }
}

到这里我们可以运行主启动类，看看主题是否创建成功

kafka-topics.sh --bootstrap-server 192.168.200.100:7000 --list

5、发送消息

①命令行监听消息

kafka-console-consumer.sh --bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000 --topic topic-spring-boot --partition 0

kafka-console-consumer.sh --bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000 --topic topic-spring-boot --partition 1

kafka-console-consumer.sh --bootstrap-server 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000 --topic topic-spring-boot --partition 2

②Java代码

package com.atguigu.kafka.test;

import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;

@SpringBootTest
public class KafkaTest {

    @Resource
    private KafkaTemplate kafkaTemplate;

    @Test
    public void testSendMessage() {

        String topicName = "topic-spring-boot";
        String message = "hello spring boot message";

        kafkaTemplate.send(topicName, message);
    }

}

二、消费者

1、配置POM

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>3.1.3</version>
    <relativePath/>
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <!--spring-kafka-->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!--hutool-->
    <dependency>
        <groupId>cn.hutool</groupId>
        <artifactId>hutool-all</artifactId>
        <version>5.8.19</version>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <configuration>
                <excludes>
                    <exclude>
                        <groupId>org.projectlombok</groupId>
                        <artifactId>lombok</artifactId>
                    </exclude>
                </excludes>
            </configuration>
        </plugin>
    </plugins>
</build>

2、配置YAML

spring:
  Kafka:
    bootstrap-servers: 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-id: consumer-group

3、主启动类

package com.atguigu.kafka;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaMainTypeConsumer {

    public static void main(String[] args) {
        SpringApplication.run(KafkaMainTypeConsumer.class, args);
    }

}

4、接收消息的监听器

package com.atguigu.kafka.listener;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class KafkaMessageListener {

    @KafkaListener(topics = {"topic-spring-boot"})
    public void simpleConsumerPartition(ConsumerRecord<String, String> record) {
        System.out.println("进入simpleConsumer方法");
        System.out.printf(
                "分区 = %d, 偏移量 = %d, key = %s, 内容 = %s, 时间戳 = %d%n",
                record.partition(),
                record.offset(),
                record.key(),
                record.value(),
                record.timestamp()
        );
    }

}

注意：这里我们没有指定具体接收哪个分区的消息，所以如果接收不到消息，那么就需要登录Zookeeper删除__consumer_offsets

deleteall /brokers/topics/__consumer_offsets

三、实体类对象类型的消息

1、创建实体类

package com.atguigu.kafka.entity;

import lombok.AllArgsConstructor;
import lombok.Data;

@Data
@AllArgsConstructor
public class UserDTO {
    private String name;
    private Integer age;
    private String mobile;
}

2、发送消息的方法

@Test
public void testSendEntity() {
    String topicName = "topic-spring-boot230628";
    UserDTO userDTO = new UserDTO("tom", 25, "12345343");

    kafkaTemplate.send(topicName, userDTO);
}

3、异常

• 异常全类名：java.lang.ClassCastException
• 异常信息：class com.atguigu.kafka.entity.UserDTO cannot be cast to class java.lang.String (com.atguigu.kafka.entity.UserDTO is in unnamed module of loader 'app'; java.lang.String is in module java.base of loader 'bootstrap')
• 异常原因：目前使用的序列化器是StringSerializer，不支持非字符串序列化
• 解决办法：把序列化器换成支持复杂类型的

4、修改YAML配置

spring:
  kafka:
    bootstrap-servers: 192.168.200.100:7000,192.168.200.100:8000,192.168.200.100:9000
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      # value-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

转载链接：https://juejin.cn/post/7332865125638602764