Implementierung des RAFT-Konsensalgorithmus für verteilten KV-Speicher in Java

Allgemeine Lösungsbeschreibung

• Anhänger (Anhänger). Akzeptiert Leseanforderungen vom Client. Nimmt dem Anführer einen Herzschlag ab
• Kandidat (Kandidat). Akzeptiert Leseanforderungen vom Client. Sendet Abstimmungsanfragen an andere Knoten
• Führer Akzeptiert Lese- und Schreibanforderungen. Sendet Heartbeat-Anforderungen an andere Knoten. Sendet Daten zum Anhängen von Anforderungen an andere Knoten.

Datenspeicher

public interface OperationsLog {
   void append(Operation operation);
   Operation get(Integer index);
   List<Operation> all();

   Long getTerm(Integer index);
   Integer getLastIndex();
   Long getLastTerm();

   void removeAllFromIndex(int newOperationIndex);
}

public interface Storage {
   List<Entry> all();
   String get(Long key);
   void insert(Long key, String val);
   void update(Long key, String val);
   void delete(Long key);
}

Timer

• Abstimmung bei der Durchführung einer Abstimmungsrunde
• Anhängen, auch bekannt als Heartbeat (falls ohne Daten), um Protokolldaten an Follower zu replizieren und den Beginn einer neuen Abstimmungsrunde zu verhindern.

• Abstimmung. Eine Abstimmungsrunde starten. Jeder Knoten hat ein eigenes Intervall, nach dem er versucht, eine neue Abstimmung zu starten. Der Countdown beginnt von neuem, wenn der Anführer einen Herzschlag erhält.
• Herzschlag. Senden einer Anfrage an Follower durch den Anhängeführer. Wenn der Knoten keinen Herzschlag empfängt und der Abstimmungszeitgeber abgelaufen ist, wird er ein Kandidat und leitet Wahlen ein, erhöht die Anzahl der Abstimmungsrunden und sendet Abstimmungsanfragen an andere Knoten. Wenn der Knoten die Mehrheit der Stimmen sammelt, wird er zum Anführer und sendet einen Herzschlag aus.

Der aktuelle Status des Knotens

public interface Context {
   Integer getId(); //    
   State getState();//: , ,  
   Integer getVotedFor(); 
               //          
   Long getCurrentTerm(); //  
   Integer getCommitIndex(); //    
   List<Peer> getPeers(); //      
}

public interface Peer {
   Integer getId(); //  
   Integer getNextIndex(); //  ,    
   Integer getMatchIndex();//   
   Boolean getVoteGranted(); //     
}

Wählen

public interface ElectionService {
   void processElection();
   AnswerVoteDTO vote(RequestVoteDTO requestVoteDTO);
} 

Senden eines Antrags auf Abstimmung

//1
context.setState(CANDIDATE); 
Long term = context.incCurrentTerm(); 
context.setVotedFor(context.getId()); 

List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());
long voteGrantedCount = 1L;
long voteRevokedCount = 0L;

//2
while (checkCurrentElectionStatus(term)) {
   List<AnswerVoteDTO> answers = getVoteFromAllPeers(term, peersIds);
   peersIds = new ArrayList<>();
   for (AnswerVoteDTO answer : answers) {
       //3
       if (answer.getStatusCode().equals(OK)) {
           //4
           if (answer.getTerm()>context.getCurrentTerm()) {
               context.setTermGreaterThenCurrent(answer.getTerm());
               return;
           }
           if (answer.isVoteGranted()) {
               //5 
               context.getPeer(answer.getId()).setVoteGranted(true);
               voteGrantedCount++;
           } else
               //6 
               voteRevokedCount++;
       } else {
          peersIds.add(answer.getId());
       }
   }
  //7
  if (voteGrantedCount >= context.getQuorum()) {
       winElection(term);
       return;
   } else if (voteRevokedCount >= context.getQuorum()) {
       loseElection(term);
       return;
   } 

1. Stellen Sie den Status "Kandidat" ein. Erhöhen Sie die runde Zahl und stimmen Sie für uns ab.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. Der Knoten hat für uns gestimmt! Wir erhöhen die Anzahl der Knoten, die Stimmen für uns abgeben, und korrigieren, dass dieser Knoten für uns gestimmt hat.
6. Nicht für uns gewählt, glauben wir auch.
7. Wenn das Quorum gesammelt wird und der Knoten die Wahl gewonnen hat, legen wir den Status „Leader“ fest. Ansonsten werden wir Anhänger und warten.

  private void winElection(Long term) {
       context.setState(LEADER);
       context.getPeers().forEach(peer ->
               peer.setNextIndex(operationsLog.getLastIndex()+1)

       );
   }

Bearbeitung von Abstimmungsanfragen

class RequestVoteDTO {
   private final Long term; //     
   private final Integer candidateId; //  
   private final Integer lastLogIndex; //     
   private final Long lastLogTerm; //       
}

public AnswerVoteDTO vote(RequestVoteDTO dto) {
   
       boolean termCheck;
       //1
       if (dto.getTerm() < context.getCurrentTerm())
           return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),false);
       else //2
       if (dto.getTerm().equals(context.getCurrentTerm())) {
           termCheck = (context.getVotedFor() == null||
                          context.getVotedFor().equals(dto.getCandidateId()));
       }
       else
       {   //3
           termCheck = true;
             context.setTermGreaterThenCurrent(dto.getTerm());
       }

       //4  
       boolean logCheck = !((operationsLog.getLastTerm() > dto.getLastLogTerm()) ||
               ((operationsLog.getLastTerm().equals(dto.getLastLogTerm())) &&
                       (operationsLog.getLastIndex() > dto.getLastLogIndex())));


       boolean voteGranted = termCheck&&logCheck;

       //5
       if (voteGranted) {
           context.setVotedFor(dto.getCandidateId());
       }
       //6   
       return new AnswerVoteDTO(context.getId(),context.getCurrentTerm(),voteGranted);
   }

1. Wenn die aktuelle Runde des Knotens größer ist als die Runde des Kandidaten, lehnen wir ab, da dies eine Anfrage eines verzögerten Knotens ist, der anscheinend einige Zeit außerhalb des Clusters lag und das Wahlverfahren gestartet hat, weil er den aktuellen Führer nicht gesehen hat.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. Mit einem positiven Ergebnis korrigieren wir die Tatsache, dass der Knoten an den Wahlen teilgenommen hat, und geben eine Stimme für den Kandidaten ab.
6. Senden Sie das Ergebnis an den Kandidaten zurück

Reproduzieren

class RequestAppendDTO {
   private final Long term; //   
   private final Integer leaderId; //   

   private final Integer prevLogIndex;//   
   private final Long prevLogTerm;//   
   private final Integer leaderCommit;//      
   private final Operation operation; //
}

public interface ReplicationService {
   void appendRequest();
   AnswerAppendDTO append(RequestAppendDTO requestAppendDTO);
}

Senden Sie eine Datenänderungsanforderung

private CompletableFuture<AnswerAppendDTO> sendAppendForOnePeer(Integer id) {
   return CompletableFuture.supplyAsync(() -> {
       try {
           //1
           Peer peer = context.getPeer(id);

           Operation operation;
           Integer prevIndex;
           //2    
           if (peer.getNextIndex() <= operationsLog.getLastIndex()) {
               operation = operationsLog.get(peer.getNextIndex());
               prevIndex = peer.getNextIndex() - 1;
           } else 
           //3  
           {
               operation = null;
               prevIndex = operationsLog.getLastIndex();
           }


           RequestAppendDTO requestAppendDTO = new RequestAppendDTO(
                   context.getCurrentTerm(), //   
                   context.getId(), //  
                   prevIndex,//      
                   operationsLog.getTerm(prevIndex),//  
                   context.getCommitIndex(),
                               //      
                   Operation //
           );

...
/*   http     */
}

1. Follower-Metadaten
2. , . ( ), , , , . , , , ,
3. , , ; , , ,

public void appendRequest() {
       List<Integer> peersIds = context.getPeers().stream().map(Peer::getId).collect(Collectors.toList());

       //1 
       while (peersIds.size() > 0) {
           //2 
           List<AnswerAppendDTO> answers = sendAppendToAllPeers(peersIds);
           peersIds = new ArrayList<>();
           for (AnswerAppendDTO answer : answers) {
               //3
               if (answer.getStatusCode().equals(OK)) {
                   //4
                   if (answer.getTerm() > context.getCurrentTerm()) {
                        context.setTermGreaterThenCurrent(answer.getTerm());
                       return;
                   }
                   Peer peer = context.getPeer(answer.getId());
                   //5     
                   if (answer.getSuccess()) {                      
                       peer.setNextIndex(answer.getMatchIndex() + 1);
                       peer.setMatchIndex(answer.getMatchIndex());
                       if (peer.getNextIndex() <= operationsLog.getLastIndex())
                           peersIds.add(answer.getId());
                   //6      
                   } else {
                       peer.decNextIndex();
                       peersIds.add(answer.getId());
                   }
               }
           }
           //7
           tryToCommit();
       }
}

1. Wir wiederholen die Anfrage, bis wir von allen Followern eine Antwort erhalten, dass die Replikation erfolgreich war. Da pro Anforderung eine Operation gesendet wird, kann es mehrere Iterationen dauern, um die Protokolle der Follower zu synchronisieren
2. Senden Sie Anfragen an alle Follower und erhalten Sie eine Liste mit Antworten
3. Wir berücksichtigen nur Antworten von verfügbaren Followern
4. Wenn sich herausstellt, dass die Runde eines der Anhänger mehr ist als die Runde des Anführers, stoppen wir alles und verwandeln uns in einen Anhänger
5. Wenn der Follower geantwortet hat, dass alles erfolgreich war, aktualisieren wir die Follower-Metadaten: Wir speichern den letzten Index des Follower-Protokolls und den Index der nächsten vom Follower erwarteten Operation.
6. , , , , . , , . , . , .
7. , . .

public AnswerAppendDTO append(RequestAppendDTO dto) {
     
       //1     
       if (dto.getTerm() < context.getCurrentTerm()) {
           return new AnswerAppendDTO(context.getId(),context.getCurrentTerm(),false, null);
       } else if (dto.getTerm() > context.getCurrentTerm()) {
           //2 
           context.setCurrentTerm(dto.getTerm());
           context.setVotedFor(null);
       }
       //3  
       applicationEventPublisher.publishEvent(new ResetElectionTimerEvent(this));

       if (!context.getState().equals(FOLLOWER)) {
           context.setState(FOLLOWER);
       }
        
       //4  
       if ((dto.getPrevLogIndex() > operationsLog.getLastIndex()) ||                                                                                        !dto.getPrevLogTerm().equals(operationsLog.getTerm(dto.getPrevLogIndex()))) {
                      return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), false, null);
       }


       Operation newOperation = dto.getOperation();
       if (newOperation != null) {
           int newOperationIndex = dto.getPrevLogIndex() + 1;
           
         synchronized (this) {
               //5
               if ((newOperationIndex <= operationsLog.getLastIndex()) &&
                      (!newOperation.getTerm().equals(operationsLog.getTerm(newOperationIndex)))){
                   operationsLog.removeAllFromIndex(newOperationIndex);
               }
               //6
               if (newOperationIndex <= operationsLog.getLastIndex())
               {
                 return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true,      operationsLog.getLastIndex());
               }
               //7
               operationsLog.append(newOperation);
           }
        }
        //8 
        if (dto.getLeaderCommit() > context.getCommitIndex()) {
           context.setCommitIndex(Math.min(dto.getLeaderCommit(), operationsLog.getLastIndex()));
       }

                 
       return new AnswerAppendDTO(context.getId(), context.getCurrentTerm(), true, operationsLog.getLastIndex());
   }

1. Wenn die Runde des Anführers geringer ist als die Runde des Nachfolgers, senden wir unserem Anführer eine Runde und ein Zeichen, dass seine Anfrage abgelehnt wurde. Sobald der Anführer eine Runde erhält, die größer ist als seine Antwort, wird er zu einem Anhänger
2. Wenn die Runde des Anführers größer ist als die Runde des Nachfolgers, setzen Sie diese Runde auf den Nachfolger.
3. Da die Anfrage vom Vorsitzenden eingegangen ist, setzen wir den Abstimmungszeitgeber zurück, unabhängig davon, ob Daten vorhanden sind oder nicht, und wenn wir kein Anhänger sind, werden wir es
4. , , , , , , . , ,
5. , . . , , - , , , , . , .
6. , . ,
7. ,
8. , , , .

  private void tryToCommit() {
       while (true) {
           //1
           int N = context.getCommitIndex() + 1;
           //2
           Supplier<Long> count = () ->
               context.getPeers().stream().map(Peer::getMatchIndex).
                       filter(matchIndex -> matchIndex >= N).count() + 1;

           //3 
           if (operationsLog.getLastIndex() >= N &&
                   operationsLog.getTerm(N).equals(context.getCurrentTerm())&&
                      count.get()>=context.getQuorum()
           )
           {
               context.setCommitIndex(N);
           } else
               return;
       }
   }

1. Wir erhalten den folgenden Index der Operation, die auf die Datenbank angewendet wird
2. Wir zählen, wie viele Follower eine Operation mit einem solchen Index in ihren Protokollen haben, und vergessen nicht, einen Leader hinzuzufügen
3. Wenn die Anzahl solcher Follower beschlussfähig ist und sich die Operation mit einem solchen Index im Protokoll des Leiters befindet und die Runde dieser Operation der aktuellen entspricht, wendet der Leiter die Operation auf die Datenbank an und erhöht den Index der zuletzt angewendeten Operation. Operationen aus der vorherigen Runde können nicht angewendet werden, da ein anderer Anführer für sie verantwortlich war und ein Konflikt entstehen könnte. Jeder Anführer wendet nur Operationen seiner aktuellen Runde an.

Fazit

Verweise

• Repository .
• Spezifikation
• Kurzbeschreibung