Implémentation de l'algorithme de consensus RAFT pour le stockage KV distribué en Java

Description générale de la solution

• Suiveur (suiveur). Accepte les demandes de lecture du client. Prend un battement de coeur du leader
• Candidat (candidat). Accepte les demandes de lecture du client. Envoie des demandes de vote à d'autres nœuds
• Chef Accepte les demandes de lecture et d'écriture. Envoie des requêtes de pulsation à d'autres nœuds. Envoie des données de demande d'ajout à d'autres nœuds.

Stockage de données

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);
}

Minuteries

• voter lors d'un tour de scrutin
• ajouter, alias battement de cœur (s'il n'y a pas de données), pour répliquer les données du journal aux abonnés et pour empêcher le début d'un nouveau tour de scrutin.

• voter. Pour commencer un tour de scrutin. Chaque nœud a son propre intervalle, après quoi il essaiera de commencer un nouveau vote. Le compte à rebours recommence lorsque vous recevez un battement de cœur du leader.
• battement de coeur. Pour envoyer une demande aux abonnés par le leader de l'ajout. Si le nœud ne reçoit pas de battement de cœur et que le délai de vote a expiré, il devient candidat et déclenche des élections, augmente le nombre de tours de scrutin et envoie des demandes de vote à d'autres nœuds. Si le nœud recueille la majorité des votes, il devient le leader et commence à envoyer des battements de cœur.

L'état actuel du nœud

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(); //     
}

Vote

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

Envoi d'une demande de vote

//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. Définissez le statut de «Candidat». Augmentez le nombre de tours et votez pour nous-mêmes.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. Le nœud a voté pour nous! Nous augmentons le nombre de nœuds votant pour nous et corrigeons que ce nœud a voté pour nous.
6. Voté non pour nous, nous croyons aussi.
7. Si le quorum est collecté et que le nœud a remporté l'élection, nous établissons le statut de «Leader». Sinon, nous devenons un suiveur et attendons.

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

       );
   }

Traitement des demandes de vote

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. Si le tour du nœud actuel est plus grand que le tour du candidat, nous refusons, car il s'agit d'une demande de nœud en retard, qui, apparemment, était en dehors du cluster depuis un certain temps et a commencé la procédure électorale parce qu'il n'a pas vu le chef actuel.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. Avec un résultat positif, nous corrigeons le fait que le nœud a participé aux élections et a voté pour le candidat.
6. Renvoyer le résultat au candidat

La réplication

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);
}

Soumettre une demande de modification des données

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. Métadonnées du suiveur
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. Nous répétons la demande jusqu'à ce que nous recevions une réponse de tous les abonnés que la réplication a réussi. Étant donné qu'une opération est envoyée par demande, la synchronisation des journaux des abonnés peut prendre plusieurs itérations.
2. Envoyez des demandes à tous les abonnés et obtenez une liste de réponses
3. Nous considérons les réponses uniquement des abonnés disponibles
4. S'il s'avère que le tour d'un des suiveurs est supérieur au tour du leader, nous arrêtons tout et devenons un suiveur
5. Si le suiveur répond que tout a réussi, nous mettons à jour les métadonnées du suiveur: nous enregistrons le dernier index du journal du suiveur et l'index de la prochaine opération attendue par le suiveur.
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. Si le tour du leader est inférieur au tour du suiveur, alors nous envoyons à notre chef un tour et un signe que sa demande a été rejetée. Dès que le leader reçoit un round plus grand que le sien en réponse, il se transforme en suiveur
2. Si le tour du leader est supérieur au tour du suiveur, définissez ce tour sur le suiveur.
3. Depuis que la demande a été reçue du leader, qu'il y ait des données ou non, nous remettons à zéro la minuterie de vote et, si nous n'étions pas un suiveur, nous le devenons
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. Nous obtenons l'index suivant de l'opération appliquée à la base de données
2. Nous comptons combien de followers ont une opération avec un tel index dans leurs logs, et n'oubliez pas d'ajouter un leader
3. Si le nombre de ces abonnés est le quorum et que l'opération avec un tel index est dans le journal du leader, et que le cycle de cette opération est équivalent à celui en cours, le leader applique l'opération à la base de données et augmente l'index de la dernière opération appliquée. Les opérations du cycle précédent ne peuvent pas être appliquées, car un autre chef en était responsable et un conflit pourrait survenir. Chaque chef applique des opérations uniquement de son tour en cours.

Conclusion

Références

• Dépôt .
• spécification
• Brève description