cassandra 之 gossip实现

1 概述

本文的描述基于cassandra 0.6.3 源代码。

gossip协议解决了在分布式环境中，如何使用最少的网络带宽，达到数据同步的目的。同步的数据包括：集群中有哪些节点以及这些节点的状态。

重点提示：节点只能更新属于自己的状态数据。

在cassandra中，和状态相关的类图的设计见图1。endPointStateMap_存储了本节点已知的节点的状态。状态的数据由2部分组成：心跳状态数据(hbState_)和 应用数据(applicationState_)。

心跳状态数据

■generation_ 用来区分是否是宕机后重启，存储在系统表中，每次启动是增加1

■version_ 每隔1秒，增加1

应用状态数据

由多个三元组组成。三元组的结构是：<key,state_,version>。具体的数据有：

■磁盘的使用情况(keyis ”LOAD-INFORMATION”)

■节点的生命周期状态(key is ”MOVE”，value is BOOT/NORMAL/LEAVING/LEFT)。

2 实现

2.1 syn消息

1) 从endPointStateMap_获取各个节点的摘要数据(digest)，包括本节点

<addr, heartbeat-generation,max-version>

addr：节点地址

generation：从hbState_中获取

max-version：从applicationState_中获取最大的版本号

2) 把摘要数据打包

限制：数据包的大小限制是1428，一次只发送一个包。为了在数据链路层分片，所以限制了一次send的数据量大小，1428= 1500(MTU) – 20(ipheader len) – 20(tcp header len) – cluster name len(32)。

如果摘要数据大小大于1428，则只发送部分数据。选择哪部分数据？ 随机选择，但是本节点的摘要数据一定被包括。

代码参见Gossiper.GossipTimerTask.run()。

2.2 ack消息

收到syn消息后，按照如下流程进行处理

1)检查cluster name

2)根据摘要数据，更新本地为各个节点维护的心跳队列

If 远端generation > 本地generation

更新心跳队列；

elseif远端generation == 本地generation && 远端maxversion > 本地maxversion

更新心跳队列；

提示：心跳队列用来进行故障探测，具体参见我的文章《cassandra之failure detector》。

3)比较摘要数据和本地数据，生成要返回的摘要数据和状态数据

■生成返回的摘要的数据

if 本地节点没有此节点的数据

生成摘要数据：<addr, 远端generation,0>

elseif远端generation > 本地generation

生成摘要数据： <addr, 远端generation,0>

elseif远端generation == 本地generation&& 远端max-version >本地远端max-version

生成摘要数据： <addr, 远端generation,本地max-version >

■生成返回的状态数据

应用状态数据

if远端generation < 本地generation

生成返回数据：此节点的所有状态数据

elseif远端generation == 本地generation && 远端max-version < 本地max-version

生成返回数据：此节点的所有版本号大于远端max-version状态数据。

心跳数据

hbState_设置和本地的一样。

■把待返回的摘要数据和状态数据打包

限制：数据包的大小限制是1428(1500-20-20-clusternamelen)，一次只发送一个包。如果待返回的数据大于1428，摘要数据的优先级大于状态数据。在摘要数据中，节点的max-version和摘要数据的差值的绝对值越大，优先级越高。在状态数据中亦然如此。

4) 发送数据包

代码参见GossipDigestSynVerbHandler。

2.3 ack2消息

2.3.1处理ack消息

1)处理ack消息中的状态数据

■更新心跳队列

远端 hbState_ . generation(记为G1) 和 hbState_ . version_(记为V1)

本地hbState_ . generation (记为G2) 和 applicationState_中最大的 version_(记为V2)

if G1 > G2

更新心跳队列；

elseif G1==G2 && V1>V2

更新心跳队列；

■更新本地的状态数据

远端 hbState_ . generation(记为G1) 和applicationState_中最大的 version_ (记为V1)

本地hbState_ . generation (记为G2) 和 applicationState_中最大的 version_(记为V2)

if 本地没有此节点的状态数据

把此节点的状态保存在本地

elseif G1 > G2

删除此节点的所有状态数据，把ack消息中此节点的状态数据存储在本地

elseif G1 == G2 && V1 > V2

比较远端的applicationState_ 和 本地applicationState_。若远端数据项的版本号比本地的新，

这更新本地的数据项目；若远端的数据项在本次不存在，则添加至本地的applicationState_。

2)处理ack消息中的摘要数据

■生成待返回的状态数据

应用状态数据

本地的应用状态数据和摘要数据中的version比较，若大于，这加入至待返回的数据中，

心跳状态数据

hbState_设置和本地的一样。

注意：没有比较generation，因为在远端生成这些数据的时候，已经比较，个人觉得还是比较为好，因为数据是不断变化的。

3)发送在上个步骤中生成的状态数据

需要保证发送的数据不超过1428字节。

代码参见GossipDigestAckVerbHandler。

2.3.2处理ack2消息

同2.3.1 中的步骤1）。

代码参见GossipDigestAck2VerbHandler。