RocketMQ从认知到实现

标签：Queue 实现认知 Broker Topic 消息 Consumer RocketMQ

一.RocketMQ的认识

1.1、RocketMQ是什么

RocketMQ是一个统一消息引擎、轻量级数据处理平台。
RocketMQ是⼀款阿⾥巴巴开源的消息中间件，双十一承载了万亿级消息的流转，2016年11⽉，阿⾥巴巴向 Apache 软件基⾦会捐赠 RocketMQ，成为 Apache 孵化项⽬，2017 年 9 ⽉，Apache 宣布 RocketMQ孵化成为 Apache 顶级项⽬（TLP ）成为国内⾸个互联⽹中间件在 Apache 上的顶级项⽬。

1.2、RocketMQ的特征

支持集群模型、负载均衡、水平扩展能力
亿级别消息堆积能力
采用零拷贝的原理，顺序写盘，随机读 -nio,零拷贝
不经过内存的互相缓存区拷贝，直接通过源文件到目标文件的复制
底层通信框架采用Netty NIO - nio通讯模型—Netty -redis-kafaka/rocketmq
NameServer代替Zookeeper，实现服务寻址和服务协调
消息失败重试机制、消息可查询
强调集群无单点，可扩展，任意一点高可用，水平可扩展
经过多次双十一的考验

二.RocketMQ的原理

2.1、RokcetMQ架构

官方文档：https://github.com/apache/rocketmq/blob/master/docs/cn/RocketMQ_Example.md
基础框架：RocketMQ架构上主要分为四部分，如图所示

在这里插入图片描述

Producer
- 消息发布的角色，支持分布式集群方式部署。Producer通过nameserver的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟
Consumer
- 消息消费的角色，支持分布式集群方式部署。支持以push推，pull拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制，可以满足大多数用户的需求。
Broker
- Broker主要负责消息的存储、投递和查询以及服务高可用保证。
NameServer
- NameServer是一个Broker与Topic路由的注册中心支持Broker的动态注册与发现主要包括两个功能
  - NameServer接受Broker集群的注册信息并且保存下来作为路由信息的基本数据。然后提供心跳检测机制，检查Broker是否还存活。
- 路由信息管理
  - 每个NameServer将保存关于Broker集群的整个路由信息和用于客户端查询的队列信息。然后Producer和Conumser通过NameServer就可以知道整个Broker集群的路由信息，从而进行消息的投递和消费

三.RocketMQ入门

3.1 导入依赖

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

3.2 设置生产者

创建producer组
设置NameServer地址
startr生产者
发送消息获取结果
结束producer

public class SyncProducer {
	public static void main(String[] args) throws Exception {
    	// 实例化消息生产者Producer
        DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
    	// 设置NameServer的地址
    	producer.setNamesrvAddr("localhost:9876");
    	// 启动Producer实例
        producer.start();
    	for (int i = 0; i < 100; i++) {
    	    // 创建消息，并指定Topic，Tag和消息体
    	    Message msg = new Message("TopicTest" /* Topic */,
        	"TagA" /* Tag */,
        	("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET) /* Message body */
        	);
        	// 发送消息到一个Broker
            SendResult sendResult = producer.send(msg);
            // 通过sendResult返回消息是否成功送达
            //sendStatus=SEND_OK ：发送状态
            // msgId ：producer 创建的消息ID
            // offsetMsgId ：Brocker创建的消息ID
            // messageQueue ：消息存储的队列
            // producer.shutdown():关闭生产者
           System.out.printf("%s%n", sendResult);
    	}
    	// 如果不再发送消息，关闭Producer实例。
    	producer.shutdown();
    }
}

3.3 设置消费者

创建consumer组
设置Name Server地址
设置消费位置，从最开始销毁
设置消息回调处理监听 -> 处理消息
Start consumer

//消息发送者
public class ConsumerTest {

    public static void main(String[] args) {
        try {
            // 实例化消息生产者Producer
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer ("consumergroup");

            // 设置NameServer的地址
            consumer.setNamesrvAddr("127.0.0.1:9876");

            //从最开始的位置开始消费
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            // 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
            //和发送者保持一致才能搜到消息
            consumer.subscribe("topic_log", "tags_error");

            // 注册回调实现类来处理从broker拉取回来的消息
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {

                    System.out.printf("%s 成功搜到消息: %s %n", Thread.currentThread().getName(), msgs);

                    // 标记该消息已经被成功消费
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            // 启动Producer实例
            consumer.start();

        }catch (Exception e){
            e.printStackTrace();
        }
    }
}

四.RocketMQ核心概念

4.1 RocketMQ工作原理
- RocketMQ 网络部署特点
  - 为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。
    不过，这里有个问题，如果某Broker节点宕机，如何保证数据不丢失呢？其解决方案是，将每个Broker集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。
  - Broker节点集群是一个主从集群，即集群中具有Master与Slave两种角色。Master负责处理读写操作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slave则会自动切换为Master去工作。所以这个Broker集群是主备集群。Consumer既可以从Master订阅消息，也可以从Slave订阅消息
  - 一个Master可以包含多个Slave，但一个Slave只能隶属于一个Master。 Maste与Slave 的对应关系是通过指定相同的BrokerName、不同的BrokerId 来确定的。BrokerId为0表示Master非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
- RocketMQ工作流程
  - 启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。
  - Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。
  - 收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。
  - Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。
  - Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。
4.2 Producer 生产者
- RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。
- RocketMQ中的消息生产者都是以生产者组（Producer Group）的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。
- Producer会使用一定的算法(随机轮询+规避故障)选择把消息发送到哪个master的某个queue中。
4.3 Producer 消费者
- Consumer 支持两种消费形式：拉取式消费、推动式消费。（主动，被动），RocketMQ中的消息消费者都是以消费者组（Consumer Group）的形式出现的。消费者组是同一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息，不同的 Consumer Group可以消费同一个Topic。
- 一个Consumer Group内的Consumer可以消费多个Topic的消息。
- 集群模式：一个Queue是不能被同一个ConsumerGroup中的多个Consumer消费的，目的是减少资源竞争提升整体性能。
4.4 Topic 消息主题
- Topic表示一类消息的集合，每个topic主题包含若干条message消息，每条message消息只能属于一个topic主题，Topic是RocketMQ进行消息订阅的基本单位。
4.5 Message
- 消息是指消息系统所传输信息的物理载体，生产和消费数据的最小单位，每条消息必须属于一个主题。
4.6.Tag 标签
- 为消息设置的标志，用于同一主题下区分不同类型的消息。来自同一业务单元的消息，可以根据不同业务目的在同一主题下设置不同标签。标签能够有效地保持代码的清晰度和连贯性，并优化RocketMQ提供的查询系统。消费者可以根据Tag实现对不同子主题的不同消费逻辑，实现更好的扩展性。Topic是消息的一级分类，Tag是消息的二级分类。
4.7 MessageQueue
- 一个Topic中可以包含多个Queue，一个Topic的Queue也被称为一个Topic中消息的分区（Partition）。
- 在集群模式下, 在一个Consumer Group内，一个Queue最多只能分配给一个Consumer，一个Cosumer可以分配得到多个Queue。这样的分配规则，每个Queue只有一个消费者，可以避免消费过程中的多线程处理和资源锁定，有效提高各Consumer消费的并行度和处理效率。
- 消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的 Consumer将不能消费消息。如果一个Consmer挂了，该Consumer Group中的其它Consumer可以接着消费原Consumer消费的Queue。
- 【注意】一个Topic可以对应多个消费者，一个Queue只能对应一个组中的一个消费者。
- 【注意】为了防止消息紊乱，一个Consumer Group 中的Consumer都是订阅相同Topic下的Queue。
- 在广播模式下一个队列要可以被多个消费者对应.在广播模式下，同一个 ConsumerGroup 中的每个 Consumer 监听全部的队列。需要注意的是，广播模式下因为每个 Consumer 实例都需要处理全部的消息，因此这种模式仅推荐在**通知推送、配置同步类小流量场景使用。
- Queue分为写队列(对于生产者来说是写) 和读队列(对于消费者来说) ，默认创建数量是都是4 ，这个读写队列是从逻辑上进行划分在物理上读/写是一个队列，Producer发送的消息进入写队列，Consumer从读队列获取数据，一半情况下读写队列数量是一样的。
- 可以通过可视化界面修改Topic中的队列数量
- perm用于设置对当前创建Topic的操作权限：2表示只写，4表示只读，6表示读写。
4.8 MessageId/Key
- RocketMQ中每个消息拥有唯一的MessageId，且可以携带具有业务标识的Key，以方便对消息的查询。不过需要注意的是，MessageId有两个：在生产者send()消息时会自动生成一个MessageId（msgId)，
- 当消息到达Broker后，Broker也会自动生成一个MessageId(offsetMsgId)。msgId、offsetMsgId与key都称为消息标识。
4.9 Rebalance重新负载
- 当消费者数量或者Queue的数量修改，Rebalance是把⼀个Topic下的多个Queue重新分配给Consumer Group下的Consumer。目的是增加消费能力。
- 由于一个队列值分配给一个Consumer,那么当Consumer Group中的消费者数量大于队列数量，那么多出来的Consumer分配不到队列。
4.10 消息拉取模式
- 消息的消费分为：拉取式 pull ,和推送是 push
- Pull:拉取式，需要消费者间隔一定时间就去遍历关联的Queue,实时性差但是便于应用控制消息的拉取
- Push:推送式，封装了Queue的遍历，实时性强，但是对系统资源占用比较多。
4.11 消息消费模式
- 广播模式(一对多):同一个Consumer Group 下的所有Consumer都会受到同一个Topic的所有消息。同一个消息可能会被消费多次。
- 集群模式(一对一):同一个Gonsumer Group 下的Consumer平分同一个Topic下的消息。同一个消息只是被消费一次。
4.12 Queue的分配算法
- Queue是如何分配给Consumer的，这对应了四种算法：平均分配策略，环形平均策略，一致性Hash策略，同机房策略。
- 平均分配【默认】：根据 qeueuCount (4)/ consumerCount (3)作为每个消费者平均分配数量，如果多出来的queue就再依次逐个分配给Consumer。
- 环形平均策略：根据消费者的顺序，一个一个的分配Queue即可类似于发扑克牌。
- 一致性Hash策略：该算法将Consumer的Hash值作为节点放到Hash环上，然后将Queue的hash值也放入Hash环上，通过顺时针进行就近分配。
- 同机房策略：该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分配策略或环形平均策略对所有queue进行分配。
- 平均分配性能比较高，一致性Hash性能不高，但是能减少Rebalance，如果Consumer数量变动频繁可以使用一致性Hash。
4.13 Offset管理
- RockertMQ通过Offset来维护Consumer的消费进度，比如：消费者从哪个位置开始持续消费消息的？这里有三个枚举来指定从什么位置消费
- CONSUME_FROM_LAST_OFFSET：从queue的最后一条消息开始消费
- CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费
- CONSUME_FROM_TIMESTAMP：从某个时间戳位置的消息开始消费。
- 消费者消费结束之后，会向Consumer会提交其消费进度offset给Broker。Offset信息的存储分为本地 Offset管理和远程Offset管理
- 远程Offset管理:Brocker通过 store/config/consumerOffset.json 文件以JSON方式来存储offset相关数据以json的形式：适用于集群模式
- 本地Offset管理：offset相关数据以json的形式持久化到Consumer本地磁盘文件中，路径为当前用户主目录下的.rocketmq_offsets/c l i e n t I d / {clientId}/clientId/{group}/Offsets.json ：适用于广播模式
- Offset的同步提交与异步提交: 集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种：
- 同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，等待broker的成功响应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费者是阻塞的。其严重影响了消费者的吞吐量。
- 异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。
4.13 Offset管理
- 消息不会被单独清理，消息是顺序存储到commitlog的，消息是以commitlog为单位进行清理，RocketMQ有自己的清理规则，默认是72小时候后进行清理
- 到达时间清理点，自动清理过期的文件(凌晨4点)
- 磁盘空间使用率达到了过期清理阈值(75%),自动清理过期的文件。
- 磁盘占用率达到清理阈值(85%)，开始按照设定的规则清理文件，从老的文件开始。
- 磁盘占用率达到系统危险阈值(90%),拒绝写入数据。

标签：Queue,实现,认知,Broker,Topic,消息,Consumer,RocketMQ
From： https://blog.csdn.net/weixin_43855095/article/details/145165048

一.RocketMQ的认识

1.1、RocketMQ是什么

1.2、RocketMQ的特征

二.RocketMQ的原理

2.1、RokcetMQ架构

三.RocketMQ入门

3.1 导入依赖

3.2 设置生产者

3.3 设置消费者

四.RocketMQ核心概念

相关文章

赞助商

阅读排行