springmvc整合rocketMQ 解决问题：MQClientException: No route info of this topic, TopicTest

傷城~ 2021-12-12 16:15 1636阅读 0赞

## 整合步骤 ##

1.  下载rocketMq 启动（jdk1.8）
2.  创建生产者
3.  创建消费者
4.  测试

## 1、下载rocketMQ 与启动 ##

去Apache官网下载，解压后路径。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70]  
cd到bin目录，先启动NAMESERVER：start mqnamesrv.cmd  
显示如下图则表示正常启动。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 1]  
然后启动BORKER：start mqbroker.cmd -n 127.0.0.1:9876 autoCreateTopicEnable=true  
（备注：autoCreateTopicEnable=true 表示可以动态创建topic）  
显示如下图表示启动正常。（备注：端口10911的是borker的默认端口地址，端口9876的是我们配置到NAMESERVER的地址,）  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 2]

## 创建生产者 ##

pom.xml 依赖：

<dependency>
    			<groupId>org.apache.rocketmq</groupId>
    			<artifactId>rocketmq-client</artifactId>
    			<version>4.3.0</version>
    		</dependency>
    		
    		<dependency>
    			<groupId>com.alibaba</groupId>
    			<artifactId>fastjson</artifactId>
    			<version>1.2.29</version>
    		</dependency>
    		
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<version>4.12</version>
    			
    		</dependency>
    		<dependency>
    			<groupId>org.springframework</groupId>
    			<artifactId>spring-test</artifactId>
    			<version>4.2.6.RELEASE</version>
    		</dependency>

rocket-producer.xml 配置：

JUnitProducer.java 测试 生产消息类：

import org.apache.rocketmq.client.producer.DefaultMQProducer;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.common.message.Message;
    import org.apache.rocketmq.remoting.common.RemotingHelper;
    import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration({ "classpath:conf/rocket-producer.xml"})
    public class JUnitProducer { 
    
        @Autowired
        private DefaultMQProducer producer;
    
        @Test
        public void producerData() throws InterruptedException { 
            for (int i = 0; i < 10; i++) {   // 发10条消息
                try { 
                    Message msg = new Message("TopicTest", // topic
                            "TagA", // tag
                            ("你好" + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                    );
                    /*Message message = new Message(); message.setBody("test".getBytes());*/
    
                    // 调用producer的send()方法发送消息
                    // 这里调用的是同步的方式，所以会有返回结果
                    SendResult sendResult = producer.send(msg);
    
                    // 打印返回结果，可以看到消息发送的状态以及一些相关信息
                    System.out.println(sendResult);
                } catch (Exception e) { 
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
        }
    }

## 创建消费者 ##

pom.xml 与生产者一样  
rocket-consumer.xml：

<?xml version="1.0" encoding="UTF-8"?>
    <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd ">
    
    	<bean id="consumerImplTest" class="com.api.mq.RocketMQListener" />
    
    	<bean id="rocketmqConsumer" class="org.apache.rocketmq.client.consumer.DefaultMQPushConsumer" init-method="start" destroy-method="shutdown">
    		<property name="consumerGroup" value="concurrent_consumer"/>
    		<property name="namesrvAddr" value="192.168.100.223:9876"/>
    		<property name="messageListener" ref="consumerImplTest"/>
    		<property name="subscription">
    			<map>
    				<entry key="TopicTest">
    					<value>*</value>
    				</entry>
    			</map>
    		</property>
    	</bean>
    </beans>

创建 rocket-consumer.xml 中的 RocketMQListener 监听类

import java.util.List;
    
    import org.apache.commons.logging.Log;
    import org.apache.commons.logging.LogFactory;
    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;
    
    public class RocketMQListener implements MessageListenerConcurrently { 
    
    	private Log log = LogFactory.getLog(RocketMQListener.class);
    
    	@Override
    	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
    	    try { 
                for (MessageExt message : msgs) { 
                    String msgBody = new String(message.getBody());
                    log.info("msgId:" + message.getMsgId() + " msg:" + msgBody);
                }
            }catch (Throwable throwable){ 
    	        throwable.printStackTrace();
                log.error("MQ消费异常");
            }
    		//ConsumeConcurrentlyStatus.RECONSUME_LATER 消费失败，需要稍后重新消费
    		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    	}
    }

创建JUnitConsumer.java 测试消费 类

import org.junit.Test;
    import org.junit.runner.RunWith;
    import org.springframework.test.context.ContextConfiguration;
    import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
    
    @RunWith(SpringJUnit4ClassRunner.class)
    @ContextConfiguration({ "classpath:conf/rocket-consumer.xml"})
    public class JUnitConsumer { 
    
    
        @Test
        public void runConsumer() { 
            System.out.println("Consumer Started.");
    
            // 下面的代码把线程阻塞住,这样就可以先运行消费者再运行生产者.当然不要也可以,不要的话就得先运行生产者,
            //再运行消费者,生产者先把消息发送到MQ上,消费者启动后从MQ上拿消息
            synchronized (JUnitConsumer.class) { 
                while (true) { 
                    try { 
                        JUnitConsumer.class.wait();
                    } catch (Throwable e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
    }

## 4、测试 ##

1.  先启动NAMESERVER，再启动BORKER
2.  右键启动消费者 JUnitConsumer.java
3.  右键启动生产者JUnitProducer.java
4.  结果图：  
    启动 消费者：![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 3]  
    启动生产者 发送msg：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 4]  
    消费者接受消息并消费：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 5]

## 完成。 ##

总结：  
可能遇到的问题：启动消费者报错  
 org.apache.rocketmq.client.exception.MQClientException: No route info of this topic, TopicTest1   
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 6]  
目前网上常见的解决方法:

1.  broker没启动成功,启动命令有问题  
    linux: nohup sh mqbroker -n 192.168.180.133:9876 autoCreateTopicEnable=true  
    windows : mqbroker.exe -n localhost:9876 autoCreateTopicEnable=true
2.  broker没有自动创建topic  
    新版本中都是默认autoCreateTopicEnable=true  
    启动./mqbroker -n 127.0.0.1:9876 -p -p是查看配置信息  
    查看autoCreateTopicEnable的值是否为true
3.  不知道broker是否启动成功  
    查看broker.log,windows的默认在C:\\Users\\Admin\\logs\\rocketmqlogs  
    linux的在~/logs/rocketmqlogs/  
    看到register broker to name server localhost:9876 OK 说明已经成功注册到name server上了
4.  fastjson, slf4j的jar包缺失  
    其实这个新版本应该没啥影响,不过以防万一的话还是都加上, 看上面我贴出来的pom.xml
5.  版本不对 （我用的是4.3.0）
6.  自动创建topic失败，改成手动创建topic。创建命令：  
    D:\\Sofeware\\rocketmq-all-4.5.1-bin-release\\bin>mqadmin updateTopic -b 192.168.10  
    0.223:10911 -n 192.168.100.223:9876 -t TopicTest

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70]: /images/20211210/993ec71c208249c78971623b74281f54.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 1]: /images/20211210/c878f769660742b38addd8e934bf5263.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 2]: /images/20211210/03ce8cd2b0c94076b5986eb46d58747a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 3]: /images/20211210/54c784e30c454d748e34fe97c17dda28.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 4]: /images/20211210/a6580e67733e4b3aab669606acd5fe33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 5]: /images/20211210/7942610fb6a24ac49b7fd210337c398a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ZhbmdjaGFvMjAxMQ_size_16_color_FFFFFF_t_70 6]: /images/20211210/b88c81d6cd0044b48605e43353bd02bf.png