一、 Kafka
1.kafka介绍
Kafka 是一个分布式流媒体平台,类似于消息队列或企业消息传递系统。
生产者发送消息，多个消费者只能有一个消费者接收到消息
生产者发送消息，多个消费者都可以接收到消息
- producer：发布消息的对象称之为主题生产者（Kafka topic producer）
- topic：Kafka将消息分门别类，每一类的消息称之为一个主题（Topic）
- consumer：订阅消息并处理发布的消息的对象称之为主题消费者（consumers）
- broker：已发布的消息保存在一组服务器中，称之为Kafka集群。集群中的每一个服务器都是一个 代理（Broker）。 消费者可以订阅一个或多个主题（topic），并从Broker拉数据，从而消费这些 已发布的消息。
2.kafka 下载(windows下)
ksfka下载链接如下，点击链接进入官网即可下载
kafka官网：http://kafka.apach e.org/
温馨提示：JDK版本至少需要1.8，高版本也可兼容；
第一步：将kafka文件夹解压缩，
注意：一定一定要解压到自己磁盘的根目录下，不然会出现闪退情况！！！
第二步：创建一个 data 空文件夹
后续需要用来存放日志文件，只要创建完成就可以了，kafka启动后会自动生成日志文件；
第三步： 修改 zookeeper.properties 配置文件
我们点击进入config文件夹，找到 zookeeper.properties 配置文件，双击进行修改
然后，我们找到 dataDir ，将它的值修改为我们刚才创建的 data 文件的路径，还要注意一点，在后面还要多加一个 "/zk"，因为一会还要配置 server.properties ，所以要用将她们两个区分开，
第四步：修改 server.properties 配置文件
和刚才一样，我们双击修改 "server.properties" 配置文件
我们修改 log.dirs 的值为刚才创建的 data 文件夹的路径，在路径末尾再添加上 "/kafka" ，用来和刚才的zk做区分，kafka 文件夹用来存放kafka的日志文件，zk 文件夹用来存放zoopeeper的日志文件；
第五步：创建 "zk.cmd" windows脚本文件
以记事本的方式打开，然后加入下面这句话，
这句话的含义就是启动 Zookeeper ，并且启动文件为 "zookeeper.properties" ；
call bin/windows/zookeeper-server-start.bat config/zookeeper.properties第六步：创建 "kfk.cmd" windows脚本文件
仍然以记事本的方式打开，然后加入下面这句话，
这句话的含义就是启动 kafka ，并且启动文件为 "server.properties" ；
call bin/windows/kafka-server-start.bat config/server.properties注意：先启动双击 zk.cmd 启动 zookeeper双击 kafka.cmd 启动 kafka，关闭的时候，需要先关闭 kafka，再关闭 zookeeper ；
3.kafka使用：
（1）创建kafka-demo项目，导入依赖
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>（2）生产者发送消息
/**
* @author 生产者
* @version 1.0
* @since 2024/9/2
*/
public class ProducerQuickStart {
public static void main(String[] args) {
// 1.kafka链接配置信息
Properties prop = new Properties();
// kafka链接地址
// 指定 Kafka 集群的地址。在这个例子中，Kafka 服务运行在本地主机（localhost）的 9092 端口上。
prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"localhost:9092");
// key和value的序列化
// 配置消息的 key 和 value 序列化类。
// Kafka 需要将 key 和 value 转换为字节数组进行传输，这里使用的是 StringSerializer，表示 key 和 value 都是字符串类型。
prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
// 2.创建kafka生产者对象
// 生产者对象依赖于前面配置的 Properties 对象来连接 Kafka 集群并发送消息。
KafkaProducer<String,String> producer = new KafkaProducer<String, String>(prop);
//3.发送消息
/**
* 第一个参数 ：topic
* 第二个参数：消息的key
* 第三个参数：消息的value
*/
// topic-first：消息要发送到的 Kafka 主题（topic）。
// key-001：消息的 key，用于分区策略或消息的唯一标识。
// hello kafka：消息的内容（value）。
ProducerRecord<String,String> kvProducerRecord = new ProducerRecord<String,String>("topic-first","key-001","hello kafka");
// 同步发送消息
// 使用 Kafka 生产者的 send 方法将消息发送到 Kafka 集群中。
// 这是一个异步操作，意味着消息会在后台发送，不会阻塞主线程。
producer.send(kvProducerRecord);
//4.关闭消息通道 必须要关闭，否则消息发送不成功
producer.close();
}
}（3）消费者接收消息
/**
* @author 甜甜
* @version 1.0
* @since 2024/9/2
*/
public class ConsumerQuickStart1 {
public static void main(String[] args) {
// 1.kafka的配置信息
Properties prop = new Properties();
// 链接地址
prop.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//key和value的反序列化器
prop.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
prop.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
// 设置消费者组
// 指定消费者所属的消费者组（group1）。
//
prop.put(ConsumerConfig.GROUP_ID_CONFIG,"group1");
// 2.创建消费者对象
// Kafka 使用消费者组来协调同一组中多个消费者的消息消费情况，确保每条消息只会被组内的一个消费者处理。
KafkaConsumer<String,String> consumer = new KafkaConsumer<String, String>(prop);
//3.订阅主题
// 消费者订阅 topic-first 主题。
// Collections.singletonList("topic-first") 创建了一个包含 topic-first 的列表，表示消费者只订阅一个主题。
// 对应的生产者那边的主题
consumer.subscribe(Collections.singletonList("topic-first"));
//4.拉取消息
// 无限循环，用于不断地从 Kafka 中拉取消息。
while (true) {
// 读取数据，读取超时时间为100ms ，即每个1000ms拉取一次
ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(1000));
// ConsumerRecord 是 Kafka 中表示一条消息的对象，它包含消息的 key、value 以及一些元数据信息。
for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
System.out.println(consumerRecord.key());
System.out.println(consumerRecord.value());
}
}
}
}使用情景：
- 生产者发送消息，多个消费者订阅同一个主题(多个消费者都是一个组)只能有一个消费者收到消息 （一对一）
- 生产者发送消息，多个消费者订阅同一个主题(多个消费者不是一个组)所有消费者都能收到消息 （一对多）
4.springboot集成kafka
4.1导入spring-kafka依赖信息
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<!-- kafkfa -->
<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>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
<version>5.11.0-M2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
4.2在resources下创建文件application.yml
server:
port: 9991
spring:
application:
name: kafka-demo
kafka:
bootstrap-servers: localhost: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
4.3消息生产者
4.4消息消费者
传递消息为对象
目前springboot整合后的kafka，因为序列化器是StringSerializer，这个时候如果需要传递对象可以有 两种方式
方式一：可以自定义序列化器，对象类型众多，这种方式通用性不强，本章节不介绍
方式二：可以把要传递的对象进行转json字符串，接收消息后再转为对象即可，本项目采用这种方式
- 发送消息
- 接收消息
二、消息中间件对比
消息中间件对比-选择建议
![[Linux]：进程(下)](https://img-blog.csdnimg.cn/img_convert/f0c07e46c07b79a5f64d43840d99e37d.png)