RocketMQ初识

青旅半醒 2022-06-06 13:50 365阅读 0赞

**产品概述**  
  
消息队列（Message Queue，简称 MQ）是阿里巴巴集团中间件技术部自主研发的专业消息中间件。产品基于高可用分布式集群技术，提供消息发布订阅、消息轨迹查询、定时（延时）消息、资源统计、监控报警等一系列消息云服务，是企业级互联网架构的核心产品。MQ 历史超过9年，为分布式应用系统提供异步解耦、削峰填谷的能力，同时具备海量消息堆积、高吞吐、可靠重试等互联网应用所需的特性，是阿里巴巴双11使用的核心产品。  
  
MQ 是阿里云正式商用的产品，目前在阿里云多个地域（Region）提供了高可用消息云服务，单个域内采用多机房部署，可用性极高，即使整个机房都不可用，仍然可以为应用提供消息发布服务，产品稳定性及可用性完全按照阿里巴巴内部标准来实施，无单点。  
  
MQ 目前提供 TCP、HTTP、MQTT 三种协议层面的接入方式，支持 Java、C++ 以及 .NET 不同语言，方便不同编程语言开发的应用快速接入 MQ 消息云服务。用户可以将应用部署在阿里云 ECS、企业自建云，或者嵌入到移动端、物联网设备中与 MQ 建立连接进行消息收发，同时本地开发者也可以通过公网接入 MQ 服务进行消息收发。  
  
**产品功能**  
  
MQ 提供了多种协议和开发语言的接入方式以及多维度的管理工具，同时针对不同的应用场景提供了一系列的特色功能。  
  
MQ 产品功能和特性概览  
  
**多协议接入**  
  
    支持 HTTP 协议：支持 RESTful 风格 HTTP 协议完成收发消息,可以解决跨语言使用 MQ 问题。  
    支持 MQTT 协议：支持主动推送模型，多级 Topic 模型支持一次触达 1000万+ 终端，可广泛应用于物联网和社交即时通信场景。  
    支持 TCP 协议：区别于 HTTP 简单的接入方式，提供更为专业、可靠、稳定的 TCP 协议的 SDK 接入。  
  
**管理工具**  
  
    Web 控制台，支持 Topic 管理、发布管理、订阅管理、消息查询、消息轨迹、资源报表以及监控报警管理。  
    Open API，提供 API 允许用户将 MQ 管理工具集成到自己的控制台。  
    mqadmin 命令集，专有云输出提供一套丰富的管理命令集，以命令方式对 MQ 服务进行管理。  
  
**特色功能**  
  
    事务消息，实现类似 X/Open XA 的分布事务功能，以达到事务最终一致性状态。  
    定时（延时）消息，允许消息生产者指定消息进行定时（延时）投递，最长支持40天。  
    大消息，目前默认支持最大 256KB 消息，华北2 地域支持最大 4MB 消息。  
    消息轨迹，通过消息轨迹，用户能清晰定位消息从发布者发出，经由 MQ 服务端，投递给消息订阅者的完整链路，方便定位排查问题。  
    广播消息，允许一个 Consumer ID 所标识的所有 Consumer 都会各自消费某条消息一次。  
    顺序消息，允许消息消费者按照消息发送的顺序对消息进行消费。  
    重置消费进度，根据时间重置消费进度，允许用户进行消息回溯或者丢弃堆积消息。  
  
**专有云部署**  
  
    专家定制，提供技术方案设计；专家现场技术支持与培训。  
    灵活部署，支持专有云独立部署，同时支持混合云架构。  
    运维管控，专有云支持 mqadmin 命令集、Open API 运维管理工具，方便管控平台集成以及统一运维。  
  
**产品优势**  
  
**专业**  
  
    消息领域业内专业的消息中间件，产品历史超过 9 年，消息保证不丢，技术体系丰富成熟。  
    阿里内部产品名 MetaQ、Notify；开源社区产品名为 RocketMQ；产品多次在国内外获奖。  
    阿里内部 1000+ 核心应用使用，每天流转几千亿条消息，经过双11交易、商品等核心链路真实场景的验证，稳定可靠。  
  
**高可靠**  
  
    一份消息多份落盘存储，经过严格的断电测试，消息依然保证不丢失。  
    允许海量消息堆积，单个 Topic 可堆积 100亿+条消息，系统高流量压力下依然可靠。  
    默认消息持久化存储 3 天，支持重置消费位点消费3天之内任何时间点的消息。  
  
**高性能**  
  
    同一网络内，消息传输网络时延在 10 毫秒之内，性能测试下，网卡可被打满。  
    默认单 Topic 发送消息上限为每秒 5000 条，最高可申请扩展至 10W 以上。  
    默认单条消息大小最大支持 256KB，华北2 地域支持 4MB 大消息。  
  
**多协议接入**  
  
    支持 HTTP 协议：支持 RESTful 风格 HTTP 协议完成收发消息，可以解决跨语言使用 MQ 问题。  
    支持 MQTT 协议：支持主动推送模型，多级 Topic 模型支持一次触达1000万+ 终端，可广泛应用于物联网和社交即时通信场景。  
    支持 TCP 协议：区别于 HTTP 简单的接入方式，提供更为专业、可靠、稳定的 TCP 协议的 SDK 接入。  
  
**独立部署**  
  
    支持专有云独立输出，支持物理机和虚拟机，仅几台机器便可搭建完整消息云服务。  
    专有云配套 mqadmin 命令集和管理类 Open API，方便运维人员实时监控系统状态。  
    支持混合云架构，允许用户通过专线的方式接入服务。  
  
**应用场景**  
  
MQ 可应用在多个领域，包括异步通信解耦、企业解决方案、金融支付、电信、电子商务、快递物流、广告营销、社交、即时通信、手游、视频、物联网、车联网等。  
  
MQ 可以应用但不局限于以下业务场景：  
  
    一对多，多对多异步解耦，基于发布订阅模型，对分布式应用进行异步解耦，增加应用的水平扩展能力。  
    削峰填谷，大促等流量洪流突然来袭时，MQ 可以缓冲突发流量，避免下游订阅系统因突发流量崩溃。  
    日志监控，作为重要日志的监控通信管道，将应用日志监控对系统性能影响降到最低。  
    消息推送，为社交应用和物联网应用提供点对点推送，一对多广播式推送的能力。  
    金融报文，发送金融报文，实现金融准实时的报文传输，可靠安全。  
    电信信令，将电信信令封装成消息，传递到各个控制终端，实现准实时控制和信息传递。  
  
**应用行业**  
![Center][]  
  
**专业术语**  
  
**Message Queue**  
  
消息队列，阿里云商用的专业消息中间件，是企业级互联网架构的核心产品，提供基于高可用分布式集群技术搭建的消息发布订阅、轨迹查询、资源统计、定时（延时）、监控报警等一系列消息云服务。  
  
**Message**  
  
消息，消息队列中信息传递的载体。  
  
**Message ID**  
  
消息的全局唯一标识，由 MQ 系统自动生成，唯一标识某条消息。  
  
**Message Key**  
  
消息的业务标识，由消息生产者（Producer）设置，唯一标识某个业务逻辑。  
  
**Topic**  
  
消息主题，一级消息类型，通过 Topic 对消息进行分类。  
  
**Tag**  
  
消息标签，二级消息类型，用来进一步区分某个 Topic 下的消息分类。  
  
**Producer**  
  
消息生产者，也称为消息发布者，负责生产并发送消息。  
  
**Producer ID**  
  
一类 Producer 的标识，这类 Producer 通常生产并发送一类消息，且发送逻辑一致。  
  
**Producer 实例**  
  
Producer 的一个对象实例，不同的 Producer 实例可以运行在不同进程内或者不同机器上。Producer 实例线程安全，可在同一进程内多线程之间共享。  
  
**Consumer**  
  
消息消费者，也称为消息订阅者，负责接收并消费消息。  
  
**Consumer ID**  
  
一类 Consumer 的标识，这类 Consumer 通常接收并消费一类消息，且消费逻辑一致。  
  
**Consumer 实例**  
  
Consumer 的一个对象实例，不同的 Consumer 实例可以运行在不同进程内或者不同机器上。一个 Consumer 实例内配置线程池消费消息。  
  
**集群消费**  
  
一个 Consumer ID 所标识的所有 Consumer 平均分摊消费消息。例如某个 Topic 有 9 条消息，一个 Consumer ID 有 3 个 Consumer 实例，那么在集群消费模式下每个实例平均分摊，只消费其中的 3 条消息。  
  
**广播消费**  
  
一个 Consumer ID 所标识的所有 Consumer 都会各自消费某条消息一次。例如某个 Topic 有 9 条消息，一个 Consumer ID 有 3 个 Consumer 实例，那么在广播消费模式下每个实例都会各自消费 9 条消息。  
  
**定时消息**  
  
Producer 将消息发送到 MQ 服务端，但并不期望这条消息立马投递，而是推迟到在当前时间点之后的某一个时间投递到 Consumer 进行消费，该消息即定时消息。  
  
**延时消息**  
  
Producer 将消息发送到 MQ 服务端，但并不期望这条消息立马投递，而是延迟一定时间后才投递到 Consumer 进行消费，该消息即延时消息。  
  
**事务消息**  
  
MQ 提供类似 X/Open XA 的分布事务功能，通过 MQ 事务消息能达到分布式事务的最终一致。  
  
**顺序消息**  
  
MQ 提供的一种按照顺序进行发布和消费的消息类型, 分为全局顺序消息和分区顺序消息。  
  
**顺序发布**  
  
对于指定的一个 Topic，客户端将按照一定的先后顺序进行发送消息。  
  
**顺序消费**  
  
对于指定的一个 Topic，按照一定的先后顺序进行接收消息，即先发送的消息一定会先被客户端接收到。  
  
**全局顺序消息**  
  
对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。  
  
**分区顺序消息**  
  
对于指定的一个 Topic，所有消息根据 sharding key 进行区块分区。同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。Sharding key 是顺序消息中用来区分不同分区的关键字段，和普通消息的 key 是完全不同的概念。  
  
**消息堆积**  
  
Producer 已经将消息发送到 MQ 服务端，但由于 Consumer 消费能力有限，未能在短时间内将所有消息正确消费掉，此时在 MQ 服务端保存着未被消费的消息，该状态即消息堆积。  
  
**消息过滤**  
  
订阅者可以根据消息标签（Tag）对消息进行过滤，确保订阅者最终只接收被过滤后的消息类型。消息过滤在 MQ 服务端完成。  
  
**消息轨迹**  
  
在一条消息从发布者发出到订阅者消费处理过程中，由各个相关节点的时间、地点等数据汇聚而成的完整链路信息。通过消息轨迹，用户能清晰定位消息从发布者发出，经由 MQ 服务端，投递给消息订阅者的完整链路，方便定位排查问题。  
  
**重置消费位点**  
  
以时间轴为坐标，在消息持久化存储的时间范围内（默认3天），重新设置消息订阅者对其订阅 Topic 的消费进度，设置完成后订阅者将接收设定时间点之后由消息发布者发送到 MQ 服务端的消息。  
  
**中继**  
  
中继服务（Relay）是由消息队列(MQ)提供的服务发布与订阅组件，该组件的主要作用是在不同的网络环境下实现服务之间互联互通的能力。  
  
**总结**

![Center 1][]

**产品特性**  
  
RcoketMQ 是一款低延迟、高可靠、可伸缩、易于使用的消息中间件。具有以下特性：  
  
    1、支持发布/订阅（Pub/Sub）和点对点（P2P）消息模型  
    2、在一个队列中可靠的先进先出（FIFO）和严格的顺序传递  
    3、支持拉（pull）和推（push）两种消息模式  
    4、单一队列百万消息的堆积能力  
    5、支持多种消息协议，如 JMS、MQTT 等  
    6、分布式高可用的部署架构,满足至少一次消息传递语义  
    7、提供 docker 镜像用于隔离测试和云集群部署  
    8、提供配置、指标和监控等功能丰富的 Dashboard  
  
  
**RocketMQ 架构**  
![RocketMQ 架构][RocketMQ]  
  
由这张图可以看到有四个集群，分别是 NameServer 集群、Broker 集群、Producer 集群和 Consumer 集群：  
  
    1、NameServer: 提供轻量级的服务发现和路由。 每个 NameServer 记录完整的路由信息，提供等效的读写服务，并支持快速存储扩展。  
    2、Broker: 通过提供轻量级的 Topic 和 Queue 机制来处理消息存储,同时支持推（push）和拉（pull）模式以及主从结构的容错机制。  
    3、Producer：生产者，产生消息的实例，拥有相同 Producer Group 的 Producer 组成一个集群。  
    4、Consumer：消费者，接收消息进行消费的实例，拥有相同 Consumer Group 的Consumer 组成一个集群。  
  
简单说明一下图中箭头含义，从 Broker 开始，Broker Master1 和 Broker Slave1 是主从结构，它们之间会进行数据同步，即 Date Sync。同时每个 Broker 与 NameServer 集群中的所有节点建立长连接，定时注册 Topic 信息到所有 NameServer 中。

Producer 与 NameServer 集群中的其中一个节点（随机选择）建立长连接，定期从 NameServer 获取 Topic 路由信息，并向提供 Topic 服务的 Broker Master 建立长连接，且定时向 Broker 发送心跳。Producer 只能将消息发送到 Broker master，但是 Consumer 则不一样，它同时和提供 Topic 服务的 Master 和 Slave建立长连接，既可以从 Broker Master 订阅消息，也可以从 Broker Slave 订阅消息。

**RocketMQ 集群部署模式  
  
    1、单 master 模式**

也就是只有一个 master 节点，称不上是集群，一旦这个 master 节点宕机，那么整个服务就不可用，适合个人学习使用。

**2、多 master 模式**

多个 master 节点组成集群，单个 master 节点宕机或者重启对应用没有影响。  
  
    优点：所有模式中性能最高  
  
    缺点：单个 master 节点宕机期间，未被消费的消息在节点恢复之前不可用，消息的实时性就受到影响。

注意：使用同步刷盘可以保证消息不丢失，同时 Topic 相对应的 queue 应该分布在集群中各个节点，而不是只在某各节点上，否则，该节点宕机会对订阅该 topic 的应用造成影响。

**    3、多 master 多 slave 异步复制模式**  
  
    在多 master 模式的基础上，每个 master 节点都有至少一个对应的 slave。master  
  
    节点可读可写，但是 slave 只能读不能写，类似于 mysql 的主备模式。  
  
    优点： 在 master 宕机时，消费者可以从 slave 读取消息，消息的实时性不会受影响，性能几乎和多 master 一样。

缺点：使用异步复制的同步方式有可能会有消息丢失的问题。

**4、多 master 多 slave 同步双写模式**  
  
    同多 master 多 slave 异步复制模式类似，区别在于 master 和 slave 之间的数据同步方式。  
  
    优点：同步双写的同步模式能保证数据不丢失。  
  
    缺点：发送单个消息 RT 会略长，性能相比异步复制低10%左右。  
  
    刷盘策略：同步刷盘和异步刷盘（指的是节点自身数据是同步还是异步存储）  
  
    同步方式：同步双写和异步复制（指的一组 master 和 slave 之间数据的同步）  
  
    注意：要保证数据可靠，需采用同步刷盘和同步双写的方式，但性能会较其他方式低。  
  
**RocketMQ 单主部署**  
  
鉴于是快速入门，我选择的是第一种单 master 的部署模式。先说明一下我的安装环境：  
  
    1、Centos 7.2  
    2、jdk 1.8  
    3、Maven 3.2.x  
    4、Git  
  
这里 git 可用可不用，主要是用来直接下载 github 上的源码。也可以选择自己到  
  
github 上下载，然后上传到服务器上。以git操作为示例。  
  
  **  ** clone 源码并用 maven 编译

> git clone https://github.com/alibaba/RocketMQ.git /opt/RocketMQ
    > cd /opt/RocketMQ && mvn -Dmaven.test.skip=true clean package install assembly:assembly -U
    > cd target/alibaba-rocketmq-broker/alibaba-rocketmq

此处可能遇到的问题   
一、执行"git clone https://github.com/alibaba/RocketMQ.git /home/inspkgs/RocketMQ"时出现以下提示：

fatal: unable to access 'https://github.com/alibaba/RocketMQ.git/': Could not resolve host: github.com; Unknown error

解决办法：一般是由于网络原因造成的，执行以下命令

> ping github.com

确定可以 ping 通之后，再重新执行 git clone 命令。  
二、执行"mvn -Dmaven.test.skip=true clean package install assembly:assembly -U"编译时，可能出现下载相关jar很慢的情况。  
这也是由于默认 maven 中央仓库在国外的原因，可以根据需要在 /home/maven/conf/setting.xml 中的 <mirrors></mirrors> 添加以下内容后重新编译：

<mirror>
        <id>aliyun</id>
        <mirrorOf>central</mirrorOf>
        <name>aliyun maven</name>
        <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
    </mirror>

启动 Name Server

> nohup sh /opt/RocketMQ/bin/mqnamesrv &
    
    //执行 jps 查看进程
    > jps
    25913 NamesrvStartup
    //查看日志确保服务已正常启动
    > tail -f ~/logs/rocketmqlogs/namesrv.log
    The Name Server boot success...

启动 broker

> nohup sh /opt/RocketMQ/bin/mqbroker -n localhost:9876 &
    //执行 jps 查看进程
    > jps
    25954 BrokerStartup
    //查看日志确保服务已正常启动
    > tail -f ~/logs/rocketmqlogs/broker.log
    The broker[broker-a, 10.1.54.121:10911] boot success...

1、发送和接收消息发送/接收消息之前,我们需要告诉客户端 NameServer 地址。RocketMQ 提供了多种方式来实现这一目标。为简单起见,我们使用环境变量NAMESRV\_ADDR。

> export NAMESRV_ADDR=localhost:9876
    > sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Producer
    SendResult [sendStatus=SEND_OK, msgId= ...
    > sh /opt/RocketMQ/bin/tools.sh com.alibaba.rocketmq.example.quickstart.Consumer
    ConsumeMessageThread_%d Receive New Messages: [MessageExt...

关闭服务

> sh /opt/RocketMQ/bin/mqshutdown broker
    The mqbroker(36695) is running...
    Send shutdown request to mqbroker(36695) OK
    > sh /opt/RocketMQ/bin/mqshutdown namesrv
    The mqnamesrv(36664) is running...
    Send shutdown request to mqnamesrv(36664) OK

**生产者、消费者 Demo  
  
    生产者**

public class Producer {
        public static void main(String[] args) throws MQClientException, InterruptedException {
    
            //声明并初始化一个producer
            //需要一个producer group名字作为构造方法的参数，这里为producer1
            DefaultMQProducer producer = new DefaultMQProducer("producer1");
            
            //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
            //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
            producer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
            
            //调用start()方法启动一个producer实例
            producer.start();
    
            //发送10条消息到Topic为TopicTest，tag为TagA，消息内容为“Hello RocketMQ”拼接上i的值
            for (int i = 0; i < 10; i++) {
                try {
                    Message msg = new Message("TopicTest",// topic
                            "TagA",// tag
                            ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)// body
                    );
                    
                    //调用producer的send()方法发送消息
                    //这里调用的是同步的方式，所以会有返回结果
                    SendResult sendResult = producer.send(msg);
                    
                    //打印返回结果，可以看到消息发送的状态以及一些相关信息
                    System.out.println(sendResult);
                } catch (Exception e) {
                    e.printStackTrace();
                    Thread.sleep(1000);
                }
            }
    
            //发送完消息之后，调用shutdown()方法关闭producer
            producer.shutdown();
        }
    }

**消费者**

public class Consumer {
    
        public static void main(String[] args) throws InterruptedException, MQClientException {
        
            //声明并初始化一个consumer
            //需要一个consumer group名字作为构造方法的参数，这里为consumer1
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer1");
    
            //同样也要设置NameServer地址
            consumer.setNamesrvAddr("10.1.54.121:9876;10.1.54.122:9876");
    
            //这里设置的是一个consumer的消费策略
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
    
            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe("TopicTest", "*");
    
            //设置一个Listener，主要进行消息的逻辑处理
            consumer.registerMessageListener(new MessageListenerConcurrently() {
    
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
    
                    System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                    
                    //返回消费状态
                    //CONSUME_SUCCESS 消费成功
                    //RECONSUME_LATER 消费失败，需要稍后重新消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
    
            //调用start()方法启动consumer
            consumer.start();
    
            System.out.println("Consumer Started.");
        }
    }

[Center]: /images/20220606/8cbbcdcb0a2b4472adb25f2ded3adef8.png
[Center 1]: /images/20220606/42a17b2427b646af89a48effcac0ad4e.png
[RocketMQ]: /images/20220606/12ff584bc30e42b88d0eb2f4cbadb055.png