Gossip協議概述

Cassandra叢集中的節點沒有主次之分，它們通過一種叫做Gossip的協議進行通訊。通過Gossip協議，它們可以知道叢集中有哪些節點，以及這些節點的狀態如何？每一條Gossip訊息上都有一個版本號，節點可以對接收到的訊息進行版本比對，從而得知哪些訊息是我需要更新的，哪些訊息是我有而別人沒有的，然後互相傾訴吐槽，確保二者得到的資訊相同，這很像現實生活中的八卦（擺龍門陣）一樣，一傳十，十傳百，最後盡人皆知。 在Cassandra啟動時，會啟動Gossip服務，Gossip服務啟動後會啟動一個任務GossipTask，這個任務會週期性地與其他節點進行通訊。GossipTask是位於org.apache.cassandra.gms.Gossip類下的一個內部類，其run方法如下：

public void run(){
      MessagingService.instance().waitUntilListening();

      /* Update the local heartbeat counter. */
      endpointStateMap.get(FBUtilities.getBroadcastAddress()).getHeartBeatState().updateHeartBeat();

      final List<GossipDigest> gDigests = new ArrayList<GossipDigest>();
      Gossiper.instance.makeRandomGossipDigest(gDigests);

      if (gDigests.size() > 0){
          GossipDigestSyn digestSynMessage = new GossipDigestSyn(DatabaseDescriptor.getClusterName(),
                                                                 DatabaseDescriptor.getPartitionerName(),
                                                                 gDigests);
          MessageOut<GossipDigestSyn> message = new MessageOut<GossipDigestSyn>(MessagingService.Verb.GOSSIP_DIGEST_SYN,
                                                                                digestSynMessage,
                                                                                GossipDigestSyn.serializer);
          /* Gossip to some random live member */
          boolean gossipedToSeed = doGossipToLiveMember(message);

          /* Gossip to some unreachable member with some probability to check if he is back up */
          doGossipToUnreachableMember(message);

          if (!gossipedToSeed || liveEndpoints.size() < seeds.size())
              doGossipToSeed(message);

          doStatusCheck();
      }
}
 

主要做了幾下幾件事： 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, new GossipDigestSynVerbHandler());
        MessagingService.instance().registerVerbHandlers(MessagingService.Verb.GOSSIP_DIGEST_ACK, new GossipDigestAckVerbHandler());
        MessagingService.instance().registerVerbHandlers(MessagingService.Verb.GOSSIP_DIGEST_ACK2, new GossipDigestAck2VerbHandler());
 

可見這三種訊息分別對應三個訊息型別GOSSIP_DIGEST_SYN、GOSSIP_DIGEST_ACK、GOSSIP_DIGEST_ACK2.
下面詳細分析每個階段。

GossipDigestSynMessage

GossipTask的run方法中傳送了GOSSIP_DIGEST_SYN型別的訊息(GossipDigestSynMessage)，這種訊息交給其對應的處理器GossipDigestAckVerbHandler處理，具體處理過程在doVerb()方法中，核心程式碼如下：

public void doVerb(MessageIn<GossipDigestSyn> message, int id){
    //...check
    
    List<GossipDigest> gDigestList = gDigestMessage.getGossipDigests();
   
    doSort(gDigestList);

    List<GossipDigest> deltaGossipDigestList = new ArrayList<GossipDigest>();
    Map<InetAddress, EndpointState> deltaEpStateMap = new HashMap<InetAddress, EndpointState>();
    Gossiper.instance.examineGossiper(gDigestList, deltaGossipDigestList, deltaEpStateMap);
    logger.trace("sending {} digests and {} deltas", deltaGossipDigestList.size(), deltaEpStateMap.size());
    MessageOut<GossipDigestAck> gDigestAckMessage = new MessageOut<GossipDigestAck>(MessagingService.Verb.GOSSIP_DIGEST_ACK,
                                                                                    new GossipDigestAck(deltaGossipDigestList, deltaEpStateMap),
                                                                                    GossipDigestAck.serializer);
    Gossiper.instance.checkSeedContact(from);
    MessagingService.instance().sendOneWay(gDigestAckMessage, from);
}

1、對接受到的訊息排序：先按generation排序，如果generation相同，按maxVersion與本地版本差排序； 2、examineGossiper對比接收到的資訊與本節點的差異，本節點需要進一步獲取的訊息由deltaGossipDigestList儲存，本節點需要告訴from節點的資訊由deltaEpStateMap儲存； 3、利用deltaGossipDigestList和deltaEpStateMap封裝成GossipDigestAckMessage訊息，傳送給from節點。

GossipDigestAckMessage

同GossipDigestSynMessage訊息一樣，GOSSIP_DIGEST_ACK型別的訊息由處理器GossipDigestAckVerbHandler處理，也是doVerb()方法，實際上這三個處理器有共同的實現介面IVerbHandler。核心程式碼如下：

 public void doVerb(MessageIn<GossipDigestAck> message, int id){
    List<GossipDigest> gDigestList = gDigestAckMessage.getGossipDigestList();
    Map<InetAddress, EndpointState> epStateMap = gDigestAckMessage.getEndpointStateMap();
   
    if (epStateMap.size() > 0)
    {
        Gossiper.instance.notifyFailureDetector(epStateMap);
        Gossiper.instance.applyStateLocally(epStateMap);
    }

    Map<InetAddress, EndpointState> deltaEpStateMap = new HashMap<InetAddress, EndpointState>();
    for (GossipDigest gDigest : gDigestList)
    {
        InetAddress addr = gDigest.getEndpoint();
        EndpointState localEpStatePtr = Gossiper.instance.getStateForVersionBiggerThan(addr, gDigest.getMaxVersion());
        if (localEpStatePtr != null)
            deltaEpStateMap.put(addr, localEpStatePtr);
    }

    MessageOut<GossipDigestAck2> gDigestAck2Message = new MessageOut<GossipDigestAck2>(MessagingService.Verb.GOSSIP_DIGEST_ACK2,
                                                                                       new GossipDigestAck2(deltaEpStateMap),
                                                                                       GossipDigestAck2.serializer);
    MessagingService.instance().sendOneWay(gDigestAck2Message, from);
}

1、epStateMap 是from節點想要告訴它的訊息，呼叫applyStateLocally方法進行更新； 2、gDigestList是from節點想要進一步更新的訊息，呼叫getStateForVersionBiggerThan方法拿到from節點可以更新的訊息； 3、將2中的訊息封裝成GOSSIP_DIGEST_ACK2訊息傳送給from結點。

GossipDigestAck2Message

由GossipDigestAck2VerbHandler處理器處理，這裡就不貼程式碼，接收到from介面發過來的訊息，進行本地更新即可。 下節學習cassandra的機架感應 參考資料：