RocketMQ SpringBoot 采坑

￡神魔★判官ぃ 2021-10-12 18:22 436阅读 0赞

## 1、项目结构： ##

![在这里插入图片描述][20190819000023272.png]

## 2、Provider ##

**2.1、Provider pom**

<parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath/> 
    </parent>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
        
        <dependency>
            <groupId>org.apache.rocketmq</groupId>
            <artifactId>rocketmq-client</artifactId>
            <version>4.4.0</version>
        </dependency>
    </dependencies>

**2.2、application.yml**

server:
      port: 8080
    rocketmq:
      producer:
        groupName: hello_group_provider
        nameServer: 192.168.92.39
        port: 9876

**2.3、RocketMQProducer**

@Component
    public class RocketMQProducer {
        @Value("${rocketmq.producer.groupName}")
        private String groupName;
        @Value("${rocketmq.producer.nameServer}")
        private String nameServer;
        @Value("${rocketmq.producer.port}")
        private String port;
        private DefaultMQProducer producer;
        /**
         * 功能描述 获取当前producer
         */
        @Bean
        public DefaultMQProducer getProducer() {
            //设置producer 并且开启producer
            producer = new DefaultMQProducer(groupName);
            producer.setNamesrvAddr(nameServer + ":" + port);
            try {
                producer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            return producer;
        }
    }

**2.4、Service**

@Service
    public class ProducerService {
        @Autowired
        private DefaultMQProducer producer;
        private static final String topic = "helloTopic";
        private static final String tag = "helloTag";
        public void sendMsg() {
            Message msg = new Message(topic, tag, "hello RocketMQ".getBytes());
            try {
                SendResult send = producer.send(msg);
                System.out.println(send);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

**2.5、测试方法**

@RunWith(SpringRunner.class)
    //主application方法
    @SpringBootTest(classes = RocketMQProducerApplication.class)
    public class ProducerTest {
        @Autowired
        private ProducerService producerService;
        @Test
        public void testSend() {
            producerService.sendMsg();
        }
    }

**2.6、运行**  
No route info of this topic, xxxxxxxxxx  
解决思路：maven版本必须和RocketMQ b版本一致  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]

发送超时  
rocketmq都会根据当前网卡选择一个IP使用，当你的机器有多块网卡时，很有可能会有问题。比如，我遇到的问题是我机器上有两个IP，一个公网IP，一个私网IP, 因此需要配置broker.conf 指定当前的公网ip, 然后重新启动broker。

#修改文件
    vim  conf/broker.conf
    brokerIP1=192.168.92.39
    #重启broker
    sh ./mqshutdown broker
    nohup sh ./mqbroker -n localhost:9876  -c ../conf/broker.conf   autoCreateTopicEnable=true   >/dev/null  2>&1  &

![在这里插入图片描述][20190819000249355.png]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]  
\*\***2,7、重新运行**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]

## 3、Customer ##

**3.1、pom文件同Customer**

**3.2、application.yml**

server:
      port: 8081
    rocketmq:
      consumer:
        groupName: hello_group_consumer
        nameServer: 192.168.92.39
        port: 9876

**3.3、消费者**

@Component
    public class RocketMQCustomer {
        @Value("${rocketmq.consumer.groupName}")
        private String groupName;
        @Value("${rocketmq.consumer.nameServer}")
        private String nameServer;
        @Value("${rocketmq.consumer.port}")
        private String port;
        private static final String topic = "helloTopic";
        @Bean
        public DefaultMQPushConsumer getConsumer() {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
            //配置组名 IP
            consumer.setNamesrvAddr(nameServer + ":" + port);
            consumer.setConsumerGroup(groupName);
            //最后偏移量读取消息
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            try {
                //订阅主题，监听主题下哪些标签
                consumer.subscribe(topic, "*");
                consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                    try {
                        //默认一条一条消息消费
                        Message msg = msgs.get(0);
                        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody()));
                        String topic = msg.getTopic();
                        String body = new String(msg.getBody(), "utf-8");
                        String tags = msg.getTags();
                        String keys = msg.getKeys();
                        System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg=" + body);
                        //返回成功
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        //返回稍后进行投递
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                });
                //开始
                consumer.start();
            } catch (MQClientException e) {
                e.printStackTrace();
            }
            return consumer;
        }
    }

**3.4、启动消费者**  
![在这里插入图片描述][20190819000419832.png]

[20190819000023272.png]: /images/20211013/83be5698c9ab499f8a810595d2e60c26.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70]: /images/20211013/870c069849c147c1938eb6e38297b68e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 1]: /images/20211013/484fb86b817c4cce9c1a46c678c456c4.png
[20190819000249355.png]: /images/20211013/c6a9cd1e62ac44d0bd99ed0b2083d5d5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 2]: /images/20211013/522bd675bd61412282a05baf60f9e33f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM0MTI1OTk5_size_16_color_FFFFFF_t_70 3]: /images/20211013/afa9ab68777a46dd840331bfe291e514.png
[20190819000419832.png]: /images/20211013/26f6868e8e334efa933c5524215259df.png