【Spring Boot】IDEA + Maven + Spring Boot + RocketMQ 4.5.1

淩亂°似流年 2024-04-17 19:34 10阅读 0赞

**【引言】**

在[《Linux RocketMQ 4.5.1安装及问题总结》][Linux RocketMQ 4.5.1]博客中，完成了RocketMQ服务端的搭建，并且完成了控制台的部署工作，以便在与客户端集成过程中查询及处理问题。本篇博客，将使用SpringBoot与RocketMQ集成，实现消息的生产与消费。

**【版本说明】**

*  Java Version：1.8
 *  Spring Boot Version：2.1.7.RELEASE
 *  RocketMQ Client Version：4.5.1

**【项目结构】**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]

新建一个Maven父项目，其子模块包括上述的消息生产者和消息消费者。

**【整合过程】**

1.  **父项目核心依赖：**

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

1.  **消息生产者模块**

*  **消息服务端配置：**

# 定义name-server地址
    spring.rocketmq.name-server=192.168.17.141:9876
    # 定义发布者组名
    spring.rocketmq.producer.group=group-0827
    # 定义要发送的topic
    spring.rocketmq.topic=topic-0827
    #应用端口
    server.port=8788
    #应用名称
    spring.application.name=springboot-rocketmq-producer

*  **消息生产者接口定义：**

/**
     * 消息生产者接口
     */
    public interface IProducer {
        /**
         * 同步发送MQ
         * @param topic
         * @param entity
         */
         void send(String topic, MQEntity entity);
    
        /**
         * 发送MQ,提供回调函数，超时时间默认3s
         * @param topic
         * @param entity
         * @param sendCallback
         */
         void send( String topic, MQEntity entity, SendCallback sendCallback );
    
        /**
         * 单向发送MQ，不等待服务器回应且没有回调函数触发，适用于某些耗时非常短，但对可靠性要求并不高的场景，例如日志收集。
         * @param topic
         * @param entity
         */
         void sendOneway(String topic, MQEntity entity);
    
    }

rocketmq消息发送可分为以上三种方式，所以这里分开定义了三种方式，可根据实际需要实现。

*  **初始化消息生产者**。这里使用的是springboot中PostConstruct注解，可以实现spring容器加载后执行该方法，实现消息生产者的启动

@PostConstruct
    public void defaultMQProducer() throws Exception {
         producer = new DefaultMQProducer();
         producer.setProducerGroup( this.producerGroup );
         producer.setNamesrvAddr( this.namesrvAddr );
         /*
          * Producer对象在使用之前必须要调用start初始化，初始化一次即可<br>
          * 注意：切记不可以在每次发送消息时，都调用start方法
          */
         producer.start();
         logger.info( "[{}:{}] start successd!",producerGroup,namesrvAddr );
     }

*  **同步消息发送方法实现示例**

public Message message(String topic, MQEntity entity) {
            String keys = UUID.randomUUID().toString();
            entity.setMqKey(keys);
            String tags = entity.getClass().getName();
            //  logger.info("业务:{},tags:{},keys:{},entity:{}",topic, tags, keys, entity);
            String smsContent = MessageFormat.format("业务:{0},tags:{1},keys:{2},entity:{3}",topic,tags,keys, JSONObject.toJSONString(entity));
            logger.info(smsContent);
    
            Message msg = null;
            try {
                msg = new Message(topic, tags, keys,
                        JSON.toJSONString(entity).getBytes(RemotingHelper.DEFAULT_CHARSET));
            } catch (UnsupportedEncodingException e) {
                logger.error("消息转码失败",e);
            }
            return msg;
        }
    
        @Override
        public void send(String topic, MQEntity entity) {
            Message msg = message(topic,entity);
            try {
                producer.send(msg);
            } catch (Exception e) {
                logger.error(entity.getMqKey().concat(":发送消息失败"), e);
                throw new RuntimeException("发送消息失败",e);
            }
    
        }

*  **生产消息接口，测试使用**。

@RestController
    @RequestMapping("/mq")
    public class ProducerController {
    
        @Autowired
        private IProducer iProducer;
    
        @Value("${spring.rocketmq.topic}")
        private String topic;
    
        @RequestMapping("/producemsg")
        public void producemsg(){
            MQEntity mqEntity = new MQEntity();
            mqEntity.addExt("createTime",new Date());
            mqEntity.addExt("msg","发送一条消息");
            iProducer.send(topic,mqEntity);
        }
    }

1.  **消息消费者模块**

*  **消息服务端配置**：

# 定义name-server地址
    spring.rocketmq.name-server=192.168.17.141:9876
    # 定义消费者组名
    spring.rocketmq.consumer.group=consumer
    # 定义要发送的topic
    spring.rocketmq.topic=topic-0827
    #应用端口
    server.port=8789
    #应用名称
    spring.application.name=springboot-rocketmq-consumer

*  **消费者订阅主题，监听器实现消息消费**

@PostConstruct
    public void defaultMQPushConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        try {
            consumer.subscribe(consumerTopic, "");
            consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) -> {
                consumeOrderlyContext.setAutoCommit(true);
                for (MessageExt msg : list) {
    
                    String msgContent = null;
                    try {
                        msgContent = new String(msg.getBody(),"UTF-8");
                    } catch (UnsupportedEncodingException e) {
                        logger.error("转码失败",e);
                    }
                    logger.info("######### MSG Content start ##########");
                    logger.info(JSON.toJSONString(msgContent));
                    logger.info("#########        END        ##########");
    
                }
                return ConsumeOrderlyStatus.SUCCESS;
            });
            consumer.start();
            logger.info("Consumer started.");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

同样，使用PostConstruct注解，使得可在项目启动后，执行对应的订阅方法。

**【问题】**

1.  **org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException: sendDefaultImpl call timeout**
    
    此问题原因在于，我们在上篇博客的安装服务端博客中，启动nameserver和broker的命令没有指定公网IP，所以，会抛出远程调用超时异常，解决方案：

*  进入distribution/target/rocketmq-4.5.1/rocketmq-4.5.1/conf目录，**在broker配置文件中添加公网IP地址**：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 1]
 *  **在启动命令中添加公网IP的参数**，如下：

- 启动nameserver
    nohup sh bin/mqnamesrv -n 192.168.17.141:9876 >/dev/null 2>&1 &
    - 启动broker
    nohup sh bin/mqbroker -n 192.168.17.139:9876 &

1.  **No route info of this topic 异常**

导致此问题的原因有很多，比如：

*  **Broker禁止自动创建Topic，且用户没有通过手工方式创建Topic**
    
    解决方案，在启动broker命令中添加自动创建参数，如下：

nohup sh bin/mqbroker -n 192.168.17.141:9876 -c conf/broker.conf autoCreateTopicEnable=true &

我的问题就属于这一种，启动添加参数后，客户端就可以正常发送消息了。

如果上述命令已经执行还是不成功，可以检查以下几个方面:

*  **检查下nameserver中是否成功注册了broker**，检查方式在bin目录下执行命令：

sh mqadmin clusterList -n localhost:9876

若出现以下结果，则表明连接是成功的。  
![在这里插入图片描述][20190831003018145.png]

*  **检查Producer是否连接到Name Server**

若按照上篇博客的安装配置教程，也提到防火墙的问题，那么在此就能避免此问题的出现了。

1.  **connect to 192.168.17.139:10911 failed**

此问题是在使用控制台的时候出现的，在配置文件中，指定了VIPChannel参数为true，  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 2]  
由于自己虚拟机ip地址换了，启动nameserver和broker都使用了新的IP，而broker配置文件中没有进行对应的修改，所以在使用控制台查看消息相关信息出现了该问题。

**【demo】**

本篇博客代码已上传至github，地址：https://github.com/huzhiting/springboot-rocketmq

**【总结】**

本篇博客重点在于springboot与rocketmq客户端代码的集成，而关于rocketmq相关的一些理论知识，也将在后面的博客中更新。

[Linux RocketMQ 4.5.1]: https://blog.csdn.net/hu_zhiting/article/details/100151959
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/79cd974284384aa3a091170e602461bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/24bec877729446a1a27c38031fbcbfe4.png
[20190831003018145.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/e9247f1205a843728cada6e235e538e2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTMwMzQyMjM_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/d8486daa450144d192c98343403bb36f.png