下面是如何在Spring Boot项目中整合RocketMQ并实现发送提醒消息的详细过程和关键代码。这个过程包括配置RocketMQ客户端，编写消息发送和接收的代码，以及一些基本的Spring Boot配置。
一、环境准备
- 下载并启动RocketMQ：
- 下载并解压RocketMQ的安装包。
- 启动Name Server和Broker。
1. nohup sh mqnamesrv & 2. nohup sh mqbroker -n localhost:9876 &
添加Maven依赖： 在pom.xml文件中添加RocketMQ的依赖。
<dependencies> <dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> <version>2.2.0</version> </dependency> </dependencies>
二、配置文件
在application.yml文件中添加RocketMQ的配置。
rocketmq: name-server: 127.0.0.1:9876 producer: group: springboot-producer-group consumer: group: springboot-consumer-group topic: ReminderTopic
三、生产者代码
通过上述步骤，我们已经成功在Spring Boot项目中整合了RocketMQ，并实现了发送和接收提醒消息的功能。这个示例可以根据实际需求进一步扩展和优化，比如增加消息重试机制、消息过滤等功能。
创建消息发送服务：
package com.example.rocketmq.producer; import org.apache.rocketmq.spring.core.RocketMQTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @Service public class ReminderProducer { @Autowired private RocketMQTemplate rocketMQTemplate; public void sendReminder(String message) { rocketMQTemplate.convertAndSend("ReminderTopic", message); } }
调用消息发送服务：
package com.example.rocketmq.controller; import com.example.rocketmq.producer.ReminderProducer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; @RestController public class ReminderController { @Autowired private ReminderProducer reminderProducer; @GetMapping("/sendReminder") public String sendReminder(@RequestParam String message) { reminderProducer.sendReminder(message); return "Reminder sent: " + message; } }
四、消费者代码
创建消息接收服务：
package com.example.rocketmq.consumer; import org.apache.rocketmq.spring.annotation.MessageModel; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; import org.springframework.stereotype.Service; @Service @RocketMQMessageListener(topic = "ReminderTopic", consumerGroup = "springboot-consumer-group", messageModel = MessageModel.CLUSTERING) public class ReminderConsumer implements RocketMQListener<String> { @Override public void onMessage(String message) { System.out.println("Received reminder: " + message); // 在这里处理接收到的提醒消息 } }
五、启动项目
通过Spring Boot的主程序启动项目
package com.example.rocketmq; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class RocketMqApplication { public static void main(String[] args) { SpringApplication.run(RocketMqApplication.class, args); } }
六、测试
- 启动Spring Boot应用。
- 访问
http://localhost:8080/sendReminder?message=HelloRocketMQ发送提醒消息。 - 检查控制台输出，确认消费者接收到消息。
七、详细解析
配置文件：
rocketmq.name-server：指定RocketMQ Name Server的地址。
rocketmq.producer.group：指定生产者组的名称。
rocketmq.consumer.group：指定消费者组的名称。
rocketmq.consumer.topic：指定要消费的Topic。
消息发送服务：
RocketMQTemplate：Spring提供的RocketMQ模板类，用于发送消息。
convertAndSend方法：用于将消息对象转换为字符串并发送到指定的Topic。
消息接收服务：
RocketMQMessageListener注解：用于标记一个类为RocketMQ消息监听器。
onMessage方法：用于处理接收到的消息。
通过上述步骤，我们已经成功在Spring Boot项目中整合了RocketMQ，并实现了发送和接收提醒消息的功能。这个示例可以根据实际需求进一步扩展和优化，比如增加消息重试机制、消息过滤等功能。
