master 启动后创建 acceptsocketservice 服务 , 用来创建客户端到服务端的 tcp 链接。
rocketmq 主从复制是 rocketmq 高可用机制之一,数据可以从主节点复制到一个或多个从节点。
这篇文章,我们聊聊 rocketmq 的主从复制,希望大家读完之后,能够理解主从复制的精髓。
图片
在 rocketmq 的集群模式中,broker 分为 master 与 slave,一个 master 可以对应多个 slave,但是一个 slave 只能对应一个 master。
每个 broker 与 name server 集群中的所有节点建立长连接,定时注册 topic 信息到所有 name server。
图片
master 节点负责接收客户端的写入请求,并将消息持久化到磁盘上。而 slave 节点则负责从 master 节点复制消息数据,并保持与 master 节点的同步。
图片
每个 master 配置一个 slave ,有多对 master-slave ,ha 采用同步双写方式,即只有主备都写成功,才向应用返回成功。
这种模式的优缺点如下:
图片
每个 master 配置一个 slave ,有多对 master-slave ,ha 采用异步复制方式,主备有短暂消息延迟(毫秒级),这种模式的优缺点如下:
复制流程分为两个部分:元数据复制和消息数据复制。
slave broker 定时任务每隔 10 秒会同步元数据,包括主题,消费进度,延迟消费进度,消费者配置。
图片
同步主题时, slave broker 向 master broker 发送 rpc 请求,返回数据后,首先加入本地缓存里,然后持久化到本地。
图片
下图是 master 和 slave 消息数据同步的流程图。
图片
master 启动后创建 acceptsocketservice 服务 , 用来创建客户端到服务端的 tcp 链接。
图片
rocketmq 抽象了链接对象 haconnection , haconnection 会启动两个线程,分别用于读服务和写服务:
图片
haclient 是客户端 slave 的核心类 ,负责和 master 创建连接和数据交互。
图片
客户端在启动后,首先尝试连接 master , 查询当前消息存储中最大的物理偏移量 ,并存储在变量 currentreportedoffset 里。
图片
上报进度的数据格式是一个 long 类型的 offset , 8个字节 , 非常简洁 。
图片
发送到 socket 缓冲区后 , 修改最后一次的写时间 lastwritetimestamp 。
当 slave 上报数据到 master 时,触发 selectionkey.op_read 事件,master 将请求交由 readsocketservice 服务处理:
图片
当 slave broker 传递了自身 commitlog 的 maxphyoffset 时,master 会马上中断 selector.select(1000) ,执行 processreadevent 方法。
图片
processreadevent 方法的核心逻辑是设置 slave 的当前进度 offset ,然后通知复制线程当前的复制进度。
写服务 writesocketservice 从消息文件中检索该偏移量后的所有消息(传输批次数据大小限制),并将消息数据发送给 slave。
图片
图片
首先 haclient 类中调用 dispatchreadrequest 方法 , 解析出消息数据 ;
图片
然后将消息数据 append 到本地的消息存储。
图片
从数据复制流程图,我们发觉数据复制本身就是一个异步执行的,但是同步是如何实现的呢?
master broker 接收到写入消息的请求后 ,调用 commitlog 的 aysncputmessage 方法写入消息。
图片
这段代码中,当 commitlog 执行完 appendmessage 后, 需要执行刷盘任务和同步复制两个任务。
但这两个任务并不是同步执行,而是异步的方式,使用了 completablefuture 这个异步神器。
当 haconnection 读服务接收到 slave 的进度反馈,发现消息数据复制成功,则唤醒 future 。
图片
最后 broker 组装响应命令 ,并将响应命令返回给客户端。
rocketmq 主从复制的实现思路非常简洁,slave 启动一个线程,不断从 master 拉取 commit log 中的数据,然后在异步 build 出 consume queue 数据结构。
核心要点如下:
1、主从复制包含元数据复制和消息数据复制两个部分;
2、元数据复制
slave broker 定时任务每隔 10 秒向 master broker 发送 rpc 请求,将元数据同步到缓存后,然后持久化到磁盘里;
3、消息数据复制
4、同步的实现
当 commitlog 执行完 appendmessage 后, 需要执行刷盘任务和同步复制两个任务,这里用到了 completablefuture 这个异步神器。
当 haconnection 读服务接收到 slave 的进度反馈,发现消息数据复制成功,则唤醒 future 。最后 broker 组装响应命令 ,并将响应命令 返回给客户端 。
友情链接