Springboot 集成 RocketMq（入门）

男娘i 2024-02-20 13:02 93阅读 0赞

## 1.RocketMq安装部署 ##

[Linux 安装 RocketMq-CSDN博客][Linux _ RocketMq-CSDN]

## 2.添加依赖包 ##

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

## 3.配置 ##

rocketmq:
      name-server: 127.0.0.1:9876
      # 生产者
      producer:
        group: group_one
        # 消息发送超时时间
        send-message-timeout: 30000
        # 消息最大长度4M
        max-message-size: 4096
        # 消息发送失败重试次数
        retry-times-when-send-failed: 3
        # 异步消息发送失败重试次数
        retry-times-when-send-async-failed: 2
      # 消费者
      consumer:
        # 每次提取的最大消息数
        pull-batch-size: 5

## 4.生成者代码 ##

@RestController
    @Slf4j
    public class ProducerController {
    
        @Resource
        private RocketMQTemplate rocketMqTemplate;
        @Resource
        private DefaultMQProducer defaultMqProducer;
    
        @GetMapping("/send/msg1/{messageBody}")
        public String sendMsg1(@PathVariable("messageBody") String messageBody) {
            // 发送消息
            rocketMqTemplate.convertAndSend("topic1", messageBody);
            return "OK";
        }
    
        @GetMapping("/send/msg2/{messageBody}")
        public String sendMsg2(@PathVariable("messageBody") String messageBody) {
            // 构建消息对象
            Message message = new Message();
            message.setTopic("topic2");
            message.setTags("boot-mq-tag");
            message.setKeys("boot-mq-key");
            message.setBody(messageBody.getBytes());
            // 发送消息，打印日志
            SendResult sendResult = null;
            try {
                sendResult = defaultMqProducer.send(message);
                log.info("sendMsg2 msgId:{},sendStatus:{}", sendResult.getMsgId(), sendResult.getSendStatus());
            } catch (Exception e) {
                log.info("sendMsg2 error", e);
                return "FAIL";
            }
            return "OK";
        }
    }

## 5.消费者 ##

@Component
    @RocketMQMessageListener(topic = "topic1", consumerGroup = "topic1")
    @Slf4j
    public class MyTopic1Consumer implements RocketMQListener<String> {
    
        @Override
        public void onMessage(String message) {
            // 处理消息的逻辑
            log.info("Received topic1 message: {}", message);
        }
    
    }

@Component
    @RocketMQMessageListener(topic = "topic2", consumerGroup = "topic2")
    @Slf4j
    public class MyTopic2Consumer implements RocketMQListener<Message> {
    
        @Override
        public void onMessage(Message message) {
            // 处理消息的逻辑
            log.info("Received topic2 message: {}", new String(message.getBody()));
        }
    
    }

[  
  
Springboot 集成 RocketMQ(进阶-消息)-CSDN博客][Springboot _ RocketMQ_-_-CSDN]

[Linux _ RocketMq-CSDN]: https://blog.csdn.net/qq_34253002/article/details/125119483?spm=1001.2014.3001.5501
[Springboot _ RocketMQ_-_-CSDN]: https://blog.csdn.net/qq_34253002/article/details/134243344?spm=1001.2014.3001.5502