RocketMQ学习笔记（2）—— 集成SpringBoot

前���知识：

RocketMQ学习笔记（1）—— 基础使用-CSDN博客

7.集成SpringBoot

7.1 producer

依赖

  org.apache.rocketmq  rocketmq-spring-boot-starter  2.0.2  com.alibaba  fastjson  2.0.25

配置文件

rocketmq:    name-server: hadoop104:9876    producer:        group: boot-producer-group    

配置了name server的ip以及生产者组的名称

消息发送

常用函数如下：

1. 发送同步消息：syncSend(topic,msg)
2. 发送异步消息：asyncSend(topic,msg,callback)
3. 发送单向消息：sendOneWay(topic,msg)
4. 发送延迟消息：syncSend(topic,msg,timeout,delayLevel)（这里的timeout与延迟时间无关，是消息发送的超时时间）
5. 发送顺序消息：syncSendOrderly(topic,msg,hashKey)（将hashKey相同的消息放入同一个队列中去）
6. 发送带tag的消息：syncSend(topic:tag,msg)
7. 发送带key的消息：MessageBuilder.withPayload(msg).setHeader(RocketMQHeaders.KEYS, key).build();（key是写在消息头中的）

代码：

@Testvoid bootProducer() {    //1.同步    rocketMQTemplate.syncSend("bootTestTopic","我是一条消息");    //2.异步    rocketMQTemplate.asyncSend("bootAsyncTopic", "我是一条异步消息", new SendCallback() {        @Override        public void onSuccess(SendResult sendResult) {            System.out.println("消息发送成功");        }        @Override        public void onException(Throwable e) {            System.out.println("消息发送异常");        }    });    //3.单向    rocketMQTemplate.sendOneWay("bootOnewayTopic","我是一条单向消息");    //4.延迟消息    Message msgDelay = MessageBuilder.withPayload("我是一条延迟消息").build();    rocketMQTemplate.syncSend("bootMsTopic",msgDelay,3000,2);    //5.顺序消息    List msgModels = Arrays.asList(        new MsgModel("qwer", 1, "下单"),        new MsgModel("qwer", 1, "短信"),        new MsgModel("qwer", 1, "物流"),        new MsgModel("zxcv", 2, "下单"),        new MsgModel("zxcv", 2, "短信"),        new MsgModel("zxcv", 2, "物流")    );    //发送时一般以json格式进行处理    msgModels.forEach(msgModel -> {        rocketMQTemplate.syncSendOrderly("bootOrderlyTopic", JSON.toJSONString(msgModel),msgModel.getOrderSn());    });    //6.带tag的消息    rocketMQTemplate.syncSend("bootTgTopic:tagA","我是tagA的消息");    //7.带key的消息    Message msgWithKey = MessageBuilder.withPayload("我是带key的消息").setHeader(        RocketMQHeaders.KEYS, "thisIsAKey"    ).build();    rocketMQTemplate.syncSend("bootKeyTopic",msgWithKey);}

7.2 consumer

依赖

  org.apache.rocketmq  rocketmq-spring-boot-starter  2.0.2  com.alibaba  fastjson  2.0.25

配置文件

server:    port: 8081rocketmq:    name-server: hadoop104:9876

设置springboot的端口号以及name server的ip

代码

基础结构

@Component@RocketMQMessageListener(    topic = "bootTestTopic",    consumerGroup = "boot-test-consumer-group")public class ABootSimpleListener implements RocketMQListener {    @Override    public void onMessage(MessageExt messageExt) {        System.out.println(Arrays.toString(messageExt.getBody()));    }}

1.监听器的使用：

通过注册监听器的形式监听topic中的内容并进行消费，主要是通过继承RocketMQListener类，实现onMessage方法来完成的；

RocketMQListener的泛型如果指定了固定的类型，那么onMessage方法中的参数类型与泛型指定的类型一致；MessageExt类型代表了消息的全部内容

如果消息消费过程中没有报错，则签收；如果报错了，则拒收并重试

2.注解的使用

通过@RocketMQMessageListener注解来为监听器设置一些属性：

如topic，设置订阅的主题；consumerGroup，设置消费者组

接下来发送一条简单的消息：rocketMQTemplate.syncSend("bootTestTopic","我是一条消息");

然后开启consumer监听器的springboot Application，即可进行消费：

顺序消息接收

@Component@RocketMQMessageListener(        topic = "bootOrderlyTopic",        consumerGroup = "boot-orderly-consumer-group",        consumeMode = ConsumeMode.ORDERLY, //顺序消费模式 单线程        maxReconsumeTimes = 5 //消费重试的次数)public class BOrderlyListener implements RocketMQListener {    @Override    public void onMessage(MessageExt messageExt) {        MsgModel msgModel = JSON.parseObject(new String(messageExt.getBody()), MsgModel.class);        System.out.println(msgModel.toString());    }}

需要在注解中添加consumeMode = ConsumeMode.ORDERLY，表示以单线程模式进行消费，避免因为并发导致的顺序错误

运行结果如下：

带tag的消息接收

@Component@RocketMQMessageListener(        topic = "bootTagTopic",        consumerGroup = "boot-tag-consumer-group",        selectorType = SelectorType.TAG,        selectorExpression = "tagA")public class CTagListener implements RocketMQListener {    @Override    public void onMessage(MessageExt messageExt) {        System.out.println(new String(messageExt.getBody()));    }}

需要通过selectorType指定过滤模式，默认使用tag进行过滤，还可以选择SQL92模式，但一般不用；

通过selectorExpression来设置用于过滤的表达式，默认是*，如果需要选取多个tag使用||分割即可，如tagA || tagB

运行结果如下：

7.3 两种消费模式

Rocketmq消息消费的模式分为两种：负载均衡模式和广播模式

• 负载均衡模式表示多个消费者交替消费同一个主题里面的消息
• 广播模式表示每个每个消费者都消费一遍订阅的主题的消息

  负载均衡模式

  需要在注解中设置：messageModel = MessageModel.CLUSTERING

  会将消息均匀地发送到各个队列中去

  示例代码：

  生产者：

  @Test    void modeTest() throws Exception{        for (int i = 1; i  /opt/module/rocketmq-4.9.2/logs/dashboard.log & 
  否则会报错：
     
  具体使用
   
  代码中
   
  需要在application.yml配置文件中设置：
   
  生产者：
     
  消费者：
     
  控制面板
   
  需要输入用户密码才能登录
     
  默认是admin/admin