RocketMQ与Springboot整合实战教程

一、引入依赖
在Spring Boot项目中，要使用RocketMQ，首先需要在pom.xml文件中引入RocketMQ的starter依赖。

org.apache.rocketmq
rocketmq-spring-boot-starter
版本号

二、配置RocketMQ
在Spring Boot的配置文件（application.properties或application.yml）中，需要配置RocketMQ的相关参数，例如NameServer地址、生产者组名、消费者组名等。

例如，在application.properties文件中配置如下：

rocketmq.name-server=127.0.0.1:9876
rocketmq.producer.group=my-producer-group
rocketmq.consumer.group=my-consumer-group

三、发送消息
在Spring Boot项目中，可以使用RocketMQ的Template模式来发送消息。首先创建一个RocketMQ的配置类，并在其中注入MessageConverter和NameServer地址。

@Configuration
public class RocketMQConfig {
@Value(“${rocketmq.name-server}”)
private String nameServer;

@Bean
public Producer producer() throws MQClientException {
    DefaultMQProducer producer = new DefaultMQProducer("my-producer-group");
    producer.setNamesrvAddr(nameServer);
    producer.start();
    return producer;
}

}

然后，在需要发送消息的地方，注入RocketMQ的Template，并使用其send方法发送消息。

@Autowired
private RocketMQTemplate rocketMQTemplate;

public void sendMessage(String topic, String message) {
Message msg = new Message(topic, message);
rocketMQTemplate.send(msg);
}

四、接收消息
在Spring Boot项目中，可以使用RocketMQ的监听器模式来接收消息。首先创建一个消息监听器类，实现MessageListener接口，并实现onMessage方法。

public class MyMessageListener implements MessageListenerConcurrently {
@Override
public ConsumeConcurrentlyStatus onMessage(List msgs) {
for (MessageExt msg : msgs) {
System.out.println(“Received message: “ + new String(msg.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
}

然后，在需要接收消息的地方，注入RocketMQ的配置类和消息监听器。

@Autowired
private RocketMQConfig rocketMQConfig;

@Autowired
private MyMessageListener myMessageListener;

public void startConsumer() throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(“my-consumer-group”);
consumer.setNamesrvAddr(rocketMQConfig.getProducer().getNamesrvAddr());
consumer.registerMessageListener((MessageListenerConcurrently) myMessageListener);
consumer.start();
}

五、异常处理
在处理消息队列时，可能会遇到各种异常情况。为了更好地处理这些异常，可以在消息监听器类中实现try-catch语句来捕获并处理异常。同时，也可以在Spring Boot的配置文件中开启全局异常处理功能，以便对异常进行统一处理。例如：

@ControllerAdvice
public class GlobalExceptionHandler {
@ExceptionHandler(value = Exception.class)
public ResponseEntity