SpringBoot整合 RocketMQ

我不是女神ヾ 2022-10-15 10:55 169阅读 0赞

### SpringBoot整合 RocketMQ ###

*  前言
 *  一、引入maven
 *  二、修改配置文件
 *  三、创建生产者
 *   *   *  工具类
 *  四、创建消费者
 *  五、创建Controller
 *  六、查看结果

--------------------

# 前言 #

springboot版本为2.2.2.RELEASE  
RocketMQ版本为4.5.1  
RocketMQ Docker版安装[https://blog.csdn.net/qq\_43548590/article/details/117219211?spm=1001.2014.3001.5501][https_blog.csdn.net_qq_43548590_article_details_117219211_spm_1001.2014.3001.5501]

--------------------

提示：以下是本篇文章正文内容，下面案例可供参考

# 一、引入maven #

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-devtools</artifactId>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
            #rocketmq
            <dependency>
                <groupId>org.apache.rocketmq</groupId>
                <artifactId>rocketmq-spring-boot-starter</artifactId>
                <version>2.2.0</version>
            </dependency>
        </dependencies>

# 二、修改配置文件 #

spring:
      application:
        name: rocketmq
    server:
      port: 8802
    rocketmq:
      name-server: 8.131.252.185:9876
      producer:
        group: dome-mq
        groupName: hyh-rocketmq-group

# 三、创建生产者 #

package com.koala.service;
    
    
    import com.koala.config.RocketMqHelper;
    import com.koala.entity.Person;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    import java.util.List;
    
    @Service
    public class ProducerService { 
    
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        @Resource
        private RocketMqHelper rocketMqHelper;
    	#使用RocketMQTemplate发送同步
        public void send(String message){ 
            rocketMQTemplate.convertAndSend("test-topic",message);
        }
    
    	#使用RocketMqHelper工具类发送异步
        public void helper(Person person){ 
            rocketMqHelper.asyncSend("PERSON_ADD", MessageBuilder.withPayload(person).build());
        }
    }

### 工具类 ###

package com.koala.config;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.Message;
    import javax.annotation.PostConstruct;
    import javax.annotation.PreDestroy;
    
    @Component
    @Slf4j
    public class RocketMqHelper { 
    
        /**
         * rocketmq模板注入
         */
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
    
        @PostConstruct
        public void init() { 
            log.info("---RocketMq助手初始化---");
        }
    
        /**
         * 发送异步消息
         *
         * @param topic   消息Topic   如果要使用tag 就是形如  topic:tags  中间以: 隔开
         * @param message 消息实体
         */
        public void asyncSend(Enum topic, Message<?> message) { 
            asyncSend(topic.name(), message, getDefaultSendCallBack());
        }
    
    
        /**
         * 发送异步消息
         *
         * @param topic        消息Topic
         * @param message      消息实体
         * @param sendCallback 回调函数
         */
        public void asyncSend(Enum topic, Message<?> message, SendCallback sendCallback) { 
            asyncSend(topic.name(), message, sendCallback);
        }
    
        /**
         * 发送异步消息
         *
         * @param topic   消息Topic
         * @param message 消息实体
         */
        public void asyncSend(String topic, Message<?> message) { 
            rocketMQTemplate.asyncSend(topic, message, getDefaultSendCallBack());
        }
    
        /**
         * 发送异步消息
         *
         * @param topic        消息Topic
         * @param message      消息实体
         * @param sendCallback 回调函数
         */
        public void asyncSend(String topic, Message<?> message, SendCallback sendCallback) { 
            rocketMQTemplate.asyncSend(topic, message, sendCallback);
        }
    
        /**
         * 发送异步消息
         *
         * @param topic        消息Topic
         * @param message      消息实体
         * @param sendCallback 回调函数
         * @param timeout      超时时间
         */
        public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout) { 
            rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout);
        }
    
        /**
         * 发送异步消息
         *
         * @param topic        消息Topic
         * @param message      消息实体
         * @param sendCallback 回调函数
         * @param timeout      超时时间
         * @param delayLevel   延迟消息的级别
         */
        public void asyncSend(String topic, Message<?> message, SendCallback sendCallback, long timeout, int delayLevel) { 
            rocketMQTemplate.asyncSend(topic, message, sendCallback, timeout, delayLevel);
        }
    
        /**
         * 发送顺序消息
         *
         * @param message
         * @param topic   如果要使用tag 就是形如  topic:tags  中间以: 隔开**
         * @param hashKey
         */
        public void syncSendOrderly(Enum topic, Message<?> message, String hashKey) { 
            syncSendOrderly(topic.name(), message, hashKey);
        }
    
    
        /**
         * 发送顺序消息
         *
         * @param message
         * @param topic
         * @param hashKey
         */
        public void syncSendOrderly(String topic, Message<?> message, String hashKey) { 
            log.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey);
            rocketMQTemplate.syncSendOrderly(topic, message, hashKey);
        }
    
        /**
         * 发送顺序消息
         *
         * @param message
         * @param topic
         * @param hashKey
         * @param timeout
         */
        public void syncSendOrderly(String topic, Message<?> message, String hashKey, long timeout) { 
            log.info("发送顺序消息，topic:" + topic + ",hashKey:" + hashKey + ",timeout:" + timeout);
            rocketMQTemplate.syncSendOrderly(topic, message, hashKey, timeout);
        }
    
        /**
         * 默认CallBack函数
         *
         * @return
         */
        private SendCallback getDefaultSendCallBack() { 
            return new SendCallback() { 
                @Override
                public void onSuccess(SendResult sendResult) { 
                    log.info("---发送MQ成功---");
                }
    
                @Override
                public void onException(Throwable throwable) { 
                    log.error("---发送MQ失败---"+throwable.getMessage(), throwable.getMessage());
                }
            };
        }
    
    
        @PreDestroy
        public void destroy() { 
            log.info("---RocketMq助手注销---");
        }
    }

# 四、创建消费者 #

package com.koala.config;
    
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
    
    @RocketMQMessageListener(topic = "test-topic", consumerGroup = "${rocketmq.producer.groupName}")
    @Component
    public class ConsumerListener implements RocketMQListener<String> { 
        @Override
        public void onMessage(String s) { 
            System.out.println("接收到的消息"+s);
        }
    }

# 五、创建Controller #

package com.koala.controller;
    
    import com.koala.entity.Person;
    import com.koala.service.ProducerService;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    import java.util.ArrayList;
    import java.util.List;
    
    @RestController
    @Slf4j
    public class ProducerController { 
    
        @Resource
        private ProducerService producerService;
    
    
        @GetMapping("/send")
        public String Send(String message){ 
            producerService.send(message);
            return "success";
        }
    
        @GetMapping("/add")
        public String addPerson(Person person){ 
            log.info("参数"+person.toString());
            producerService.helper(person);
            return "Add success!";
        }
    
        @GetMapping("/order")
        public String order(){ 
            List<Person> personList = new ArrayList<>();
            for (int i=0;i<3;i++){ 
                Person person = new Person();
                person.setName("name"+i);
                person.setAge(i);
                personList.add(person);
            }
            producerService.orderMessage(personList);
            return "Add success!";
        }
    }

# 六、查看结果 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70 1]

[https_blog.csdn.net_qq_43548590_article_details_117219211_spm_1001.2014.3001.5501]: https://blog.csdn.net/qq_43548590/article/details/117219211?spm=1001.2014.3001.5501
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70]: /images/20221014/fdcd1cfef2d54d1188c1824cc74fa2d2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNTQ4NTkw_size_16_color_FFFFFF_t_70 1]: /images/20221014/720525db7f264d5db3ab9a90e8520df8.png