MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

RocketMQ基础入门与核心概念解析

2023-09-186.3k 阅读

一、RocketMQ 简介

RocketMQ 是一款由阿里巴巴开源的分布式消息中间件，具有低延迟、高并发、高可用以及可伸缩等特点。它最初是为了解决阿里巴巴集团内部海量消息的可靠传递而开发，在 2016 年开源给 Apache 软件基金会，并于 2017 年成为 Apache 顶级项目。

RocketMQ 广泛应用于互联网领域，例如电商系统中的订单处理、库存更新，以及物流系统中的消息通知等场景。它能在分布式架构中有效地解耦系统模块，提高系统的可靠性和性能。

二、RocketMQ 核心概念

2.1 生产者（Producer）

生产者负责产生消息，它是消息的发送方。在 RocketMQ 中，生产者可以向一个或多个主题（Topic）发送消息。生产者实例化后，可以通过配置属性来指定其分组名称（Producer Group）。

代码示例：

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

public class Producer {
    public static void main(String[] args) throws Exception {
        // 创建生产者实例，指定生产者组名称
        DefaultMQProducer producer = new DefaultMQProducer("producer_group_demo");
        // 设置 NameServer 地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();

        for (int i = 0; i < 10; i++) {
            // 创建消息实例，指定主题、标签和消息体
            Message msg = new Message("TopicTest" /* 主题 */,
                    "TagA" /* 标签 */,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* 消息体 */);
            // 发送消息并获取发送结果
            SendResult sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }

        // 关闭生产者
        producer.shutdown();
    }
}

在上述代码中，首先创建了一个 DefaultMQProducer 实例，并指定了生产者组名称。然后设置了 NameServer 地址，启动生产者后，循环发送 10 条消息到 “TopicTest” 主题。

2.2 消费者（Consumer）

消费者负责接收并处理消息，是消息的接收方。消费者同样属于一个消费者分组（Consumer Group），同一分组内的消费者实例对消息的消费是竞争关系，而不同分组的消费者实例对消息的消费是广播关系。

代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

public class Consumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者实例，指定消费者组名称
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group_demo");
        // 设置 NameServer 地址
        consumer.setNamesrvAddr("localhost:9876");
        // 设置消息模式为集群模式
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 订阅主题和标签
        consumer.subscribe("TopicTest", "TagA");

        // 注册消息监听器
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 启动消费者
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

上述代码创建了一个 DefaultMQPushConsumer 实例，指定了消费者组名称，设置了 NameServer 地址和消息模式为集群模式。然后订阅了 “TopicTest” 主题下的 “TagA” 标签，并注册了消息监听器来处理接收到的消息。

2.3 主题（Topic）

主题是消息的逻辑分类，生产者将消息发送到特定的主题，消费者从主题中订阅并接收消息。一个主题可以有多个消费者订阅，也可以有多个生产者向其发送消息。

2.4 标签（Tag）

标签是对主题的进一步细化，它为消息提供了额外的分类方式。生产者在发送消息时可以为消息指定标签，消费者在订阅主题时可以通过指定标签来过滤消息，只接收感兴趣的消息。

2.5 队列（Queue）

队列是物理概念，一个主题可以划分为多个队列。队列的主要作用是提高消息的并行处理能力和负载均衡。生产者发送消息时，会根据一定的算法将消息发送到不同的队列，消费者在消费消息时，也会从不同的队列中拉取消息进行处理。

2.6 名称服务器（NameServer）

NameServer 是 RocketMQ 的路由信息管理中心，它负责存储和管理集群中各个 Broker 的路由信息，包括 Broker 的地址、所负责的主题和队列等信息。生产者和消费者通过 NameServer 来获取 Broker 的地址信息，从而进行消息的发送和接收。NameServer 本身是一个轻量级的、无状态的服务，可以部署多个实例，相互之间独立工作，通过保持数据的一致性来提供可靠的服务。

2.7 代理服务器（Broker）

Broker 是 RocketMQ 的核心组件，负责接收生产者发送的消息、存储消息以及向消费者发送消息。Broker 可以分为 Master Broker 和 Slave Broker，Master Broker 负责处理读写请求，Slave Broker 则用于数据备份和读请求分担。Master 和 Slave 之间通过异步复制的方式保持数据同步，当 Master 出现故障时，Slave 可以提升为 Master 继续提供服务，从而保证系统的高可用性。

三、RocketMQ 消息发送模式

3.1 同步发送

同步发送是指生产者发送消息后，会等待 Broker 返回发送结果，在收到结果之前，生产者线程会被阻塞。这种发送模式可靠性高，适用于对消息发送成功与否有严格要求的场景，例如订单处理消息。

代码示例：

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

public class SyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("sync_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("SyncTopic",
                "TagA",
                "Sync Message Body".getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(msg);
        System.out.printf("%s%n", sendResult);

        producer.shutdown();
    }
}

在上述代码中，producer.send(msg) 方法会阻塞当前线程，直到 Broker 返回发送结果。

3.2 异步发送

异步发送是指生产者发送消息后，不会等待 Broker 的响应，而是继续执行后续代码。当 Broker 返回响应时，会通过回调函数来处理发送结果。这种发送模式适用于对响应时间敏感，且对消息发送可靠性要求相对较低的场景，例如日志记录消息。

代码示例：

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

public class AsyncProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("async_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("AsyncTopic",
                "TagA",
                "Async Message Body".getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.send(msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("Send message success: %s%n", sendResult);
            }

            @Override
            public void onException(Throwable e) {
                System.out.printf("Send message failed: %s%n", e);
            }
        });

        Thread.sleep(1000);
        producer.shutdown();
    }
}

在上述代码中，producer.send(msg, new SendCallback()) 方法在发送消息后立即返回，当 Broker 返回响应时，会调用 SendCallback 中的 onSuccess 或 onException 方法。

3.3 单向发送

单向发送是指生产者只负责发送消息，不关心 Broker 的响应，这种发送模式速度最快，但可靠性最低，适用于对消息可靠性要求极低的场景，例如监控数据上报。

代码示例：

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

public class OnewayProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("oneway_producer_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        Message msg = new Message("OnewayTopic",
                "TagA",
                "Oneway Message Body".getBytes(RemotingHelper.DEFAULT_CHARSET));
        producer.sendOneway(msg);

        producer.shutdown();
    }
}

在上述代码中，producer.sendOneway(msg) 方法发送消息后不会等待任何响应。

四、RocketMQ 消息消费模式

4.1 集群消费

在集群消费模式下，同一消费者组内的多个消费者实例共同消费主题中的消息。每个消费者实例只负责消费主题中一部分队列的消息，从而实现负载均衡。当有新的消费者实例加入或已有消费者实例退出时，RocketMQ 会自动进行负载均衡调整。

代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

public class ClusterConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cluster_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.subscribe("ClusterTopic", "TagA");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Cluster Consumer Started.");
    }
}

上述代码中，通过设置 consumer.setMessageModel(MessageModel.CLUSTERING) 来指定消费模式为集群消费。

4.2 广播消费

在广播消费模式下，同一消费者组内的每个消费者实例都会接收主题中的全部消息，即消息会被广播给消费者组内的所有实例。这种消费模式适用于需要每个消费者都处理相同消息的场景，例如配置更新消息。

代码示例：

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;

import java.util.List;

public class BroadcastConsumer {
    public static void main(String[] args) throws Exception {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("broadcast_consumer_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.setMessageModel(MessageModel.BROADCASTING);
        consumer.subscribe("BroadcastTopic", "TagA");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                          ConsumeConcurrentlyContext context) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Broadcast Consumer Started.");
    }
}

上述代码中，通过设置 consumer.setMessageModel(MessageModel.BROADCASTING) 来指定消费模式为广播消费。

五、RocketMQ 高可用性机制

5.1 Master - Slave 架构

RocketMQ 通过 Master - Slave 架构来保证高可用性。每个 Broker 可以配置为 Master 或 Slave，Master 负责处理读写请求，Slave 则定期从 Master 同步数据。当 Master 出现故障时，Slave 可以被手动或自动提升为 Master，继续提供服务。

5.2 数据同步方式

RocketMQ 支持同步复制和异步复制两种数据同步方式。

  • 同步复制：Master 在接收到生产者发送的消息后，会等待所有 Slave 都成功写入消息后才向生产者返回成功响应。这种方式保证了数据的强一致性，但会降低系统的写入性能。
  • 异步复制：Master 在接收到生产者发送的消息后，会立即向生产者返回成功响应，然后异步将消息复制到 Slave。这种方式提高了写入性能，但在 Master 故障时，可能会丢失少量未同步到 Slave 的消息。

六、RocketMQ 负载均衡

6.1 生产者负载均衡

生产者在发送消息时，会根据消息队列的负载情况，采用轮询、随机等算法将消息发送到不同的队列。默认情况下，RocketMQ 采用轮询算法，以保证消息在各个队列上的均匀分布。

6.2 消费者负载均衡

在集群消费模式下，消费者负载均衡是指将主题中的队列分配给同一消费者组内的不同消费者实例。RocketMQ 采用的负载均衡算法主要有平均分配算法和环形分配算法。

  • 平均分配算法：将队列按照消费者实例的数量进行平均分配，尽量保证每个消费者实例消费的队列数量相同。
  • 环形分配算法：将消费者实例和队列按照一定顺序组成一个环形结构，每个消费者实例从自己开始，依次获取队列进行消费。

七、RocketMQ 应用场景

7.1 异步处理

在一些业务场景中，有些操作不需要立即得到结果，例如订单处理完成后的短信通知、邮件通知等。可以将这些操作封装成消息发送到 RocketMQ，由消费者异步处理，从而提高系统的响应速度。

7.2 系统解耦

在大型分布式系统中，各个模块之间的耦合度较高。通过使用 RocketMQ 作为消息中间件，可以将模块之间的直接调用改为通过消息进行通信，降低模块之间的耦合度，提高系统的可维护性和可扩展性。

7.3 流量削峰

在一些高并发场景中，例如电商的促销活动，瞬间会有大量的请求涌入。RocketMQ 可以作为一个缓冲区，将请求转化为消息进行存储，然后消费者按照一定的速度从队列中消费消息进行处理，从而避免系统因瞬间高并发而崩溃。

7.4 数据分发

在一些需要将数据分发给多个系统的场景中，例如数据采集系统采集到的数据需要分发给数据分析系统、报表系统等。可以将采集到的数据发送到 RocketMQ，不同的系统从 RocketMQ 中订阅相应的主题进行消费，实现数据的分发。

八、RocketMQ 与其他消息队列的比较

8.1 与 Kafka 的比较

  • 性能：Kafka 在高吞吐量场景下表现出色，适合处理海量日志数据等场景；RocketMQ 在低延迟和高可靠性方面表现较好，更适合对消息处理的实时性和可靠性要求较高的场景。
  • 功能特性：Kafka 的设计更侧重于日志存储和流式处理，其分区和副本机制相对简单；RocketMQ 则具备更丰富的功能，如消息顺序性、事务消息等。
  • 应用场景：Kafka 常用于大数据领域的日志收集、数据传输等场景；RocketMQ 则广泛应用于电商、金融等对可靠性和功能特性要求较高的业务场景。

8.2 与 RabbitMQ 的比较

  • 性能：RabbitMQ 基于 Erlang 语言开发，在并发性能上表现良好，但在高吞吐量场景下不如 RocketMQ 和 Kafka。
  • 功能特性：RabbitMQ 支持多种消息协议，如 AMQP、STOMP 等，功能非常灵活，适用于对消息协议兼容性要求较高的场景；RocketMQ 则在分布式架构下的高可用性和扩展性方面表现更优。
  • 应用场景：RabbitMQ 常用于企业级应用集成、微服务架构中的消息通信等场景；RocketMQ 则在互联网分布式系统中应用更为广泛。

九、RocketMQ 部署与运维

9.1 单机部署

单机部署是最简单的部署方式，适用于开发和测试环境。只需要下载 RocketMQ 的安装包，解压后配置好 NameServer 和 Broker 的相关参数，启动 NameServer 和 Broker 即可。

9.2 集群部署

集群部署可以提高 RocketMQ 的性能和可用性。在集群部署中，需要部署多个 NameServer 实例以保证路由信息的可靠性，同时部署多个 Broker 实例，包括 Master 和 Slave，以实现负载均衡和数据备份。

9.3 运维监控

RocketMQ 提供了丰富的运维监控工具，如 RocketMQ Console，可以实时监控集群的运行状态，包括消息的发送和消费情况、Broker 的负载情况等。同时，还可以通过配置日志输出，对系统运行过程中的关键信息进行记录和分析，以便及时发现和解决问题。

十、总结

RocketMQ 作为一款优秀的分布式消息中间件，具备丰富的功能和良好的性能。通过深入理解其核心概念、消息发送和消费模式、高可用性机制、负载均衡等内容，开发人员可以更好地将 RocketMQ 应用于实际项目中，解决分布式系统中的消息传递和异步处理等问题。在实际应用中，需要根据具体的业务场景和需求，合理选择 RocketMQ 的部署方式和配置参数，以充分发挥其优势，提高系统的可靠性和性能。同时，与其他消息队列的比较也有助于开发人员在不同场景下做出更合适的技术选型。在运维方面，通过有效的监控和管理工具，可以确保 RocketMQ 集群的稳定运行。希望本文能为读者提供一个全面深入的 RocketMQ 入门指导，助力大家在后端开发中更好地运用这一强大的工具。