如图所示为RocketMQ基本的部署结构，主要分为NameServer集群、Broker集群、Producer集群和Consumer集群四个部分。
NameServer的作用是注册中心，类似于Zookeeper，但又有区别于它的地方。每个NameServer节点互相之间是独立的，没有任何信息交互，也就不存在任何的选主或者主从切换之类的问题，因此NameServer与Zookeeper相比更轻量级。单个NameServer节点中存储了活跃的Broker列表（包括master和slave），这里活跃的定义是与NameServer保持有心跳。
Broker是具体提供业务的服务器，单个Broker节点与所有的NameServer节点保持长连接及心跳，并会定时将Topic信息注册到NameServer，顺带一提底层的通信和连接都是基于Netty实现的。
Broker中分master和slave两种角色，每个master可以对应多个slave，但一个slave只能对应一个master，master和slave通过指定相同的Brokername，不同的BrokerId （master为0）成为一个组。master和slave之间的同步方式分为同步双写和异步复制，异步复制方式master和slave之间虽然会存在少量的延迟，但性能较同步双写方式要高出10%左右。
另外，Broker中还存在一些非常重要的名词需要说明：
RocketMQ的Topic/Queue和JMS中的Topic/Queue概念有一定的差异，JMS中所有消费者都会消费一个Topic消息的副本，而Queue中消息只会被一个消费者消费；但到了RocketMQ中Topic只代表普通的消息队列，而Queue是组成Topic的更小单元，集群消费模式下一个消费者只消费该Topic中部分Queue中的消息，当一个消费者开启广播模式时则会消费该Topic下所有Queue中的消息。Topic和Queue的具体关系可以参考下图
Tags是Topic下的次级消息类型（注：Tags也支持 TagA || TagB 这样的表达式），可以在同一个Topic下基于Tags进行消息过滤。Tags的过滤需要经过两次比对，首先会在Broker端通过Tag hashcode进行一次比对过滤，匹配成功传到consumer端后再对具体Tags进行比对，以防止Tag hashcode重复的情况。Queue中具体的存储单元结构如下图：
单个Producer和一台nameserver保持长连接，定时查询topic配置信息，如果该nameserver挂掉，生产者会自动连接下一个nameserver，直到有可用连接为止，并能自动重连。与nameserver之间没有心跳。
单个Producer和与其关联的所有broker保持长连接，并维持心跳。默认情况下消息发送采用轮询方式，会均匀发到对应Topic的所有queue中。
单个Consumer和一台nameserver保持长连接，定时查询topic配置信息，如果该nameserver挂掉，消费者会自动连接下一个nameserver，直到有可用连接为止，并能自动重连。与nameserver之间没有心跳。
单个Consumer和与其关联的所有broker保持长连接，并维持心跳，失去心跳后，则关闭连接，并向该消费者分组的所有消费者发出通知，分组内消费者重新分配队列继续消费。
该配置用于在Topic不存在时自动创建，会造成的问题是自动新建的Topic只会存在于一台broker上，后续所有对该Topic的请求都会局限在单台broker上，造成单点压力。
broker master宕机，虽然理论上来说不能向该broker写入但slave仍然能支持消费，但受限于rocketmq的网络连接机制，默认情况下最多需要30秒，消费者才会发现该情况，这个时间可通过修改参数 pollNameServerInteval 来缩短。这个时间段内，发往该broker的请求都是失败的，而且该broker的消息无法消费，因为此时消费者不知道该broker已经挂掉。 直到消费者得到master宕机通知后，才会转向slave进行消费，但是slave不能保证master的消息100%都同步过来了，因此会有少量的消息丢失。但是消息最终不会丢，一旦master恢复，未同步过去的消息仍然会被消费掉。
RocketMQ架构上主要分为四部分构成：
消息生产者，负责生产消息。Producer通过MQ的负载均衡模块选择相应的Broker集群队列进行消息投递，投递的过程支持快速失败并且低延迟
RocketMQ中的消息生产者都是以生产者组(Producer Group)的形式出现的。生产者组是同一类生产者的集合，这类Producer发送相同Topic类型的消息。一个生产者组可以同时发送多个主题的消息。
消息消费者，负责消费消息。一个消息消费者会从Broker服务器中获取到消息，并对消息进行相关业务处理
RocketMQ中的消息消费者都是以消费者组(Consumer Group)的形式出现的。消费者组是统一类消费者的集合，这类Consumer消费的是同一个Topic类型的消息。消费者组使得在消息消费方法，实现负载均衡（讲一个Topic中不同的Queue平均分配给同一个Consumer Group的不同Consumer，并不是负载均衡）和容错(一个Consumer挂了，该Consumer Group中的其他Consumer可以接着消费元Consumer消费的Queue)的目标变得非常容易
消费者组中Consumer的数量应小于等于Topic的Queue数量。如果超出Queue数量，则多出的Consumer将不能消费消息。
不过一个Topic类型的消息可以被多个消费者组同时消费。
NameServer是一个Broker与Topic路由的注册中心，支持Broker的动态注册与发现。
RocketMQ的思想来自于Kafuka，而Kafka是以来了Zookeeper的。所以，在RocketMQ的早期版本也依赖Zookeeper。从30开始去掉了Zookeeper的依赖，使用了自己的NameServer。
NameServer通常也是以集群的方式部署，不过，NameServer是无状态的，即NameServer集群中的各个节点之间是无差异的，各个节点相互不进行信息通讯。那各个节点中的数据是如何进行数据同步的呢？在Broker节点启动时，轮询NameServer列表，与每个NameServer节点建立长连接，发起注册请求。在NameServer内部维护者一个Broker列表，用来动态存储Broker信息
Broker节点为了证明自己是活着的，为了维护与NameServer间的长连接，会将最新的信息以心跳包的方式上报给NameServer，每30秒发送一次心跳。心跳包中包含BrokerId、Broker地址(IP+Port)、Broker名称、Broker所属集群名称等等。NameServer在接收到心跳包后，会更新心跳时间戳，记录这个Broker的最新存活时间。
由于Broker关机、宕机或网络抖动等原因，NameServer没有收到Broker的心跳，NameServer可能会将其从Broker列表中剔除
NameServer中有一个定时任务，每隔10秒就会扫描一次Broker表，查看每一个Broker的最新心跳时间戳距离当前时间是否超过120秒，如果超过，则会判定Broekr失效，然后将其从Broker列表中剔除。
RocketMQ的路由发现采用的是Pull模型。当Topic路由信息出现变化时，NameServer不会主动推送给客户端，而是客户端定时拉取最新的路由。默认每30秒拉取一次最新的路由
客户端再配置时必须要写上NameServer集群的地址，那么客户端道理连接在哪个NameServer节点呢客户端首先会生产一个随机数，然后再与NameServer节点数取模，此时得到的就是要连接的节点索引，然后就会进行连接。如果连接失败，则会采用round-robin策略，逐个尝试去连接其他节点。
首先采用的是 随机策略 进行选择，失败后采用的是轮询策略。
Broker充当着消息中转角色，负责存储消息、转发消息。Broker在RocketMQ系统中负责接收并存储从生产者发送来的消息，同时为消费者的拉取请求作准备。Broker同时也存储着消息相关的元数据，包括消费者组、消费进度偏移offset、主题、队列等
Remoting Module ：整个Broker的实体，负责处理来自clients端的请求。而这个Broker实体则由以下模块构成。
Client Manager ：客户端管理器。负责接收、解析客户端(Producer/Consumer)请求，管理客户端。
Store Service ：存储服务。提供方便简单的API接口，处理消息存储到物理硬盘和消息查询功能。
HA Service ：高可用服务，提供Master Broker和Slave Broker之间的数据同步功能。
Index Service ：索引服务。根据特定的Message Key，对投递到Broker的消息进行索引服务，同时也提供根据Message Key对消息进行快速查询的功能
为了增强Broker性能与吞吐量，Broker一般都是以集群形式出现的。各集群节点中可能存放着相同Topic的不同Queue。
如果某Broker节点宕机，如何保证数据不丢失呢？
其解决方案是，将每个Broekr集群节点进行横向扩展，即将Broker节点再建为一个HA集群，解决单点问题。
Broker节点集群是一个主从集群，即有Master和Slave两种角色。Master负责处理读写 *** 作请求，Slave负责对Master中的数据进行备份。当Master挂掉了，Slaver会自动切换为Master去工作。所以这个Broker集群式主备集群。Master与Slave的对应关系是通过指定相同的BrokerName、不同的BrokerId来确定的。BrokerId为0表示Master，非0表示Slave。每个Broker与NameServer集群中的所有节点建立长连接，定时注册Topic信息到所有NameServer。
①启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接
②启动Broker时，Broker会与所有的NameServer保持长连接，每30秒向NameServer定时发送心跳包
③发送消息前，可以先创建Topic ,创建Topic时需要指定该Topic要存储在哪些Broker上，当然，在创建Topic时也会将Topic与Broker的关系写入到NameServer中。也可以在发送消息时自动创建Topic。
④Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取路由信息，即当前发送Topic的Queue与Broker地址的映射关系。然后根据算法策略从队选择一个Queue，与队列所在的Broker建立长连接从而向Broker发送消息。
⑤Consumer与Producer类似，跟其中一台NameServer建立长连接，获取其所订阅Topic的路由信息，然后根据算法策略从路由信息中获取到其要消费的Queue，然后与Broker建立长连接，消费其中的消息。Consumer会向Broker发送心跳，以确保Broker的存活状态
手动创建Topic时，有两种模式：
自动创建Topic时，默认采用的是Broker模式，会为每个Broker默认创建四个Queue
从物理上讲，读/写队列是同一个队列。所以，不存在读/写队列数据同步问题。读/写队列是逻辑上进行区分的概念 。一般来说，读/写队列数量是相同的。
读/写队列数量不同是有问题的。
但这样可以方便缩容
perm用于设置对当前创建Topic的 *** 作权限：2表示只写，4表示只读，6表示读写
11 需要java环境，jdk18
rocketMQ包为：rocketmq-all-440-bin-releasezip
启动NAMESERVER
进入至‘MQ文件夹\bin’下，然后执行‘nohup sh mqnamesrv &’，启动NAMESERVER。
查看日志的命令：tail -f ~/logs/rocketmqlogs/namesrvlog
启动BROKER
进入至‘MQ文件夹\bin’下，然后执行‘nohup sh mqbroker -n localhost:9876 &’，启动BROKER。
你也可以nohup sh mqbroker-c /conf/brokerconf -n 1921680128:9876 autoCreateTopicEnable=true &
这样启动的服务器可以自动创建主题（客户端）,不过生产一般不推荐
查看日志的命令：tail -f ~/logs/rocketmqlogs/brokerlog
这个时候rocket服务已经正常启动，本地能访问，但是外部服务无法访问。
进入conf/brokerconf中，添加namesrvAddr=IP:9876、brokerIP1=IP地址
关闭broker服务，使用nohup sh mqbroker-c /conf/brokerconf -n 1921680128:9876 autoCreateTopicEnable=true & 重新启动服务，外部服务就能访问到rocker服务了
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）