springboot集成rocketmq

太过爱你忘了你带给我的痛 2023-01-13 04:01 138阅读 0赞

[rocketmq安装参考][rocketmq]

1.集成依赖，pom中引入以下依赖

<dependency>
 <groupId>org.apache.rocketmq</groupId>
 <artifactId>rocketmq-spring-boot-starter</artifactId>
 <version>2.1.0</version>
 </dependency>

2.在application.properties添加配置

rocketmq.name-server=127.0.0.1:9876
    rocketmq.producer.group=test-group

定义不同类型消息发送的方法，同时在注释部分说明具体方法的使用场景。其中将发送的参数封装为MqMsg对象

import lombok.Data;
    
    @Data
    public class MqMsg {
        /**
         * 一级消息：消息topic
         */
        private String topic;
        /**
         * 二级消息：消息topic对应的tags
         */
        private String tags;
        /**
         * 消息内容
         */
        private String content;
        // 省略getter/setter方法
    }

发送消息生产者service接口

import com.example.demo.util.MqMsg;
 public interface RocketMqService {
 /**
 * 同步发送消息 
 * 
 * 当发送的消息很重要是，且对响应时间不敏感的时候采用sync方式;
 *
 * @param mqMsg 发送消息实体类
 */
 void send(MqMsg mqMsg);
 /**
 * 异步发送消息，异步返回消息结果 
 * 
 * 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式；
 *
 * @param mqMsg 发送消息实体类
 */
 void asyncSend(MqMsg mqMsg);
 /**
 * 直接发送发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送; 
 * 
 * 当发送的消息不重要时，采用one-way方式，以提高吞吐量；
 *
 * @param mqMsg 发送消息实体类
 */
 void syncSendOrderly(MqMsg mqMsg);
 }

service接口实现类

import com.example.demo.service.RocketMqService;
    import com.example.demo.util.MqMsg;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Service;
    
    import javax.annotation.Resource;
    @Service("rocketMqService")
    public class RocketMqServiceImpl implements RocketMqService {
        private static final Logger log = LoggerFactory.getLogger(RocketMqServiceImpl.class);
        @Resource
        private RocketMQTemplate rocketMQTemplate;
        @Override
        public void send(MqMsg mqMsg) {
            log.info("send发送消息到mqMsg={}", mqMsg);
            rocketMQTemplate.send(mqMsg.getTopic() + ":" + mqMsg.getTags(),
                    MessageBuilder.withPayload(mqMsg.getContent()).build());
        }
        @Override
        public void asyncSend(MqMsg mqMsg) {
            log.info("asyncSend发送消息到mqMsg={}", mqMsg);
            rocketMQTemplate.asyncSend(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent(),
                    new SendCallback() {
                        @Override
                        public void onSuccess(SendResult sendResult) {
                            // 成功不做日志记录或处理
                        }
                        @Override
                        public void onException(Throwable throwable) {
                            log.info("mqMsg={}消息发送失败", mqMsg);
                        }
                    });
        }
        @Override
        public void syncSendOrderly(MqMsg mqMsg) {
            log.info("syncSendOrderly发送消息到mqMsg={}", mqMsg);
            rocketMQTemplate.sendOneWay(mqMsg.getTopic() + ":" + mqMsg.getTags(), mqMsg.getContent());
        }
    }

定义常量类MqTopicConstant，主要是统一定义对应的topic和tag

public class MqTopicConstant {
        /**
         * 示例消息队列，topic1个
         */
        public static final String DEMO_TOPIC = "test-top-1";
        /**
         * 注册tag
         */
        public static final String DEMO_TAG_REGISTERED = "registered";
        /**
         * 修改tag
         */
        public static final String DEMO_TAG_MODIFY = "modify";
        /**
         * consumer group
         */
        public static final String DEMO_CONSUMER_GROUP_REGISTERED = "consumer_test-top-1_registered";
        public static final String DEMO_CONSUMER_GROUP_MODIFY = "consumer_test-top-1_modify";
    }

### 消费者示例 ###

import com.example.demo.common.MqTopicConstant;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
 
 @Service
 @RocketMQMessageListener(topic = MqTopicConstant.DEMO_TOPIC
 , consumerGroup = MqTopicConstant.DEMO_CONSUMER_GROUP_MODIFY
 , selectorExpression = MqTopicConstant.DEMO_TAG_MODIFY)
 public class MqModifyListenerDemo implements RocketMQListener<String> {
 
 private static final Logger log = LoggerFactory.getLogger(MqModifyListenerDemo.class);
 @Override
 public void onMessage(String message) {
 log.info("received modify message: {}", message);
 }
 
 }

测试controller

import com.example.demo.common.MqTopicConstant;
 import com.example.demo.service.RocketMqService;
 import com.example.demo.util.MqMsg;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
 import java.util.HashMap;
 import java.util.Map;
 
 @RestController
 @RequestMapping("/demo")
 public class MqController {
 @Resource
 private RocketMqService rocketMqService;
 @GetMapping("/send")
 public void send() {
 MqMsg mqMsg = new MqMsg();
 mqMsg.setTopic(MqTopicConstant.DEMO_TOPIC);
 mqMsg.setTags(MqTopicConstant.DEMO_TAG_MODIFY);
 // 此处可为其他VO对象，替换掉Map
 Map<String, String> userInfo = new HashMap<>();
 userInfo.put("username", "wangsheng");
 userInfo.put("age", "26");
 // 此处可封装为json等格式
 mqMsg.setContent(userInfo.toString());
 // 发送修改消息
 rocketMqService.send(mqMsg);
 }
 
 @PostConstruct
 public void test(){
 send();
 }
 }

启动可以看到消息发送成功并消费

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xldHRlcnNz_size_16_color_FFFFFF_t_70][]

[rocketmq]: https://mp.csdn.net/editor/html/115723030
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xldHRlcnNz_size_16_color_FFFFFF_t_70]: /images/20221022/72eb4fccd81048cd91560f7676dd46c5.png