RocketMQ介绍以及常用API使用

一时失言乱红尘 2022-09-03 14:29 223阅读 0赞

### 文章目录 ###

*  RocketMQ介绍以及常用API使用
 *   *  安装
     *  具体介绍
     *  RocketMQ的使用场景
     *   *   *   *  应用解耦
                 *  流量削峰
                 *  数据分发
     *  RocketMQ架构介绍
     *  RocketMQ中的角色术语
     *   *   *   *  1 消息模型（Message Model）
                 *  2 Producer
                 *  3 Consumer
                 *  4 PushConsumer
                 *  5 PullConsumer
                 *  6 ProducerGroup
                 *  7 ConsumerGroup
                 *  8 Broker
                 *  9)广播消费
                 *  10 集群消费
                 *  11)顺序消息
                 *  12 普通顺序消息
                 *  13 严格顺序消息
                 *  14 Message Queue
                 *  15)标签（Tag）
     *  消费模式Push or Pull
     *  相关API使用
     *  启动报错可参考

# RocketMQ介绍以及常用API使用 #

## 安装 ##

可以看这篇 Linux 可以看 [点击][Link 1]

## 具体介绍 ##

可以直接看官方介绍 [https://github.com/apache/rocketmq/tree/master/docs/cn ][https_github.com_apache_rocketmq_tree_master_docs_cn]

## RocketMQ的使用场景 ##

##### 应用解耦 #####

系统的耦合性越高，容错性就越低。以电商应用为例，用户创建订单后，如果耦合调用库存系统、  
物流系统、支付系统，任何一个子系统出了故障或者因为升级等原因暂时不可用，都会造成下单操作异  
常，影响用户使用体验。

##### 流量削峰 #####

应用系统如果遇到系统请求流量的瞬间猛增，有可能会将系统压垮。有了消息队列可以将大量请求  
缓存起来，分散到很长一段时间处理，这样可以大大提到系统的稳定性和用户体验。  
举例:业务系统正常时段的QPS如果是1000，流量最高峰是10000，为了应对流量高峰配置高性能  
的服务器显然不划算，这时可以使用消息队列对峰值流量削峰

##### 数据分发 #####

通过消息队列可以让数据在多个系统之间进行流通。数据的产生方不需要关心谁来使用数据，只需  
要将数据发送到消息队列，数据使用方直接在消息队列中直接获取数据即可

## RocketMQ架构介绍 ##

RocketMQ的角色介绍  
Producer：消息的发送者；举例：发信者  
Consumer：消息接收者；举例：收信者  
Broker：暂存和传输消息；举例：邮局  
NameServer：管理Broker；举例：各个邮局的管理机构  
Topic：区分消息的种类；一个发送者可以发送消息给一个或者多个Topic；一个消息的接收者可以订阅一个或者多个Topic消息  
Message Queue：相当于是Topic的分区；用于并行发送和接收消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]

`NameServer`是一个几乎无状态节点，可集群部署，节点之间无任何信息同步。

`Broker`部署相对复杂，Broker分为Master与Slave，一个Master可以对应多个Slave，但是一个Slave只能对应一个Master，Master与Slave 的对应关系通过指定相同的BrokerName，不同的BrokerId来定义，BrokerId为0表示Master，非0表示Slave。Master也可以部署多个。

每个`Broker`与`NameServer`集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。 注意：当前RocketMQ版本在部署架构上支持一Master多Slave，但只有BrokerId=1的从服务器才会参与消息的读负载。

`Producer`与`NameServer`集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic 服务的Master建立长连接，且定时向Master发送心跳。Producer完全无状态，可集群部署。

`Consumer`与`NameServer`集群中的其中一个节点（随机选择）建立长连接，定期从NameServer获取Topic路由信息，并向提供Topic服务的Master、Slave建立长连接，且定时向Master、Slave发送心跳。Consumer既可以从Master订阅消息，也可以从Slave订阅消  
息，消费者在向Master拉取消息时，Master服务器会根据拉取偏移量与最大偏移量的距离  
（判断是否读老消息，产生读I/O），以及从服务器是否可读等因素建议下一次是从Master还 是Slave拉取。

`执行流程`

1.  启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
2.  Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
3.  收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。 4. Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
4.  Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息

## RocketMQ中的角色术语 ##

##### 1 消息模型（Message Model） #####

RocketMQ主要由 Producer、Broker、Consumer 三部分组成：

Producer 负责生产消息

Consumer 负责消费消息

Broker 负责存储消息。

Broker 在实际部署过程中对应一台服务器，每个Broker 可以存储多个Topic的消息，每个Topic的消息也可以分片存储于不同的 Broker。Message

Queue 用于存储消息的物理地址，每个Topic中的消息地址存储于多个 Message Queue 中。

ConsumerGroup 由多个Consumer 实例构成。

##### 2 Producer #####

消息生产者，负责产生消息，一般由业务系统负责产生消息。

##### 3 Consumer #####

消息消费者，负责消费消息，一般是后台系统负责异步消费。

##### 4 PushConsumer #####

Consumer消费的一种类型，该模式下Broker收到数据后会主动推送给消费端。  
应用通常向Consumer对象注册一个Listener接口，一旦收到消息，Consumer对象立刻回调Listener接口方法。该消费模式一般实时性较高。

##### 5 PullConsumer #####

Consumer消费的一种类型，应用通常主动调用Consumer的拉消息方法从Broker服务器拉消息、主动权由应用控制。一旦获取了批量消息，应用就会启动消费过程。

##### 6 ProducerGroup #####

同一类Producer的集合，这类Producer发送同一类消息且发送逻辑一致。如果发送的是事务消息且原始生产者在发送之后崩溃，则Broker服务器会联系同一生产者组的其他生产者实例以提交或回溯消费。

##### 7 ConsumerGroup #####

同一类Consumer的集合，这类Consumer通常消费同一类消息且消费逻辑一致。消费者组使得在消息消费方面，实现负载均衡和容错的目标变得非常容易。要注意的是，消费者组的消费者实例必须订阅完全相同的Topic。RocketMQ 支持两种消息模式：集群消费（Clustering）和广播消费（Broadcasting）。

##### 8 Broker #####

消息中转角色，负责存储消息，转发消息，一般也称为 Server。在 JMS 规范中称为Provider。

##### 9)广播消费 #####

一条消息被多个 Consumer 消费，即使这些 Consumer 属于同一个 Consumer Group，消息也会被 Consumer Group 中的每个 Consumer 都消费一次，广播消费中的 Consumer Group 概念可以认为在消息划分方面无意义。

在 CORBA Notification 规范中，消费方式都属于广播消费。在 JMS 规范中，相当于 JMS Topic（ publish/subscribe ）模型

##### 10 集群消费 #####

一个 Consumer Group 中的 Consumer 实例平均分摊消费消息。例如某个 Topic 有 9 条消息，其中一个 Consumer Group 有 3 个实例(可能是 3 个进程，或者3台机器)，那举每个实例只消费其中的 3条消息。

##### 11)顺序消息 #####

消费消息的顺序要同发送消息的顺序一致，在RocketMQ 中主要指的是局部顺序，即一类消息为满足顺序性，必须Producer单线程顺序发送，且发送到同一个队列，这样Consumer 就可以按照Producer发送的顺序去消费消息。

##### 12 普通顺序消息 #####

顺序消息的一种，正常情况下可以保证完全的顺序消息，但是一旦发生通信异常，Broker 重启，由于队列总数发生发化，哈希取模后定位的队列会发化，产生短暂的消息顺序不一致。 如果业务能容忍在集群异常情况(如某个Broker 宕机或者重启)下，消息短暂的乱序，使用普通顺序方式比较合适。

##### 13 严格顺序消息 #####

顺序消息的一种，无论正常异常情况都能保证顺序，但是牺牲了分布式 Failover特性，即Broker集群中只要有一台机器不可用，则整个集群都不可用，服务可用性大大降低。 如果服务器部署为同步双写模式，此缺陷可通过备机自动切换为主避免，不过仍然会存在几分钟的服务不可用。(依赖同步双写，主备自动切换，自动切换功能目前还未实现)

目前已知的应用只有数据库 binlog 同步强依赖严格顺序消息，其他应用绝大部分都可以容忍短暂乱序，推荐使用普通的顺序消息。

##### 14 Message Queue #####

在 RocketMQ 中，所有消息队列都是持久化的，长度无限的数据结构，所谓长度无限是指队列中的每个存储单元都是定长，访问其中的存储单元使用Offset来访问，offset 为 java long 类型，64 位，理论上在 100 年内不会溢出，所以认为为是长度无限，另外队列中只保存最近几天的数据，之前的数据会按照过期时间来删除。也可以认为Message Queue是一个长度无限的数组，offset 就是下标。

##### 15)标签（Tag） #####

为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展  
性。

## 消费模式Push or Pull ##

RocketMQ消息订阅有两种模式，一种是Push模式（MQPushConsumer），即MQServer主动向消费端推送；另外一种是Pull模式（MQPullConsumer），即消费端在需要时，主动到MQ Server拉取。  
但在具体实现时，Push和Pull模式本质都是采用消费端主动拉取的方式，即consumer轮询从broker拉取消息。

*  Push模式特点:

好处就是实时性高。不好处在于消费端的处理能力有限，当瞬间推送很多消息给消费端时，容易造成消费端的消息积压，严重时会压垮客户端。

*  Pull模式

好处就是主动权掌握在消费端自己手中，根据自己的处理能力量力而行。缺点就是如何控制Pull的频率。定时间隔太久担心影响时效性，间隔太短担心做太多“无用功”浪费资源。比较折中的办法就是长轮询。

*  Push模式与Pull模式的区别:

Push方式里，consumer把长轮询的动作封装了，并注册MessageListener监听器，取到消息后，唤醒MessageListenerconsumeMessage()来消费，对用户而言，感觉消息是被推送过来的。Pull方式里，取消息的过程需要用户自己主动调用，首先通过打算消费的Topic拿到MessageQueue的集合，遍历MessageQueue集合，然后针对每个MessageQueue批量取消息，一次取完后，记录该队列下一次要取的开始offset，直到取完了，再换另一个MessageQueue。

RocketMQ使用长轮询机制来模拟Push效果，算是兼顾了二者的优点。

## 相关API使用 ##

`注意`  
启动时，必须先启动namesever,然后再启动broker  
第一步启动namesrv

./mqnamesrv

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]

第二步 启动broker

./mqbroker -n localhost:9876

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]

主要演示原生API 生产消息以及推送消息，拉取消息进行消费

导入依赖

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

来新建一个生产者，用来生产消息

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException { 
    
    
            //实例化生产者 指定生产组名称
            DefaultMQProducer producer = new DefaultMQProducer("myproducer_group_01");
    
            //namesrvAddr 地址
            producer.setNamesrvAddr("localhost:9876");
    
    
            //超时时间
            producer.setSendMsgTimeout(1000000);
    
            //初始化
            producer.start();
    
            //可以指定批量发送 也可以单个发送
            Collection<Message> msg = new ArrayList<>();
    
    
            Message message = new Message("top_demo_01","我是消息".getBytes(RemotingHelper.DEFAULT_CHARSET));
            msg.add(message);
    
    
            //发送消息 以及结果
            SendResult send = producer.send(msg);
    
            System.out.println(send);
    
    
            //关闭
            producer.shutdown();
    
        }

可以看到打印后发送返回信息  
![在这里插入图片描述][5b39ea3bc15b47a29f66176cbe7936a0.png]

推消息消费

public static void main(String[] args) throws MQClientException, InterruptedException { 
    
            // 实例化推送消息消费者的对象，同时指定消费组名称
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("myproducer_group_01");
    
            // 指定nameserver的地址
            consumer.setNamesrvAddr("localhost:9876");
    
            // 订阅主题
            consumer.subscribe("top_demo_01", "*");
    
            // 添加消息监听器，一旦有消息推送过来，就进行消费
            consumer.setMessageListener(new MessageListenerConcurrently() { 
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { 
    
                    final MessageQueue messageQueue = context.getMessageQueue();
                    System.out.println(messageQueue);
    
                    for (MessageExt msg : msgs) { 
                        try { 
                            System.out.println(new String(msg.getBody(), "utf-8"));
                        } catch (UnsupportedEncodingException e) { 
                            e.printStackTrace();
                        }
                    }
    
                    // 消息消费成功
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    // 消息消费失败
    // return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            });
    
            // 初始化消费者，之后开始消费消息
            consumer.start();
    
            // 此处只是示例，生产中除非运维关掉，否则不应停掉，长服务
    // Thread.sleep(30_000);
    // // 关闭消费者
    // consumer.shutdown();
        }

拉取消费消费

public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException, UnsupportedEncodingException { 
            // 拉取消息的消费者实例化，同时指定消费组名称
            DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("myproducer_group_01");
            // 设置nameserver的地址
            consumer.setNamesrvAddr("localhost:9876");
    
            // 对消费者进行初始化，然后就可以使用了
            consumer.start();
    
            // 获取指定主题的消息队列集合
            final Set<MessageQueue> messageQueues = consumer.fetchSubscribeMessageQueues("top_demo_01");
    
            // 遍历该主题的各个消息队列，进行消费
            for (MessageQueue messageQueue : messageQueues) { 
                // 第一个参数是MessageQueue对象，代表了当前主题的一个消息队列
                // 第二个参数是一个表达式，对接收的消息按照tag进行过滤
                // 支持"tag1 || tag2 || tag3"或者 "*"类型的写法；null或者"*"表示不对消息进行tag过滤 可以看做是rabbitmq里的路由key
                // 第三个参数是消息的偏移量，从这里开始消费
                // 第四个参数表示每次最多拉取多少条消息
                final PullResult result = consumer.pull(messageQueue, "*", 0, 10);
                // 打印消息队列的信息
                System.out.println("message******queue******" + messageQueue);
                // 获取从指定消息队列中拉取到的消息
                final List<MessageExt> msgFoundList = result.getMsgFoundList();
                if (msgFoundList == null) { 
                    System.out.println("暂无消息");
                    continue;
                }
    
                for (MessageExt messageExt : msgFoundList) { 
                    System.out.println(messageExt);
                    System.out.println(new String(messageExt.getBody(), "utf-8"));
                }
            }
    
            // 关闭消费者
            consumer.shutdown();
        }

可以看到拉取到的消息  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]

RocketMq和RabbitMq一样也支持异步发送消息

public static void main(String[] args) throws MQClientException, UnsupportedEncodingException, RemotingException, InterruptedException { 
            // 实例化生产者，并指定生产组名称
            DefaultMQProducer producer = new DefaultMQProducer("myproducer_group_01");
    
            // 指定nameserver的地址
            producer.setNamesrvAddr("localhost:9876");
    
            // 初始化生产者
            producer.start();
    
            for (int i = 0; i < 100; i++) { 
    
                Message message = new Message(
                        "top_demo_01",
                        ("hello " + i).getBytes("utf-8")
                );
    
                // 消息的异步发送
                producer.send(message, new SendCallback() { 
                    @Override
                    public void onSuccess(SendResult sendResult) { 
                        System.out.println("发送成功:" + sendResult);
                    }
    
                    @Override
                    public void onException(Throwable throwable) { 
                        System.out.println("发送失败：" + throwable.getMessage());
                    }
                });
            }
    
            // 由于是异步发送消息，上面循环结束之后，消息可能还没收到broker的响应
            // 如果不sleep一会儿，没有来得及回调关闭，会发送失败错
            Thread.sleep(10_000);
    
            // 关闭生产者
            producer.shutdown();
        }

简单的实用RocketMq发送和消费消息

RabbitMq是将消息直接发送到交换器上，然后交换器根据指定路由key发送到指定队列  
而RocketMq是先跟NameServer集群中的其中一台建立长连接，从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息，Broker再讲消息发送到队列。

## 启动报错可参考 ##

[https://blog.csdn.net/w8253497062015/article/details/94458893?utm\_medium=distribute.pc\_relevant.none-task-blog-2defaultbaidujs\_title~default-0.control&spm=1001.2101.3001.4242][https_blog.csdn.net_w8253497062015_article_details_94458893_utm_medium_distribute.pc_relevant.none-task-blog-2defaultbaidujs_title_default-0.control_spm_1001.2101.3001.4242]

[https://blog.csdn.net/jiangyu1013/article/details/81478754?utm\_medium=distribute.pc\_relevant\_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control&depth\_1-utm\_source=distribute.pc\_relevant\_t0.none-task-blog-2%7Edefault%7EBlogCommendFromMachineLearnPai2%7Edefault-1.control][https_blog.csdn.net_jiangyu1013_article_details_81478754_utm_medium_distribute.pc_relevant_t0.none-task-blog-2_7Edefault_7EBlogCommendFromMachineLearnPai2_7Edefault-1.control_depth_1-utm_source_distribute.pc_relevant_t0.none-task-blog-2_7Edefault_7EBlogCommendFromMachineLearnPai2_7Edefault-1.control]

[Link 1]: https://blog.csdn.net/weixin_38361347/article/details/119489702
[https_github.com_apache_rocketmq_tree_master_docs_cn]: https://github.com/apache/rocketmq/tree/master/docs/cn
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70]: /images/20220829/960fd842a47542f08c42c9babca5f64e.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 1]: /images/20220829/72889340d60c4027bbfd5b87c9d4fe48.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 2]: /images/20220829/56cd2c9627c546f1af7b24074ad4eea0.png
[5b39ea3bc15b47a29f66176cbe7936a0.png]: /images/20220829/05db6ff193e14eae8aebeffbc89c8db7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zODM2MTM0Nw_size_16_color_FFFFFF_t_70 3]: /images/20220829/53570f327cfe4fd99743aa9853fbd23e.png
[https_blog.csdn.net_w8253497062015_article_details_94458893_utm_medium_distribute.pc_relevant.none-task-blog-2defaultbaidujs_title_default-0.control_spm_1001.2101.3001.4242]: https://blog.csdn.net/w8253497062015/article/details/94458893?utm_medium=distribute.pc_relevant.none-task-blog-2~default~baidujs_title~default-0.control&spm=1001.2101.3001.4242
[https_blog.csdn.net_jiangyu1013_article_details_81478754_utm_medium_distribute.pc_relevant_t0.none-task-blog-2_7Edefault_7EBlogCommendFromMachineLearnPai2_7Edefault-1.control_depth_1-utm_source_distribute.pc_relevant_t0.none-task-blog-2_7Edefault_7EBlogCommendFromMachineLearnPai2_7Edefault-1.control]: https://blog.csdn.net/jiangyu1013/article/details/81478754?utm_medium=distribute.pc_relevant_t0.none-task-blog-2~default~BlogCommendFromMachineLearnPai2~default-1.control&depth_1-utm_source=distribute.pc_relevant_t0.none-task-blog-2~default~BlogCommendFromMachineLearnPai2~default-1.control