博客

浅谈Springboot整合RocketMQ使用心得

  • Post category:http

浅谈Springboot整合RocketMQ使用心得

RocketMQ是阿里巴巴开源的分布式消息中间件，具有高可靠、高吞吐量、高可用性等点。Springboot是一种快速开发框架，可以帮助我们快速构建应用程序。本文将介绍如何使用Springboot整合RocketMQ，以及使用心得。

1. 添加RocketMQ依赖

首先，我们需要在pom.xml文件中添加RocketMQ依赖。在这里，我们使用的是RocketMQ的官方Java客户端。

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

2. 配置RocketMQ

接下来，我们需要在application.properties文件中配置RocketMQ。在这里，我们需要配置RocketMQ的nameserver地址producer的group。

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

3. 创建Producer

现在，我们可以创建一个RocketMQ的Producer。在这里，我们使用的是Rocket的官方Java客户端。

@Service
public class RocketMQProducer {

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

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

    private DefaultMQProducer producer;

    @PostConstruct
    public void init() throws MQClientException {
        producer = new DefaultMQProducer(producerGroup);
        producer.setNamesrvAddr(nameServer);
        producer.start();
    }

    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.println("Send Result: " + result);
    }

    @PreDestroy
    public void destroy() {
        producer.shutdown();
    }
}

在这个示例中，我们创建了一个名为RocketMQProducer的类，它使用了@PostConstruct和@PreDestroy注解。@PostConstruct注解表示在类初始化时执行的方法，@PreDestroy注解表示在类销毁时执行的方法。在init()方法中，我们创建了一个DefaultMQProducer对象，并设置了nameserver地址和producer的group。在send()方法中，我们创建了一个Message对象，并使用producer发送了消息。在destroy()方法中，我们关闭了producer。

4. 创建Consumer

现在，我们可以创建一个RocketMQ的Consumer。在这里，我们使用的是RocketMQ的官方Java客户端。

@Service
public class RocketMQConsumer {

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

    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;

    private DefaultMQPushConsumer consumer;

    @PostConstruct
    public void init() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(nameServer);
        consumer.subscribe("my-topic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.println("Received Message: " + new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        consumer.start();
    }

    @PreDestroy
    public void destroy() {
        consumer.shutdown();
    }
}

在这个示例中，我们创建了一个名为RocketMQConsumer的类，它使用了@PostConstruct和@PreDestroy注解。在init()方法中，我们创建了一个DefaultMQPushConsumer对象，并设置了nameserver地址和consumer的group。我们还订阅了一个名为“my-topic”的主题，并使用registerMessageListener()方法注册了一个消息监听器。在消息监听器中，我们打印了接收到的消息。在destroy()方法中，我们关闭了consumer。

示例1：发送消息

假设我们要发送一条消息到名为“my-topic”的主题。

解决方案：

我们可以使用RocketMQProducer类的send()方法发送消息。正确的代码如下：

@Autowired
private RocketMQProducer producer;

@GetMapping("/send")
public String send() throws Exception {
    producer.send("my-topic", "Hello, RocketMQ!");
    return "Message Sent!";
}

在这个示例中，我们使用@Autowired注解注入了RocketMQProducer类，并在send()方法中使用producer发送了一条消息。

示例2：接收消息

假设我们要接收名为“my-topic”的主题中的消息。

解决方案：

我们可以使用RocketMQConsumer类来接收消息。正确的代码如下：

@Autowired
private RocketMQConsumer consumer;

在这个示例中，我们使用@Autowired注解注入了RocketMQConsumer类。

总结

RocketMQ是一种高可靠、高吞吐量、高可用性的分布式消息中间件。Springboot是一种快速开发框架，可以帮助快速构建应用程序。在本文中，我们介绍了如何使用Springboot整合RocketMQ，并提供了两个示例说明。在实际使用中，我们应该根据具体情况选择合适的解决方案，以确保能够正常使用RocketMQ。

Tags: