SpringBoot整合RocketMQ

Dear 丶 2022-11-14 14:45 190阅读 0赞

#### 一、RocketMQ应用场景 ####

*  削峰填谷
 *  异步解耦
 *  顺序收发
 *  分布式事务一致性
 *  大数据分析
 *  分布式缓存同步

#### 二、springboot2.4整合rocketmq ####

##### 1.pom.xml添加 #####

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

##### 2、application.yml #####

server:
      port: 8081
    rocketmq:
      name-server: 192.168.0.11:9876
      topic: lss0555
      producer:
        send-message-timeout: 10000
        group: group-lss
    spring:
      application:
        name: rockettest-lss

##### 3、生产 者 #####

@Component
    public class SpringProducer { 
        private Logger logger = LoggerFactory.getLogger(Consumer.class);
    
        @Resource
        private RocketMQTemplate rocketMQTemplate;
    
        //发送消息
        public void sendMessage(String msg, String topic) { 
            rocketMQTemplate.convertAndSend(topic, msg);
        }
    
        //发送事务消息
        public void sendMessageIntransaction(String msg, String topic) { 
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader(RocketMQHeaders.TRANSACTION_ID, UUID.randomUUID().toString())
                    .setHeader(RocketMQHeaders.TAGS, "tag-lss0555")
                    .setHeader("userid", "lss05555")
                    .build();
            String destination = topic;
            TransactionSendResult sendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, destination);
            logger.info("事务消息发送结果：", sendResult.toString());
        }
    }

##### 4、消费者 #####

通过定义topic与consumerGroup来接收信息

@Component
    @RocketMQMessageListener(topic = "lss0555",consumerGroup = "group-lss")
    public class Consumer implements RocketMQListener<String> { 
        private Logger logger = LoggerFactory.getLogger(Consumer.class);
    
        @Override
        public void onMessage(String s) { 
            logger.info("收到消息："+s);
        }
    }

##### 5、测试类 #####

@RestController
    public class MsgController { 
        @Value("${rocketmq.topic}")
        private String topic;
    
        @Resource
        private SpringProducer mqSender;
    
        //发送消息
        @GetMapping("/sendMsg")
        public String sendMsg(String msg){ 
            if(!StringUtils.isEmpty(msg)){ 
                mqSender.sendMessage( msg,topic);
                return "ok";
            }
            return "请输入要发送的消息";
        }
    
        //发送事务消息
        @GetMapping("/sendTransactionMsg")
        public String sendTransactionMsg(String msg){ 
            if(!StringUtils.isEmpty(msg)){ 
                mqSender.sendMessageIntransaction( msg,topic);
                return "ok";
            }
            return "请输入要发送的消息";
        }
    }