Springboot整合RocketMQ 基本消息处理

news2024/11/18 21:25:26

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的响应，确保消息被成功接收。

生产者：

   @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);
    }
}

@RocketMQMessageListener 注解参数如下：