RocketMQ集成SpringBoot

踩了3天的坑，发现使用rocketMQTemplte消费者一直注册不上，发现是springboot版本问题

1.版本选择

经过不严格测试发现
rocketmq-spring-boot-starter 2.3.0 支持 springboot3以上
rocketmq-spring-boot-starter 2.2.x 只能使用3以下，高版本会导致消费者注册不上的问题

2.application.yml

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

3.配置类

@Configuration
public class RocketMqConfig {

    @Value("${rocketmq.name-server}")
    private String nameServer;

    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.producer.send-message-timeout}")
    private Integer sendMsgTimeout;

    @Value("${rocketmq.producer.max-message-size}")
    private Integer maxMessageSize;

    @Value("${rocketmq.producer.retry-times-when-send-failed}")
    private Integer retryTimesWhenSendFailed ;

    @Value("${rocketmq.producer.retry-times-when-send-async-failed}")
    private Integer retryTimesWhenSendAsyncFailed ;

    @Bean
    public RocketMQTemplate rocketMqTemplate(){
        RocketMQTemplate rocketMqTemplate = new RocketMQTemplate();
        rocketMqTemplate.setProducer(defaultMqProducer());
        return rocketMqTemplate;
    }

    @Bean
    public DefaultMQProducer defaultMqProducer() {
        DefaultMQProducer producer = new DefaultMQProducer();
        producer.setNamesrvAddr(this.nameServer);
        producer.setProducerGroup(this.producerGroup);
        producer.setSendMsgTimeout(this.sendMsgTimeout);
        producer.setMaxMessageSize(this.maxMessageSize);
        producer.setRetryTimesWhenSendFailed(this.retryTimesWhenSendFailed);
        producer.setRetryTimesWhenSendAsyncFailed(this.retryTimesWhenSendAsyncFailed);
        return producer;
    }
}

4.生产者

@Component
public class OrderSaveProducer {

    @Autowired
    private RocketMQTemplate rocketMqTemplate;

    public String sendSaveOrder() throws Exception {
        SendResult sendResult =  rocketMqTemplate.syncSend("saveOrderTopic",json);
        return sendResult.getMsgId();
    }
}

5.消费者

@Component
@RocketMQMessageListener(consumerGroup = ("saveOrderConsumer"), topic = ("saveOrderTopic"))
public class OrderSaveConsumer implements RocketMQListener<MessageExt> {
    @Override
    public void onMessage(MessageExt messageExt) {
        try{
            // 获取消息
            byte[] body = messageExt.getBody();
            OrderInfo orderInfo = JSON.parseObject(new String(body), OrderInfo.class);
            System.out.println(orderInfo);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
}