Spring Boot集成RocketMQ

pom依赖

        <!--spring-boot-starter的依赖--><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!--rocketmq的依赖--><dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.0</version></dependency>

配置文件

spring.application.name=rocketmq-demorocketmq.nameServer=192.168.21.110:9876
rocketmq.producer.group=rocketmq-test-producer-group
rocketmq.consumer.group=rocketmq-test-consumer-group

消息生产者

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;/*** @author qushen* @date 2025/1/8 9:21* @description: 创建消息生产者* @Version：1.0*/
@Service
public class RocketMQProducer {@Autowiredprivate RocketMQTemplate rocketMQTemplate;private final String topic = "demo-topic";// 1.同步发送消息public void sendSyncMessage(String message){rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(message).build());System.out.printf("同步发送结果: %s\n", message);}// 2.异步发送消息public void sendAsyncMessage(String message){rocketMQTemplate.asyncSend(topic, MessageBuilder.withPayload(message).build(), new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.printf("异步发送成功: %s\n", sendResult);}@Overridepublic void onException(Throwable throwable) {System.out.printf("异步发送失败: %s\n", throwable.getMessage());}});}// 3.单向发送消息public void sendOneWayMessage(String message){rocketMQTemplate.sendOneWay(topic, MessageBuilder.withPayload(message).build());System.out.println("单向消息发送成功");}
}

消息消费者

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;/*** @author qushen* @date 2025/1/8 9:22* @description: 创建消息消费者* @Version：1.0*/
@Service
@RocketMQMessageListener(topic = "demo-topic", consumerGroup = "rocketmq-test-consumer-group", messageModel = MessageModel.CLUSTERING)
public class RocketMQConsumer implements RocketMQListener<String> {@Overridepublic void onMessage(String s) {System.out.printf("收到消息: %s\n", s);}
}

消息控制器

import com.sws.rocketmq.services.RocketMQProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;@RestController
@RequestMapping("/api")
public class RocketController {@Autowiredprivate RocketMQProducer rocketMQProducer;/*** 发送消息后，发送方会等待 Broker 的响应，确保消息成功发送到 Broker。* 可靠性高，确保消息不会丢失。* @param message* @return*/@GetMapping("/sendSync")public String sendSync(@RequestParam String message) {rocketMQProducer.sendSyncMessage(message);return "同步消息发送成功";}/*** 发送消息后，发送方不会等待 Broker 的响应，而是通过回调函数处理发送结果。* 性能较高，适合高并发场景。* @param message* @return*/@GetMapping("/sendAsync")public String sendAsync(@RequestParam String message) {rocketMQProducer.sendAsyncMessage(message);return "异步消息发送中";}/*** 发送消息后，发送方不会等待 Broker 的响应，也不关心发送结果。* 性能最高，适合对可靠性要求不高的场景。* @param message* @return*/@GetMapping("/sendOneWay")public String sendOneWay(@RequestParam String message) {rocketMQProducer.sendOneWayMessage(message);return "单向消息发送成功";}
}

测试：发送同步消息

http://localhost:8080/api/sendSync?message=HelloSync

项目地址：rocketmq: Spring Boot 与 RocketMQ 集成