【RocketMQ】SpringBoot集成RocketMQ

淡淡的烟草味﹌ 2024-03-24 18:34 51阅读 0赞

SpringBoot集成RocketMQ  
首先依旧是引入依赖

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

然后就可以编写发送不同类型消息的代码了

package blossom.project.springbootkp.seckillproducer;
    
    import blossom.project.springbootkp.seckillproducer.entity.MsgModel;
    import com.alibaba.fastjson.JSONObject;
    import org.apache.rocketmq.client.producer.SendCallback;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.apache.rocketmq.spring.support.RocketMQHeaders;
    import org.junit.jupiter.api.Test;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.boot.test.context.SpringBootTest;
    import org.springframework.messaging.Message;
    import org.springframework.messaging.support.MessageBuilder;
    
    import java.util.Arrays;
    import java.util.List;
    
    @SpringBootTest
    class SecKillProducerApplicationTests {
        
    
        private List<MsgModel> msgModels = Arrays.asList(
                new MsgModel("qwer", 1L, "下单"),
                new MsgModel("qwer", 1L, "短信"),
                new MsgModel("qwer", 1L, "物流"),
    
                new MsgModel("zxcv", 2L, "下单"),
                new MsgModel("zxcv", 2L, "短信"),
                new MsgModel("zxcv", 2L, "物流")
        );
    
        @Autowired
        private RocketMQTemplate rocketMQTemplate;
        @Test
        void syncProducer() {
        
            rocketMQTemplate.syncSend("bootTestTopic","使用springboot集成rocketmq");
        }
        @Test
        void asyncProducer(){
        
            rocketMQTemplate.asyncSend("bootTestTopic", "发送一条异步消息", new SendCallback() {
        
                @Override
                public void onSuccess(SendResult sendResult) {
        
                    System.out.println("发送成功");
                }
    
                @Override
                public void onException(Throwable throwable) {
        
                    System.out.println("发送失败"+throwable.getMessage());
                }
            });
        }
        @Test
        void oneWayProducer(){
        
            rocketMQTemplate.sendOneWay("bootTestTopic","发送一个单向消息");
        }
        @Test
        void delayProducer(){
        
            Message<String> message = MessageBuilder.withPayload("这是一条延迟消息").build();
            rocketMQTemplate.syncSend("bootTestTopic",message,3000,2);
        }
    
        @Test
        void orderedProducer(){
        
            msgModels.forEach(x->{
        
                String s = JSONObject.toJSONString(x);
                rocketMQTemplate.syncSendOrderly("orderlyTopic", s,x.getOrderSn());
            });
        }
        @Test
        void tagProducer(){
        
            rocketMQTemplate.syncSend("bootTestTopic:tagA","我是一个带标签的消息");
        }
        @Test
        void keyProducer(){
        
            Message<String> message = MessageBuilder.withPayload("我是一个带有key的消息").setHeader(RocketMQHeaders.KEYS, "testKey")
                    .build();
            rocketMQTemplate.syncSend("bootTestTopic",message);
        }
    }
    对于不同的消息类型，我们可以使用不同的方式去接收。
    创建一个顺序消息的监听器
    @Component
    @RocketMQMessageListener(
            topic = "orderlyTopic",
            consumerGroup = "boot-orderly-consumer-group",
            consumeMode = ConsumeMode.ORDERLY, //顺序消费模式 单线程
            maxReconsumeTimes = 5) //最大重试次数
    public class OrderlyMessageListener implements RocketMQListener<MessageExt> {
        
    
    
        @Override
        public void onMessage(MessageExt messageExt) {
        
            MsgModel msgModel = JSON.parseObject(new String(messageExt.getBody()), MsgModel.class);
            System.out.println(msgModel);
        }
    }
    普通的创建一个监听器
    @Component
    @RocketMQMessageListener(topic = "bootTestTopic",consumerGroup = "boot-consumuer-group")
    public class SimpleMessageListener implements RocketMQListener<MessageExt> {
        
    
        /**
         * 这个方法就是消费者方法
         * 这里的String就是消息内容
         * 这里的泛型就是这里的参数类型
         * 如果泛型指定了固定的类型 那么消息体就是我们的参数
         * 如果我们的类型设定为具体的类型 那么我们只能拿到消息体
         * 而如果我们把消息类型设定为MessageExt类型，那么我们可以拿到消息头
         * ------------------------------------------------
         * 只要这个方法不报错 就会直接完成消息的接收 而如果报错了 就会重试
         * @param msg
         */
        @Override
        public void onMessage(MessageExt msg) {
        
            String keys = msg.getKeys();
            System.out.println("接收到的keys为"+keys);
            String body = new String(msg.getBody());
            System.out.println("接收到的消息体为"+body);
    
        }
    }
    创建一个识别tag标签的监听器
    @RocketMQMessageListener(topic = "bootTestTopic",
        consumerGroup = "boot-tag-consumer-group",
    selectorType = SelectorType.TAG, //tag过滤模式
    selectorExpression = "tagA || tagB") //tag标签匹配模式
    public class TagMessageListener implements RocketMQListener<MessageExt> {
        
        @Override
        public void onMessage(MessageExt msg) {
        
            System.out.println(new String(msg.getBody()));
        }
    }