Springboot整合RocketMQ 基本消息处理

简介: 1. 同步消息2. 异步消息3. 单向消息4. 延迟消息5. 批量消息6. 顺序消息 7. Tag过滤8. 广播消息

导入依赖  

<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 注解参数如下:

相关实践学习
RocketMQ一站式入门使用
从源码编译、部署broker、部署namesrv,使用java客户端首发消息等一站式入门RocketMQ。
消息队列 MNS 入门课程
1、消息队列MNS简介 本节课介绍消息队列的MNS的基础概念 2、消息队列MNS特性 本节课介绍消息队列的MNS的主要特性 3、MNS的最佳实践及场景应用 本节课介绍消息队列的MNS的最佳实践及场景应用案例 4、手把手系列:消息队列MNS实操讲 本节课介绍消息队列的MNS的实际操作演示 5、动手实验:基于MNS,0基础轻松构建 Web Client 本节课带您一起基于MNS,0基础轻松构建 Web Client
相关文章
|
26天前
|
消息中间件 JSON Java
SpringBoot+RabbitMQ 方式收发消息
SpringBoot+RabbitMQ 方式收发消息
14 0
|
1月前
|
NoSQL Java Redis
小白版的springboot中集成mqtt服务(超级无敌详细),实现不了掐我头!!!
小白版的springboot中集成mqtt服务(超级无敌详细),实现不了掐我头!!!
240 1
|
1月前
|
消息中间件 Java
springboot整合消息队列——RabbitMQ
springboot整合消息队列——RabbitMQ
74 0
|
2月前
|
消息中间件 JSON Java
如何利用springboot + rabbitmq发送邮件?
RabbitMQ相关知识请参考: RabbitMQ消息确认、消息持久化等核心知识总结 - 简书
29 2
|
2月前
|
消息中间件 Java Maven
一文搞懂Spring Boot整合RocketMQ
一文搞懂Spring Boot整合RocketMQ
92 0
|
2月前
|
消息中间件 存储 监控
搭建消息时光机:深入探究RabbitMQ_recent_history_exchange在Spring Boot中的应用【RabbitMQ实战 二】
搭建消息时光机:深入探究RabbitMQ_recent_history_exchange在Spring Boot中的应用【RabbitMQ实战 二】
32 1
|
2月前
|
消息中间件 监控 Java
Spring Boot中的RabbitMQ死信队列魔法:从异常到延迟,一网打尽【RabbitMQ实战 一】
Spring Boot中的RabbitMQ死信队列魔法:从异常到延迟,一网打尽【RabbitMQ实战 一】
59 0
|
3月前
|
消息中间件 存储 安全
SpringBoot与RabbitMQ详解与整合
SpringBoot与RabbitMQ详解与整合
57 0
|
3月前
|
消息中间件 数据库 RocketMQ
Springboot+RocketMQ通过事务消息优雅的实现订单支付功能
RocketMQ的事务消息,是指发送消息事件和其他事件需要同时成功或同时失败。比如银行转账, A银行的某账户要转一万元到B银行的某账户。A银行发送“B银行账户增加一万元”这个消息,要和“从A银 行账户扣除一万元”这个操作同时成功或者同时失败。RocketMQ采用两阶段提交的方式实现事务消息。
113 0
|
15天前
|
消息中间件 存储 负载均衡
消息队列学习之RabbitMQ
【4月更文挑战第3天】消息队列学习之RabbitMQ,一种基于erlang语言开发的流行的开源消息中间件。
15 0

热门文章

最新文章