消息队列6-Raft协议与仲裁队列、Pull拉模式

文章目录一. Raft协议1. 节点会扮演的 角色2. 任期(term)3. 选举过程4. 选取过程中其他情况(1) 情况1(2) 情况25. 副本消息复制流程二. 仲裁队列的使用1. 声明仲裁队列2. 发送消息3. 仲裁队列信息4. 宕机演示三. 节点与仲裁队列与副本之间的关系四. Pull拉模式1. 声明队列2. 发送消息3. 消费消息4. 结果图一. Raft协议介绍仲裁队列之前免不了要提一致性算法-Raft, 这是⼀种用于管理和维护分布式系统⼀致性的协议, 它是⼀种共识算法, 旨在实现高可用性和数据的持久性, Raft 通过在节点间复制数据来保证分布式系统中的⼀致性, 即使在节点故障的情况下也能保证数据不会丢失涉及到分布式, 意味着Raft协议是在集群中使用的, 在Raft集群必须存在⼀个主节点(Leader), 客户端向集群发起的所有操作都必须经由主节点处理, 没有主节点集群就无法正常工作, 因此选举leader是Raft协议中最重要的部分1. 节点会扮演的 角色在 Raft 算法中, 每个节点都处于以下三种角色之一leader:处理客户端传来的所有请求, 然后将所有请求封装为日志传入给所有从节点进行复制, 在有leader存在时, leader定期向所有 Follower 发送心跳消息, 以维持其领导者地位, 防止 Follower进入选举过程Follower:接收leader传来的日志, 并在本地运行日志中的操作, 不直接处理客户端的请求Candidate:Follower在一段时间后没有收到 leader 传来传送的心跳, 那么就会认定主节点已经宕机, 接下来就会变为Candidate(候选者)状态, 尝试通过投票来成为新的leader2. 任期(term)每一段任期从一次选举开始到正常工作结束, 当这次选举成功后, 使用该任期号, 同时任期号是单调递增的, 服务启动期间永远不会减小leader和follower的状态维持会一直到leader自身发生异常, 只有当没有leader时才会发生选举任期2因为leader丢失发生选举, 如果进行任期3选举没成功时, 会直接结束当前任期, trem自增变为4, 进行下一次选举Term 更像是⼀个逻辑时钟(logic clock)的作用, 有了它就可以发现哪些节点的状态已经过期, 每⼀个节点都保存⼀个 current term, 在通信时会带上term如果一个节点收到了一个带着过期的任期号的请求, 那么它会拒绝这次请求如果⼀个节点的当前任期号比其他节点小, 那么它就将自己的任期号更新为较大的那个值如果⼀个 candidate 或者 leader 发现自己的任期号过期了, 它就会立刻回到 follower 状态3. 选举过程采取投票机制, 每一个服务器节点会按照先来先服务原则(first-come-first-served)只投给一个候选者(candidate) 并且 候选人的任期号不能比自己小1. 服务刚启动时, 所有节点都是follow状态2. 过期时间内没收到leader的心跳, 先过期的从节点, 自增当前的任期号, 接着会从follower切换到candidate状态, 并投自己一票, 发起选举3. 赢得了大于1/2的选票(包含自己的一票), 那么新的leader会立刻给所有节点发消息, 全部告知, 避免其余节点触发新的选举4. 选取过程中其他情况(1) 情况1如果有s1和s3两个候选者, 两者同时发起的选举, 但是s3的选举消息先到达的其他节点, 已经给s3投了一票, 那么后来的s1的选举信息到达时, 就不会再投票, 这时s3胜利后, 给所有节点发送leader心跳, s1收到后发现term值不低于自己, 就会将状态从candidate 转化为 follower, 并将主节点信息保存下来(2) 情况21. 没有任何节点获得超过半数的投票, 比如所有的 follower 同时变成 candidate, 然后它们都将票投给自己, 那这样就没有 candidate 能得到超过半数的投票了, 当这种情况发生的时候, 每个 candidate 都会进行一次超时响应, 然后通过自增任期号来开启⼀轮新的选举, 并启动另一轮的 RequestVote RPCs(请求投票通信)2. Raft 采用随机选举超时时间randomized election timeouts 来确保很少产生无结果的投票并且就算发生了也能很快地解决。为了防止选票⼀开始就被瓜分选举超时时间是从一个固定的区间比如150-300ms中随机选择5. 副本消息复制流程客户端(生产者和消费者)只会与主副本进行交互, 主副本再将这些操作命令传给从副本来进行复制, 主副本宕机后, 从副本中会选举一个新的主副本消息复制和主副本选举的操作和上面选举leader一样, 需要超过半数的副本同意, 当生产者发送一条消息, 需要超过半数的队列副本都将消息写入磁盘以后才会向生产者进行确认, 这意味着少部分比较慢的副本不会影响整个队列的性能二. 仲裁队列的使用仲裁队列Quorum Queue内部确实内置并实现了Raft协议1. 声明仲裁队列ConfigurationpublicclassQuorumConfig{Bean(quorumQueue)publicQueuequorumQueue(){// quorum() 方法来开启仲裁队列功能returnQueueBuilder.durable(Constants.QUORUM_QUEUE).quorum().build();}}2. 发送消息仲裁队列和不同队列发送消息的方式是一样的, 我们这里采用RabbitMQ的web管理界面来发送消息3. 仲裁队列信息可以看到主副本的位置在oj-rabbit-dev节点上可以看到队列中存的信息4. 宕机演示宕机之前三个节点状态都为绿色(在线)有两个镜像节点, 消息数为1, 主副本在oj-rabbit-dev节点上宕机之后oj-rabbit-dev节点变为红色(宕机)有1个镜像节点, 消息数为1, 主副本变成了在rabbit1节点上, 可以看出消息没有丢失, 也选举出了新的leader副本三. 节点与仲裁队列与副本之间的关系★一个仲裁队列有多个副本每个副本独占一个节点一个节点可以放多个不同队列的副本但每个队列在该节点上最多只有一个副本, 下面是节点与仲裁队列与副本之间的关系图:RabbitMQ 集群多个节点│├── 仲裁队列 A一个逻辑队列│ ├── 副本1位于节点1是 Leader│ ├── 副本2位于节点2是 Follower│ └── 副本3位于节点3是 Follower│└── 仲裁队列 B另一个逻辑队列├── 副本1位于节点2是 Leader├── 副本2位于节点3是 Follower└── 副本3位于节点1是 Follower1. Leader 和 Follower 指的是“副本”的身份而不是“节点”的身份2. 每个仲裁队列有多个副本, 由一个主副本和多个从副本组成, 每个副本都在不同的 RabbitMQ 节点上3. 一个节点上可以同时承载 许多不同仲裁队列 的副本4. 对于 同一个仲裁队列任意节点上最多只能存放该队列的 一个副本要么是 Leader要么是 Follower), 不允许同一个队列的两个副本落在同一节点上5. 节点只是副本的“容器”。一个节点上可能有一个副本是仲裁队列 A 的 Leader同时另一个副本是仲裁队列 B 的 Follower6. 在标准的 Raft 共识算法中, 每个参与 Raft 集群的节点确实会在这三种状态之间切换, 但是在 RabbitMQ 的仲裁队列实现中这个 Raft 集群的粒度是“每个队列”而不是整个 RabbitMQ 节点, 因为每个仲裁队列都运行着自己独立的 Raft 实例, 同一个 RabbitMQ 节点 会参与到多个不同的 Raft 集群 (即多个不同的仲裁队列) 中7. 因此节点并没有一个全局的“Leader/Follower/Candidate”身份, 它只是在不同队列的 Raft 集群中, 由它上面的那个副本扮演不同的角色四. Pull拉模式1. 推模式指的是消息中间件主动将消息推送给消费者, 有新消息会立马推送消费者2. 拉模式指的是消费者主动从消息中间件拉去消息, 有新消息不会立马推送给消费者3. RabbitMQ中, 主要基于推模式工作的, 消费者通过订阅队列, Broker有消息时会自动将消息推给消费者, 如果只想要获取单个消息, 不是持续订阅, 需要采取拉模式的方法4. RabbitMQ既支持推模式也支持拉模式1. 声明队列// pull模式Bean(pullQueue)publicQueuepullQueue(){returnQueueBuilder.durable(Constants.PULL_QUEUE).build();}2. 发送消息RequestMapping(/pull)publicStringpull(){for(inti0;i20;i){rabbitTemplate.convertAndSend(,Constants.PULL_QUEUE,pull模式...i);}return发送成功;}3. 消费消息RequestMapping(/consumer)RestControllerpublicclassConsumerController{Resource(namerabbitTemplate)privateRabbitTemplaterabbitTemplate;RequestMapping(/get)publicStringget(){MessagemessagerabbitTemplate.receive(Constants.PULL_QUEUE);System.out.println(接收到消息message: message);return拉取到消息message: message;}}4. 结果图队列中消息总数变为了19条, 消费者拉取了一条消息