RocketMQ与Springboot整合

RocketMQ和Springboot是两个非常流行的开源项目，他们都可以在大规模分布式系统中发挥重要作用。RocketMQ是一个高性能、可靠、可扩展的分布式消息中间件，支持发布/订阅、点对点、异步等消息传递模式。Springboot是一个用于创建微服务的框架，可以快速、方便地搭建基于Spring的应用。

将RocketMQ和Springboot整合起来，可以快速构建一个分布式的消息系统，实现异步消息传递、解耦等功能。下面是一个简单的RocketMQ与Springboot整合的步骤：

1. 引入RocketMQ的依赖

在Springboot的pom.xml文件中引入RocketMQ的依赖：

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client</artifactId>
    <version>4.5.2</version>
</dependency>

2. 配置RocketMQ的连接信息

在Springboot的application.properties文件中配置RocketMQ的连接信息：

rocketmq.namesrvAddr=127.0.0.1:9876
rocketmq.producer.group=myGroup

3. 创建消息生产者

在Springboot中创建一个RocketMQ的消息生产者，用于发送消息：

@Component
public class RocketMQProducer {
 
    @Value("${rocketmq.producer.group}")
    private String producerGroup;
 
    private DefaultMQProducer producer;
 
    @PostConstruct
    public void init() throws MQClientException {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.start();
    }
 
    @PreDestroy
    public void destroy() {
        producer.shutdown();
    }
 
    public void send(String topic, String message) throws MQClientException, RemotingException, InterruptedException, MQBrokerException {
        Message msg = new Message(topic, message.getBytes(StandardCharsets.UTF_8));
        SendResult result = producer.send(msg);
        System.out.printf("Message sent: %s%n", result);
    }
}

4. 创建消息消费者

在Springboot中创建一个RocketMQ的消息消费者，用于接收消息：

@Component
public class RocketMQConsumer {
 
    @Value("${rocketmq.namesrvAddr}")
    private String namesrvAddr;
 
    private DefaultMQPushConsumer consumer;
 
    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer("myGroup");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe("myTopic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("Message received: %s%n", new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }
 
    @PreDestroy
    public void destroy() {
        consumer.shutdown();
    }
}

5. 发送和接收消息

在Springboot中使用RocketMQProducer发送消息：

@Autowired
private RocketMQProducer producer;

public void sendMessage() throws Exception {
    producer.send("myTopic", "Hello, RocketMQ!");
}

在Springboot中使用RocketMQConsumer接收消息：

@Autowired
private RocketMQConsumer consumer;

至此，RocketMQ与Springboot整合完成。通过上述步骤，可以实现在Springboot中使用RocketMQ发送和接收消息的功能。