springboot 3.x 整合 RocketMQ 5.x的详细过程

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-v5-client-spring-boot-starter</artifactId>
    <version>2.3.1</version>
</dependency>

@Configuration
@EnableConfigurationProperties(RocketMQProperties.class)
@Import({MessageConverterConfiguration.class, ListenerContainerConfiguration.class, ExtTemplateResetConfiguration.class,
        ExtConsumerResetConfiguration.class, RocketMQTransactionConfiguration.class, RocketMQListenerConfiguration.class})
@AutoConfigureAfter({MessageConverterConfiguration.class})
@AutoConfigureBefore({RocketMQTransactionConfiguration.class})
public class RocketMQAutoConfiguration implements ApplicationContextAware {
  	// ... 省略
    @Bean(PRODUCER_BUILDER_BEAN_NAME)
    @ConditionalOnMissingBean(ProducerBuilderImpl.class)
    @ConditionalOnProperty(prefix = "rocketmq", value = {"producer.endpoints"})
    public ProducerBuilder producerBuilder(RocketMQProperties rocketMQProperties) {
      // ... 省略
    }
    @Bean(SIMPLE_CONSUMER_BUILDER_BEAN_NAME)
    @ConditionalOnMissingBean(SimpleConsumerBuilder.class)
    @ConditionalOnProperty(prefix = "rocketmq", value = {"simple-consumer.endpoints"})
    public SimpleConsumerBuilder simpleConsumerBuilder(RocketMQProperties rocketMQProperties) {
       // ... 省略
    }
    @Bean(destroyMethod = "destroy")
    @Conditional(ProducerOrConsumerPropertyCondition.class)
    @ConditionalOnMissingBean(name = ROCKETMQ_TEMPLATE_DEFAULT_GLOBAL_NAME)
    public RocketMQClientTemplate rocketMQClientTemplate(RocketMQMessageConverter rocketMQMessageConverter) {
       // ... 省略
    }
}

rocketmq:
  producer:
    endpoints: localhost:8081
  push-consumer:
  	endpoints: localhost:8081

@Service
public class MyService {
  @Autowired
  private RocketMQClientTemplate rocketMQClientTemplate;
  public void sendMessage() {
      byte[] bytes = "这是一个字符串".getBytes(StandardCharsets.UTF_8);
      Message<byte[]> message = MessageBuilder.withPayload(bytes).build();
      rocketMQClientTemplate.send("MyTopic", message);
  }
}

@Service
@RocketMQMessageListener(consumerGroup = "MyTopic-service", topic = "MyTopic", tag = "*")
public class MyService implements RocketMQListener {
    @Override
    public ConsumeResult consume(MessageView messageView) {
        // 从 MessageView 中获取 ByteBuffer
        ByteBuffer byteBuffer = messageView.getBody();
        // 转换 ByteBuffer 为字节数组
        byte[] body = new byte[byteBuffer.remaining()];
        byteBuffer.get(body);
        // 处理字节数组，例如转换为字符串
        String messageBody = new String(body, StandardCharsets.UTF_8);
        System.out.println("消费消息内容：" + messageBody);
     	 	return ConsumeResult.SUCCESS;
    }
}

#### 启动namesrv
$ nohup sh bin/mqnamesrv &
#### 验证namesrv是否启动成功
$ tail -f ~/logs/rocketmqlogs/namesrv.log
The Name Server boot success...

#### 先启动broker
$ nohup sh bin/mqbroker -n localhost:9876 --enable-proxy &
#### 验证broker是否启动成功, 比如, broker的ip是192.168.1.2 然后名字是broker-a
$ tail -f ~/logs/rocketmqlogs/proxy.log 
The broker[broker-a,192.169.1.2:10911] boot success...