当前位置：首页 > news >正文

Springboot整合RocketMQ 基本消息处理

news 2025/8/21 8:41:14

1. 同步消息

2. 异步消息

3. 单向消息

4. 延迟消息

5. 批量消息

6. 顺序消息

7. Tag过滤

导入依赖

       <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId></dependency>

YAML配置

rocketmq:name-server: localhost:9876     # rocketMq的nameServer地址

1. 同步消息

同步消息是发送消息后等待Broker的响应，确保消息被成功接收。

生产者：

   @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid contextLoads() {SendResult result = rocketMQTemplate.syncSend("test", MessageBuilder.withPayload("同步消息").build());
//        SendResult result = rocketMQTemplate.syncSend("test", "同步消息");System.out.println("发送状态：" + result.getSendStatus() + " 消息id：" + result.getMsgId());}

2. 异步消息

异步消息是发送消息后不等待Broker响应，通过回调函数处理发送结果。

@AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid contextLoads() {rocketMQTemplate.asyncSend("test", MessageBuilder.withPayload("异步消息").build(), new SendCallback() {@Overridepublic void onSuccess(SendResult sendResult) {System.out.println("发送成功"+sendResult);}@Overridepublic void onException(Throwable throwable) {System.out.println("发送失败"+throwable);}});}

3. 单向消息

单向消息是发送消息后不等待Broker响应，也没有回调函数。

    @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid contextLoads() {rocketMQTemplate.sendOneWay("test","单向消息");}

4. 延迟消息

延迟消息是设置消息的延迟时间，确保消息在指定时间后才被消费。

 @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid contextLoads() {//在RocketMQ中，timeout（超时时间）是指消息发送的最大等待时间。当你发送一个消息时，系统会等待一定的时间来获取发送结果，这个等待的时间就是超时时间。单位msMessage<String> message = MessageBuilder.withPayload("延迟消息").build();//延迟级别 "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h" 2对应5sSendResult result = rocketMQTemplate.syncSend("test", message, 2000, 2);}

5. 批量消息

批量消息是将多个消息打包成一个消息批次发送，提高发送效率。

    @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid 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> {@Overridepublic void onMessage(MessageExt message) {String msgId = message.getMsgId();String msg = new String(message.getBody());System.out.println("消息id："+msgId+"消息内容："+msg);}
}

6. 顺序消息

顺序消息是保证同一个消息队列中的消息按顺序消费。

生产者代码：

    @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid 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> {@Overridepublic void onMessage(MessageExt message) {String msgId = message.getMsgId();String msg = new String(message.getBody());System.out.println("消息id："+msgId+"消息内容："+msg);}
}

7. Tag过滤

消费者订阅的Tag和发送者设置的消息Tag相互匹配，则消息被投递给消费端进行消费。

生产者

    @AutowiredRocketMQTemplate rocketMQTemplate;@Testvoid 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> {@Overridepublic void onMessage(MessageExt message) {String msgId = message.getMsgId();String msg = new String(message.getBody());System.out.println("消息id："+msgId+"消息内容："+msg);}
}

@RocketMQMessageListener 注解参数如下：