导入依赖
<dependency> <groupId>org.apache.rocketmq</groupId> <artifactId>rocketmq-spring-boot-starter</artifactId> </dependency>
YAML配置
rocketmq: name-server: localhost:9876 producer: group: test-group-producer #生产者必须配备生产者组
1. 同步消息
同步消息是发送消息后等待Broker的响应，确保消息被成功接收。
生产者：
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { SendResult result = rocketMQTemplate.syncSend("test", MessageBuilder.withPayload("同步消息").build()); // SendResult result = rocketMQTemplate.syncSend("test", "同步消息"); System.out.println("发送状态：" + result.getSendStatus() + " 消息id：" + result.getMsgId()); }
2. 异步消息
异步消息是发送消息后不等待Broker响应，通过回调函数处理发送结果。
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { rocketMQTemplate.asyncSend("test", MessageBuilder.withPayload("异步消息").build(), new SendCallback() { @Override public void onSuccess(SendResult sendResult) { System.out.println("发送成功"+sendResult); } @Override public void onException(Throwable throwable) { System.out.println("发送失败"+throwable); } }); }
3. 单向消息
单向消息是发送消息后不等待Broker响应，也没有回调函数。
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { rocketMQTemplate.sendOneWay("test","单向消息"); }
4. 延迟消息
延迟消息是设置消息的延迟时间，确保消息在指定时间后才被消费。
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { //在RocketMQ中，timeout（超时时间）是指消息发送的最大等待时间。当你发送一个消息时，系统会等待一定的时间来获取发送结果，这个等待的时间就是超时时间。单位ms Message<String> message = MessageBuilder.withPayload("延迟消息").build(); //延迟级别 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 2对应5s SendResult result = rocketMQTemplate.syncSend("test", message, 2000, 2); }
5. 批量消息
批量消息是将多个消息打包成一个消息批次发送，提高发送效率。
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { List<String> list = Arrays.asList("blue", "red", "pink", "yello"); rocketMQTemplate.syncSend("test",list); }
上面所有生产者对应的消费者代码为：
@Component @RocketMQMessageListener(topic = "test",consumerGroup = "test-group-consumer") public class MQMsgListener implements RocketMQListener<MessageExt> { @Override public void onMessage(MessageExt message) { String msgId = message.getMsgId(); String msg = new String(message.getBody()); System.out.println("消息id："+msgId+"消息内容："+msg); } }
6. 顺序消息
顺序消息是保证同一个消息队列中的消息按顺序消费。
生产者代码：
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { for(int i=0;i<10;i++) { rocketMQTemplate.syncSendOrderly("test","顺序消息"+i,"1"); } }
消费者代码更改：
@Component @RocketMQMessageListener(topic = "test",consumerGroup = "test-group-consumer",consumeMode = ConsumeMode.ORDERLY) public class MQMsgListener implements RocketMQListener<MessageExt> { @Override public void onMessage(MessageExt message) { String msgId = message.getMsgId(); String msg = new String(message.getBody()); System.out.println("消息id："+msgId+"消息内容："+msg); } }
7. Tag过滤
消费者订阅的Tag和发送者设置的消息Tag相互匹配，则消息被投递给消费端进行消费。
生产者
@Autowired RocketMQTemplate rocketMQTemplate; @Test void contextLoads() { rocketMQTemplate.syncSend("test:test","hello"); }
消费者
@Component @RocketMQMessageListener(topic = "test",consumerGroup = "test-group-consumer",selectorType = SelectorType.TAG,selectorExpression = "test") public class MQMsgListener implements RocketMQListener<MessageExt> { @Override public void onMessage(MessageExt message) { String msgId = message.getMsgId(); String msg = new String(message.getBody()); System.out.println("消息id："+msgId+"消息内容："+msg); } }
8. 广播消息
消费者代码：
@Component @RocketMQMessageListener(topic = "test",consumerGroup = "test1-group-consumer",messageModel = MessageModel.BROADCASTING) public class MQMsgListener implements RocketMQListener<MessageExt> { @Override public void onMessage(MessageExt message) { String msgId = message.getMsgId(); String msg = new String(message.getBody()); System.out.println("消息id："+msgId+"消息内容："+msg); } }
@RocketMQMessageListener 注解参数如下：
