终于来了...RocketMQ扫盲篇

原创

ithorizon 8个月前 (09-07) 阅读数 106 #Java

终于来了...RocketMQ扫盲篇

随着互联网技术的飞速发展中，消息中间件在分布式系统中扮演着越来越重要的角色。RocketMQ作为一款优秀的消息中间件，因其高性能、低延迟、高可靠等特性，在众多知名企业中得到了广泛的应用。今天，我们就来为大家科普一下RocketMQ的基础知识。

一、RocketMQ简介

RocketMQ是由阿里巴巴开源的一个分布式消息中间件，它首要用于处理大规模消息的传递。RocketMQ基于高可用、可扩展的设计理念，为用户提供了一个稳定可靠、高效的消息传递服务。

二、核心概念

在了解RocketMQ之前，我们需要先熟悉以下几个核心概念：

1. 消息（Message）

消息是信息传递的基本单位，由一组字节组成。在RocketMQ中，一条消息可以包含以下内容：

主题（Topic）：用于分类消息的逻辑概念，类似于数据库中的表名。

标签（Tag）：用于进一步细分主题的消息，类似于数据库中的字段名。

键（Key）：用于标识消息的唯一性，可用于查询和去重。

2. 主题（Topic）

主题是消息的分类，生产者和消费者通过主题进行消息的发布和订阅。一个主题下可以有多个标签，用于进一步细分消息。

3. 生产者（Producer）

生产者负责创建并发送消息到RocketMQ。生产者可以将消息发送到指定的主题和标签，以便消费者订阅和处理。

4. 消费者（Consumer）

消费者负责订阅并处理消息。消费者可以依主题和标签来过滤接收到的消息，从而只处理感兴趣的消息。

三、基本操作

RocketMQ的基本操作首要包括：发送消息、接收消息、查询消息等。

1. 发送消息

生产者通过以下代码发送消息：


// 创建消息生产者
DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
// 创建消息对象
Message msg = new Message("TopicTest" /* Topic */,
        "TagA" /* Tag */,
        ("Hello RocketMQ ").getBytes() /* Message body */
);
// 发送消息
try {
    producer.send(msg);
} catch (Exception e) {
    e.printStackTrace();
}
// 停止生产者
producer.shutdown();

2. 接收消息

消费者通过以下代码接收消息：


// 创建消息消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupName");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.subscribe("TopicTest", "*");
// 注册消息监听器
consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List msgs, ConsumeConcurrentlyContext context) {
        for (MessageExt msg : msgs) {
            System.out.println(new String(msg.getBody()));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});
// 启动消费者
consumer.start();