Cassandra源码学习：Gossip协议

Gossip协议概述

Cassandra集群中的节点没有主次之分，它们通过一种叫做Gossip的协议进行通信。通过Gossip协议，它们可以知道集群中有哪些节点，以及这些节点的状态如何？每一条Gossip消息上都有一个版本号，节点可以对接收到的消息进行版本比对，从而得知哪些消息是我需要更新的，哪些消息是我有而别人没有的，然后互相倾诉吐槽，确保二者得到的信息相同，这很像现实生活中的八卦（摆龙门阵）一样，一传十，十传百，最后尽人皆知。

在Cassandra启动时，会启动Gossip服务，Gossip服务启动后会启动一个任务GossipTask，这个任务会周期性地与其他节点进行通信。GossipTask是位于org.apache.cassandra.gms.Gossip类下的一个内部类，其run方法如下：

[java] view plain copy print ?

1. publicvoidrun(){
2. MessagingService.instance().waitUntilListening();
4. /*Updatethelocalheartbeatcounter.*/
5. endpointStateMap.get(FBUtilities.getbroadcastAddress()).getHeartBeatState().updateHeartBeat();
7. finalList<GossipDigest>gDigests=newArrayList<GossipDigest>();
8. Gossiper.instance.makeRandomGossipDigest(gDigests);
9. if(gDigests.size()>0){
10. GossipDigestSyndigestSynMessage=newGossipDigestSyn(DatabaseDescriptor.getClusterName(),
11. DatabaseDescriptor.getPartitionerName(),
12. gDigests);
13. MessageOut<GossipDigestSyn>message=newMessageOut<GossipDigestSyn>(MessagingService.Verb.GOSSIP_DIGEST_SYN,
14. digestSynMessage,
15. GossipDigestSyn.serializer);
16. /*Gossiptosomerandomlivemember*/
17. booleangossipedToSeed=doGossipToLiveMember(message);
18. /*GossiptosomeunreachablememberwithsomeprobabilitytocheckifheisbackuP*/
19. doGossipToUnreachableMember(message);
20. if(!gossipedToSeed||liveEndpoints.size()<seeds.size())
21. doGossipToSeed(message);
22. doStatusCheck();
23. }
24. }

主要做了几下几件事：

1、GossipTask在Gossip启动后并不会立即运行，阻塞在listenGate这个条件变量上，当Gossip服务调用listen时才开始运行；

2、首先更新本节点的心跳版本号，然后构造需要发送给其他节点的消息gDigests；

3、从存活节点中随机选择一个节点发送、从失效节点中随机选取一个发送。如果当前存活节点数小于种子数，向其中一个种子节点发消息；

4、检查节点状态。

有关种子节点和节点状态后面再分析，本节只关注Gossip协议本身。

GossipTask用于向其他节点发送Gossip信息，Cassandra还提供了SocketThread这样一个线程来负责接收消息，接收消息的代码在org.apache.cassandra.net.IncomingTcpConnection类中。不管是发送还是接收Gossip消息，都是调用org.apache.cassandra.net.MessagingService的sendOneWay方法实现的。

一次Gossip通信分为三个阶段，如图所示：

每个阶段对应一个消息类，这三个消息类都有对应的处理器，处理器的注册是在org.apache.cassandra.service.StorageService类中完成的，参考如下代码：

MessagingService.instance().registerVerbHandlers(MessagingService.Verb.GOSSIP_DIGEST_SYN,newGossipDigestSynVerbHandler());
1. MessagingService.instance().registerVerbHandlers(MessagingService.Verb.GOSSIP_DIGEST_ACK,newGossipDigestAckVerbHandler());
2. MessagingService.instance().registerVerbHandlers(MessagingService.Verb.GOSSIP_DIGEST_ACK2,newGossipDigestAck2VerbHandler());

可见这三种消息分别对应三个消息类型GOSSIP_DIGEST_SYN、GOSSIP_DIGEST_ACK、GOSSIP_DIGEST_ACK2.
下面详细分析每个阶段。