RocketMQ实现一个随缘分BUG小功能

后端

2023-11-02 13:59:04

RocketMQ 是阿里巴巴开源的消息中间件之一，用于构建大规模分布式系统。它以高性能、高可靠性和易于维护性著称。消息队列是一种异步通信机制，在多个组件间传递信息时减少系统的耦合度。

RocketMQ基础概念

生产者（Producer）：负责发送消息到指定主题。
消费者（Consumer）：监听并处理特定主题下的消息。
Topic：消息发布和订阅的逻辑单元，可以理解为分类。
Broker：RocketMQ的消息服务器，用于存储消息、转发给消费者。

实现一个随机数功能

下面将展示如何在RocketMQ中实现一个简单的功能——随机生成数字并将其发送到消息队列。这个例子虽然简单，但能帮助开发者更深入理解RocketMQ的工作原理以及如何使用Java开发相关组件。

准备工作

确保已安装Java环境和Maven工具，并且引入了RocketMQ的相关依赖。在pom.xml中添加以下依赖：

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

实现步骤

步骤1：创建生产者发送随机数消息

首先，我们需要一个Java程序作为生产者，负责生成随机数字并将其发送至指定的Topic。

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;

public class RandomNumberSender {
    public static void main(String[] args) throws Exception {
        // 实例化生产者组名，这里使用 "RandomNumberGroup"
        DefaultMQProducer producer = new DefaultMQProducer("RandomNumberGroup");
        // 启动生产者实例
        producer.start();
        
        int randomNumber = (int)(Math.random() * 100);
        String randomMessage = Integer.toString(randomNumber);

        Message msg = new Message(
                "RandomTopic", // Topic名，发送消息需要指定一个Topic名字。
                randomMessage.getBytes()); // 消息体

        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        // 如果不再发送消息，则需关闭生产者以释放资源
        producer.shutdown();
    }
}

步骤2：创建消费者监听随机数消息

接下来，构建一个Java程序作为消费者，它将订阅RandomTopic并处理接收到的消息。

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

public class RandomNumberReceiver {
    public static void main(String[] args) throws Exception {
        // 实例化消费者，设置订阅的生产者组名为 "RandomNumberGroup"
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("RandomNumberGroup");
        
        // 订阅一个或多个Topic，这里是"RandomTopic"
        consumer.subscribe("RandomTopic", "*");
        
        // 设置消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    System.out.printf("Receive message: %s%n", new String(msg.getBody()));
                }
                // 返回消费结果为成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者实例
        consumer.start();
    }
}