Springboot集成RocketMq(消息的发布与订阅)

电玩女神 2023-03-01 10:53 17阅读 0赞

### gitee仓库地址：[https://gitee.com/wangwenlongGitHub/conformity.git][https_gitee.com_wangwenlongGitHub_conformity.git] ###

### 1、什么是RocketMq ###

RocketMQ 是阿里巴巴在 2012 年开源的分布式消息中间件，目前已经捐赠给 Apache 软件基金会，  
并于 2017 年 9 月 25 日成为 Apache 的顶级项目。作为经历过多次阿里巴巴双十一这种“超级工程”的洗礼并有稳定出色表现的国产中间件，  
以其高性能、低延时和高可靠等特性近年来已经也被越来越多的国内企业使用。

### 2、核心概念 ###

·Topic：消息主题，一级消息类型，生产者向其发送消息。  
·生产者：也称为消息发布者，负责生产并发送消息至 Topic。  
·消费者：也称为消息订阅者，负责从 Topic 接收并消费消息。  
·消息：生产者向 Topic 发送并最终传送给消费者的数据和（可选）属性的组合。  
·消息属性：生产者可以为消息定义的属性，包含 Message Key 和 Tag。  
·Group：一类生产者或消费者，这类生产者或消费者通常生产或消费同一类消息，且消息发布或订阅的逻辑一致。

### 3、消息收发模型 ###

消息队列 RocketMQ 版支持发布/订阅模型，消息生产者应用创建 Topic 并将消息发送到 Topic。  
消费者应用创建对 Topic 的订阅以便从其接收消息。通信可以是一对多（扇出）、多对一（扇入）和多对多。

![20200724135215451.png][]

### 4、应用场景 ###

·削峰填谷  
·异步解耦  
·顺序收发  
·大数据分析  
·分布式缓存同步  
·分布式事务一致性

### 5、消息类型概述 ###

![20200724135307315.png][]

**5.1、普通消息分类**

![20200724135339684.png][]

![20200724135355732.png][]

![20200724135411791.png][]

### 6、名词解释 ###

**Topic:消息主题，一级消息类型，通过 Topic 对消息进行分类**

**Tag:消息标签，二级消息类型，用来进一步区分某个 Topic 下的消息分类。**

**Topic和Tag的关系**

![20200724135549814.png][]

**消息（Message）:消息队列中信息传递的载体。**

**Message ID:消息的全局唯一标识，由消息队列 RocketMQ 版系统自动生成，唯一标识某条消息。**

**Message Key:消息的业务标识，由消息生产者（Producer）设置，唯一标识某个业务逻辑。**

**Producer:消息生产者，也称为消息发布者，负责生产并发送消息。**

**Consumer:消息消费者，也称为消息订阅者，负责接收并消费消息**

### 7、Linux安装RocketMq ###

**[https://blog.csdn.net/u013469562/article/details/104924510][https_blog.csdn.net_u013469562_article_details_104924510]**

### 8、Springboot整合RocketMq ###

**8.1、pom**

<dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-client</artifactId>
                <version>4.7.0</version>
            </dependency>

**注:maven依赖的RocketMq版本"最好"和Linux里的rocketMq保持一致，低版本的RocketMq会出现 (MQClientException: No route info of this topic)的错误，当然这个错误是可以修改配置来填补这个坑，但还是建议最好版本保持一致而且版本要稍微高点**

**8.2、application.yml**

rocket-mq: 
      namesrvAddr: 10.10.10.129:9876
      consumerGroup: MyConsumerGroup
      #主题
      topic: MyTopic
      #标签
      tag: MyTag
      instanceName: local
      producerGroup: producer

**8.3、读取配置的类**

package com.it.conformity.common.config;
    
    import lombok.Data;
    import lombok.Getter;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.stereotype.Component;
    
    /**
     * 配置文件
     *
     * @Author: 王文龙
     * @Date: 2020/7/23 10:08
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Data
    @Component
    @ConfigurationProperties(prefix = "rocket-mq")
    public class RocketMqConfig {
    
    
        private String namesrvAddr;
        private String consumerGroup;
        private String topic;
        private String tag;
        private String instanceName;
        private String producerGroup;
    
    }

**8.4、生产者**

package com.it.conformity.common.message.producer;
    
    import com.it.conformity.common.config.RocketMqConfig;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    import javax.annotation.Resource;
    
    /**
     * 生产者
     *
     * @Author: 王文龙
     * @Date: 2020/7/239:45
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Slf4j
    @Component
    public class Producer {
    
        /**
         * 消息的生产者
         *
         * @Describe: 这个类是应用程序打算发送信息的入口点
         */
        private DefaultMQProducer defaultMQProducer;
    
        /**
         * 注入消息的配置
         */
        @Resource
        private RocketMqConfig rocketMqConfig;
    
    
        /**
         * @Describe: @PostConstruct注解：被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器调用一次，类似于Serclet的inti()方法。
         * 被@PostConstruct修饰的方法会在构造函数之后，init()方法之前运行。
         */
        @PostConstruct
        public void init() throws MQClientException {
            this.defaultMQProducer = new DefaultMQProducer(rocketMqConfig.getProducerGroup());
            defaultMQProducer.setNamesrvAddr(rocketMqConfig.getNamesrvAddr());
            defaultMQProducer.setInstanceName(rocketMqConfig.getInstanceName());
            //关闭VIP通道，避免出现connect to <:10909> failed导致消息发送失败
            defaultMQProducer.setVipChannelEnabled(false);
            defaultMQProducer.start();
            log.info("RocketMq Producer start success");
        }
    
        /**
         * @Describe: @PreConstruct注解： 被@PreConstruct修饰的方法会在服务器卸载Servlet的时候运行，并且只会被服务器调用一次，
         * 类似于Servlet的destroy()方法。被@PreConstruct修饰的方法会在destroy()方法之后运行，在Servlet被彻底卸载之前
         */
        @PreDestroy
        public void destroy() {
            defaultMQProducer.shutdown();
        }
    
        public DefaultMQProducer getDefaultMQProducer() {
            return defaultMQProducer;
        }
    }

**8.5、自定义一个触发消息的控制器**

package com.it.conformity.sys.controller;
    
    import com.alibaba.fastjson.JSON;
    import com.it.conformity.common.config.RocketMqConfig;
    import com.it.conformity.common.message.producer.Producer;
    import com.it.conformity.common.util.JsonResultT;
    import com.it.conformity.sys.pojo.SysUser;
    import com.it.conformity.sys.service.SysUserService;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.springframework.security.access.prepost.PreAuthorize;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    import static org.apache.rocketmq.client.producer.SendStatus.SEND_OK;
    
    /**
     * @Author: 王文龙
     * @Date: 2020/6/1913:18
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Slf4j
    @RestController
    @RequestMapping("/conformity/sendMessage")
    public class SysSendMessageController {
    
    
        @Resource
        private SysUserService sysUserService;
        @Resource
        private RocketMqConfig rocketMqConfig;
        @Resource
        private Producer producer;
    
    
        @RequestMapping("/send")
        @PreAuthorize("hasPermission('/conformity/sendMessage/send','sys:message:send')")
        public JsonResultT send(Integer id) {
            try {
                SysUser byId = sysUserService.findById(id);
                if (byId != null) {
                    //发送消息
                    log.info("send msg:{}", byId.toString());
                    Message message = new Message(rocketMqConfig.getTopic(), rocketMqConfig.getTag(), JSON.toJSONString(byId).getBytes());
                    message.setDelayTimeLevel(3);
                    SendResult sendResult = producer.getDefaultMQProducer().send(message);
                    if (sendResult.getSendStatus() == SEND_OK) {
                        return new JsonResultT<>(201, "发送成功", sendResult.getSendStatus());
                    }
                    return new JsonResultT<>(120, "发送失败", byId);
                } else {
                    return new JsonResultT<>(210, "查询为空");
                }
            } catch (Exception e) {
                e.printStackTrace();
                return new JsonResultT<>(120, "信息出错", e.getMessage());
            }
        }
    
    }

**PostMan请求该接口，当看到date为SEND\_OK就说明消息发布成功**

{
        "code": 201,
        "msg": "发送成功",
        "date": "SEND_OK"
    }

**8.6、创建一个消费者**

package com.it.conformity.common.message.consumer;
    
    import com.it.conformity.common.config.RocketMqConfig;
    import com.it.conformity.common.message.RocketMqMessageWrapper;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
    import org.apache.rocketmq.client.exception.MQClientException;
    import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
    import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
    import org.springframework.stereotype.Component;
    
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    import javax.annotation.Resource;
    
    /**
     * 消费者
     * @Author: 王文龙
     * @Date: 2020/7/2317:12
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Slf4j
    @Component
    public class Consumer {
    
        private DefaultMQPushConsumer defaultMQPushConsumer;
        @Resource
        private RocketMqConfig rocketMqConfig;
        /**
         * 封装通用逻辑
         */
        @Resource
        private RocketMqMessageWrapper rocketMqMessageWrapper;
    
        @PostConstruct
        public void init() throws MQClientException {
            defaultMQPushConsumer = new DefaultMQPushConsumer(rocketMqConfig.getConsumerGroup());
            defaultMQPushConsumer.setNamesrvAddr(rocketMqConfig.getNamesrvAddr());
            defaultMQPushConsumer.setInstanceName(rocketMqConfig.getInstanceName());
    
            //设置订阅tag下的subExpression
            defaultMQPushConsumer.subscribe(rocketMqConfig.getTopic(), rocketMqConfig.getTag());
    
            // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
            // 如果非第一次启动，那么按照上次消费的位置继续消费
            defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //设置为集群消费(区别于广播消费)
            defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
    
            //注册监听 保证消费成功
            defaultMQPushConsumer.registerMessageListener(rocketMqMessageWrapper);
    
            //关闭VIP通道，避免接收不了消息
            defaultMQPushConsumer.setVipChannelEnabled(false);
    
            defaultMQPushConsumer.start();
            log.info("rocketMq Client start success");
        }
    
        @PreDestroy
        public void destroy() {
            defaultMQPushConsumer.shutdown();
        }
    }

**8.7、自定义消息监听**

package com.it.conformity.common.message;
    
    import com.it.conformity.sys.service.RocketMqMessageListener;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
    import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import java.util.List;
    
    /**
     * 封装逻辑
     * @Author: 王文龙
     * @Date: 2020/7/2317:18
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Service
    public class RocketMqMessageWrapper implements MessageListenerConcurrently {
    
    
        @Resource
        private RocketMqMessageListener rocketMqMessageListener;
    
        /**
         * 请求自定义消息接口的consumeMessage方法，进行具体逻辑的实现
         * @param list 讯息分机
         * @param context 并发消费上下文
         * @return ConsumeConcurrentlyStatus
         */
        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext context) {
            if (rocketMqMessageListener.onMessage(list, context)) {
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

**8.8、具体业务逻辑处理(这里就是普通的Service实现，不要被我起的名字给唬住了)**

package com.it.conformity.sys.service.impl;
    
    import com.alibaba.fastjson.JSON;
    import com.alibaba.fastjson.JSONObject;
    import com.fasterxml.jackson.databind.json.JsonMapper;
    import com.it.conformity.sys.pojo.SysUser;
    import com.it.conformity.sys.service.RocketMqMessageListener;
    import lombok.extern.slf4j.Slf4j;
    import org.apache.commons.lang3.StringUtils;
    import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
    import org.apache.rocketmq.common.message.MessageExt;
    import org.springframework.stereotype.Service;
    
    import java.nio.charset.Charset;
    import java.util.ArrayList;
    import java.util.List;
    
    /**
     * @Author: 王文龙
     * @Date: 2020/7/2317:28
     * @Version: 1.0
     * @Describe: 描述:
     */
    @Slf4j
    @Service
    public class RocketMqMessageListenerImpl implements RocketMqMessageListener {
    
        /**
         * 得到提供者的信息进行消费
         *
         * @param messages
         * @param context
         * @return true/false
         */
        @Override
        public boolean onMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
            log.info("消费者进行消费，并开始处理具体业务逻辑");
            boolean userName = false;
            for (MessageExt message : messages) {
                /**
                 * body里的数据就是提供者封装进去的数据
                 */
                byte[] body = message.getBody();
                String objString = StringUtils.toEncodedString(body, Charset.defaultCharset());
                SysUser user = JSON.parseObject(objString ,SysUser.class);
                if ("王文龙".equals(user.getName())){
                    userName=true;
                }
            }
            log.info("业务逻辑已处理完毕,结果为 {}",userName);
            return userName;
        }
    }

**这里只做了一个非常简单的业务逻辑判断，而序列化的目的只是看看提供者发布的消息，具体逻辑读者们私下随便玩，还有一点就是，当你发现消费失败后，消息会重复推送，这就对了，官方文档对此的解释为：当消费失败(也就是返回false时)会重复推送16次，16次 再返回false也不会再次推送了，我们顺便看一下控制台的打印结果**

2020-07-24 14:41:29.561  INFO 4368 --- [essageThread_19] c.i.c.s.s.i.RocketMqMessageListenerImpl  : 消费者进行消费，并开始处理具体业务逻辑
    2020-07-24 14:41:29.561  INFO 4368 --- [essageThread_19] c.i.c.s.s.i.RocketMqMessageListenerImpl  : 业务逻辑已处理完毕,结果为 true

[https_gitee.com_wangwenlongGitHub_conformity.git]: https://gitee.com/wangwenlongGitHub/conformity.git
[20200724135215451.png]: /images/20230208/31325eb8e66d4b029baf8cadc3dde396.png
[20200724135307315.png]: /images/20230208/a73a46e3c6b74f5595855d26179dcfed.png
[20200724135339684.png]: https://img-blog.csdnimg.cn/20200724135339684.png
[20200724135355732.png]: https://img-blog.csdnimg.cn/20200724135355732.png
[20200724135411791.png]: https://img-blog.csdnimg.cn/20200724135411791.png
[20200724135549814.png]: https://img-blog.csdnimg.cn/20200724135549814.png
[https_blog.csdn.net_u013469562_article_details_104924510]: https://blog.csdn.net/u013469562/article/details/104924510