Kafka是什么？RocketMQ在SpringBoot中怎么整合使用/Kafka在SpringBoot中简单配置使用

写在前面&＃xff1a;
继续记录自己的SpringBoot学习之旅&＃xff0c;这次是SpringBoot应用相关知识学习记录。若看不懂则建议先看前几篇博客&＃xff0c;详细代码可在我的Gitee仓库SpringBoot克隆下载学习使用&＃xff01;

3.5.4.5 SpringBoot整合RocketMQ

3.5.4.5.1 RocketMQ安装

• 点击RockedMQ网站下载解压&＃xff0c;这里选择4.9.2 Binary版本
• 配置环境变量&＃xff0c;默认端口号是9876&＃xff0c;如图
  
  
  
• 启动namespaceServer服务&＃xff0c;点击bin里中的mqnamesrv,如图
  
  
• 启动mqbroker服务&＃xff0c;同上&＃xff0c;点击mqbroker.cmd&＃xff0c;如图
  **注&＃xff1a;**若报错则是jdk目录安装在C盘里的Program Files中&＃xff0c;解决办法是打开打开runbroker.cmd&＃xff0c;然后将‘%CLASSPATH%’加上英文双引号&＃xff0c;如图
• 消息生产消费测试
  
  • 生产者测试&＃xff0c;cmd命令输入tools tools org.apache.rockedmq.example.quickstart.Producer&＃xff0c;如图即可
  • 消费者测试&＃xff0c;cmd命令输入tools org.apache.rocketmq.example.quickstart.Consumer,如图

3.5.4.5.2 整合

• SpringBoot添加坐标&＃xff0c;在maven仓库搜索RocketMQ找到有Starter的&＃xff0c;复制粘贴&＃xff0c;如图
• 配置yml,如图
• 
• 编辑实现类发送同步消息&＃xff0c;并将之前实现类&＃64;Service注释掉&＃xff0c;如下&＃xff1a;

&＃64;Service
public class MessageServiceRocketmqImpl implements MessageService {
&＃64;Autowired
private RocketMQTemplate rocketMQTemplate;
&＃64;Override
public void sendMessage(String id) {
System.out.println("待发送短信的订单已纳入处理队列(rocketmq),id: " &＃43; id);
rocketMQTemplate.convertAndSend("order_id",id);
}
&＃64;Override
public String doMessage() {
return null;
}
}


,运行后如图

• 编写监听器&＃xff0c;如下&＃xff1a;

&＃64;Component
&＃64;RocketMQMessageListener(topic &＃61; "order_id",consumerGroup &＃61; "group_rocketmq_01")
public class MessageListener implements RocketMQListener<String> {
&＃64;Override
public void onMessage(String id) {
System.out.println("已完成短信发送业务(rocketmq),id&＃61;" &＃43; id);
}
}


,运行后如图

• 编辑实现类发送异步消息&＃xff0c;同上&＃xff0c;更改如图
  &＃xff0c;运行如图

3.5.4.6 SpringBoot整合kafka

3.5.4.6.1 Kafka安装

• 点击Kafka网站下载版本&＃xff0c;选择2.8.2版本
• 下载后解压缩即可
• 启动zookeeper服务&＃xff0c;默认端口2181&＃xff0c;CMD命令如下zookeeper-server-start.bat ../../config/zookeeper.properties&＃xff0c;若报错命令行太长&＃xff0c;使用短的文件名
• 启动kafka服务&＃xff0c;默认端口9092&＃xff0c;命令kafka-server-start.bat ../../config/server.properties即可

3.5.4.6.2 kafka测试

• 创建topic,cmd命令kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic demo,如图
• 查看topic,cmd命令kafka-topics.bat --zookeeper 127.0.0.1:2181 --list&＃xff0c;如图
• 启动生产者&＃xff0c;cmd命令kafka-console-producer.bat --broker-list localhost:9092 --topic demo
• 启动消费者&＃xff0c;cmd命令kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic demo --from-beginning
• 生产者与消费者交互&＃xff0c;生产者操作如图
  &＃xff0c;消费者如图
  

3.5.4.6.3 整合

• 添加坐标&＃xff0c;如下&＃xff1a;


<dependency>
<groupId>org.springframework.kafkagroupId>
<artifactId>spring-kafkaartifactId>
dependency>


• 配置yml&＃xff0c;如图
• 添加实现类与监听器&＃xff0c;如图
  添加监听器如图
  &＃xff0c;结果如图