تنفيذ خوارزمية توافق RAFT لتخزين KV الموزع في جافا

وصف الحل العام

• تابع (تابع). قبول طلبات القراءة من العميل. يأخذ ضربات القلب من القائد
• مرشح (مرشح). قبول طلبات القراءة من العميل. يرسل طلبات التصويت إلى العقد الأخرى
• زعيم يقبل طلبات القراءة والكتابة. يرسل طلبات ضربات القلب إلى العقد الأخرى. يرسل إلحاق طلبات البيانات إلى العقد الأخرى.

مخزن البيانات

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

المؤقتات

• التصويت عند إجراء جولة من التصويت
• إلحاق ، ويعرف أيضًا باسم نبض القلب (إذا لم يكن هناك بيانات) ، لتكرار بيانات السجل للمتابعين ولمنع بدء جولة جديدة من التصويت.

• تصويت. لبدء جولة التصويت. كل عقدة لها فاصل زمني خاص بها ، وبعد ذلك ستحاول بدء تصويت جديد. يبدأ العد التنازلي من جديد عند تلقي نبضات من الزعيم.
• نبض القلب. لإرسال طلب للمتابعين من قبل قائد الإلحاق. إذا لم تحصل العقدة على نبضات قلب وانتهت موقت التصويت ، فإنها تصبح مرشحًا وتبدأ الانتخابات ، وتزيد عدد جولة التصويت وترسل طلبات التصويت إلى العقد الأخرى. إذا جمعت العقدة غالبية الأصوات ، فإنها تصبح الزعيم وتبدأ في إرسال نبض القلب.

الحالة الحالية للعقدة

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

التصويت

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

إرسال طلب التصويت

//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. تعيين حالة "المرشح". ارفع الرقم وصوت لأنفسنا.
2. , ( ). - , , heartbeat .
3. - , . , , -.
4. , . , heartbeat .
5. العقدة صوتت لنا! نقوم بزيادة عدد العقد التي تدلي بصوتها لنا ونصلح أن هذه العقدة صوتت لنا.
6. صوتنا ليس لنا ، نحن نعتقد أيضا.
7. إذا تم جمع النصاب القانوني وفازت العقدة في الانتخابات ، فإننا نحدد وضع "القائد". خلاف ذلك ، نصبح تابعين وننتظر.

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

       );
   }

معالجة طلب التصويت

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. إذا كانت الجولة الحالية من العقدة أكبر من جولة المرشح ، فإننا نرفض ، لأن هذا طلب لبعض العقدة المتأخرة ، والتي ، على ما يبدو ، كانت خارج الكتلة لبعض الوقت وبدأت في إجراء الانتخابات لأنها لم تر الزعيم الحالي.
2. , , , , , , ; . — .
3. ,
4. . , , , , .
5. بنتيجة إيجابية ، نثبت حقيقة أن العقدة شاركت في الانتخابات وأدلت بصوت للمرشح.
6. أرسل النتيجة مرة أخرى إلى المرشح

تكرار

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

إرسال طلب تغيير البيانات

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. البيانات الوصفية للمتابعين
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. نكرر الطلب حتى نتلقى استجابة من جميع المتابعين بأن النسخ المتماثل كان ناجحًا. نظرًا لإرسال عملية واحدة لكل طلب ، فقد يستغرق الأمر عدة تكرارات لمزامنة سجلات المتابعين
2. إرسال الطلبات إلى جميع المتابعين والحصول على قائمة بالإجابات
3. نعتبر الإجابات فقط من المتابعين المتاحين
4. إذا اتضح أن جولة أحد المتابعين هي أكثر من جولة الزعيم ، فإننا نوقف كل شيء ونتحول إلى متابعين
5. إذا رد المتابع بأن كل شيء كان ناجحًا ، فإننا نقوم بتحديث البيانات الوصفية للمتابعين: نقوم بحفظ الفهرس الأخير لسجل المتابع وفهرس العملية التالية التي يتوقعها المتابع.
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. إذا كانت جولة القائد أقل من جولة المتابعين ، فإننا نرسل قائدنا جولة وإشارة إلى رفض طلبه. بمجرد أن يتلقى الزعيم جولة أكبر من رده ، سوف يتحول إلى تابع
2. إذا كانت جولة القائد أكثر من جولة المتابع ، فقم بتعيين هذه الجولة على المتابع.
3. نظرًا لأنه تم تلقي طلب من القائد ، بغض النظر عما إذا كانت هناك بيانات أم لا ، فإننا نعيد تعيين موقت التصويت ، وإذا لم نكن متابعين ، فإننا نصبح
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. نحصل على الفهرس التالي للعملية المطبقة على قاعدة البيانات
2. نحن نحسب عدد المتابعين الذين لديهم عملية بمثل هذا الفهرس في سجلاتهم ، ولا تنسوا إضافة قائد
3. إذا كان عدد هؤلاء المتابعين هو النصاب القانوني وكانت العملية بمثل هذا الفهرس في سجل القائد ، وكانت جولة هذه العملية معادلة للعملية الحالية ، عندئذٍ يطبق القائد العملية على قاعدة البيانات ويزيد من فهرس آخر عملية تم تطبيقها. لا يمكن تطبيق العمليات من الجولة السابقة ، لأن قائدًا آخر كان مسؤولًا عنها وقد ينشأ صراع. يطبق كل قائد عمليات جولته الحالية فقط.

استنتاج

المراجع

• المستودع .
• تخصيص
• وصف قصير