framework/SpringCloud/Nacos/Nacos源码

Nacos对Raft算法的实现

必须懂raft算法， nacos使用这套算法是由客户端 请求参数决定，nacos自带内存短暂存储AP和raft算法的CP

package com.alibaba.nacos.naming.controllers;

    public static final String NACOS_SERVER_VERSION = "/v1";
    
    public static final String DEFAULT_NACOS_NAMING_CONTEXT = NACOS_SERVER_VERSION + "/ns";

/v1/ns/raft
/nacos/v1/ns/raft   

@RestController
@RequestMapping({UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft",
        UtilsAndCommons.NACOS_SERVER_CONTEXT + UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft"})
public class RaftController {
    
    
    @Autowired
    private RaftConsistencyServiceImpl raftConsistencyService;

    //服务的管理者
    @Autowired
    private ServiceManager serviceManager;

    @Autowired
    private RaftCore raftCore;
    
}

投票

@PostMapping("/vote")
 public JsonNode vote(HttpServletRequest request, HttpServletResponse response) throws Exception {

     // 通过JackJosn工具类把请求包装成一个RaftPeer对象交给raftCOre进行处理
     RaftPeer peer = raftCore.receivedVote(JacksonUtils.toObj(WebUtils.required(request, "vote"), RaftPeer.class));

     return JacksonUtils.transferToJsonNode(peer);
 }

receivedVote

 /**
     * Received vote.
     *
     * @param remote remote raft peer of vote information
     * @return self-peer information
     */
//  @Autowired
//    private RaftPeerSet peers;
    public synchronized RaftPeer receivedVote(RaftPeer remote) {
        if (!peers.contains(remote)) {
            throw new IllegalStateException("can not find peer: " + remote.ip);
        }
        //NetUtils.localServer() 获取本地的服务器地址和端口
        //然后通过 peers获取本地 当前推荐leader
        RaftPeer local = peers.get(NetUtils.localServer());
        if (remote.term.get() <= local.term.get()) {
            //如果远程的届数比本地推荐leader的届数低或者相等，那么返回本地的
            
            String msg = "received illegitimate vote" + ", voter-term:" + remote.term + ", votee-term:" + local.term;
            
            Loggers.RAFT.info(msg);
            if (StringUtils.isEmpty(local.voteFor)) {
                local.voteFor = local.ip;
            }
            
            return local;
        }
        // 如果远程的届数高于自己本地的，那么重新设置
        /*
            5秒
    public static final long LEADER_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(15L);
    //0 ~5秒
            public volatile long leaderDueMs = RandomUtils.nextLong(0, GlobalExecutor.LEADER_TIMEOUT_MS);
            //重新设置选举leader间隔时间
          public void resetLeaderDue() {
        leaderDueMs = GlobalExecutor.LEADER_TIMEOUT_MS + RandomUtils.nextLong(0, GlobalExecutor.RANDOM_MS);
    }
        
        
        */
        local.resetLeaderDue();
        //本地状态变为 跟随
        local.state = RaftPeer.State.FOLLOWER;
        // 投票的ip为远程的ip
        local.voteFor = remote.ip;
        //届数设置成远程的
        local.term.set(remote.term.get());
        
        Loggers.RAFT.info("vote {} as leader, term: {}", remote.ip, remote.term);
       //返回 
        return local;
    }

心跳

@PostMapping("/beat")
  public JsonNode beat(HttpServletRequest request, HttpServletResponse response) throws Exception {

      String entity = new String(IoUtils.tryDecompress(request.getInputStream()), StandardCharsets.UTF_8);
      String value = URLDecoder.decode(entity, "UTF-8");
      value = URLDecoder.decode(value, "UTF-8");

      JsonNode json = JacksonUtils.toObj(value);

      RaftPeer peer = raftCore.receivedBeat(JacksonUtils.toObj(json.get("beat").asText()));

      return JacksonUtils.transferToJsonNode(peer);
  }

receivedBeat

/**
   * Received beat from leader. // TODO split method to multiple smaller method.
   * 这是一个待修改的方法，TODO有说明
   * @param beat beat information from leader
   * @return self-peer information
   * @throws Exception any exception during handle
   */
  public RaftPeer receivedBeat(JsonNode beat) throws Exception {
      // 通过 peers获取本地的投票
      final RaftPeer local = peers.local();
      // 初始化一个远程投票
      final RaftPeer remote = new RaftPeer();
      // 通过收到的心跳获取 集群投票信息
      JsonNode peer = beat.get("peer");
      
      //ip设置
      remote.ip = peer.get("ip").asText();
      // 状态设置
      remote.state = RaftPeer.State.valueOf(peer.get("state").asText());
      // 届数设置
      remote.term.set(peer.get("term").asLong());
      // 心跳间隔设置
      remote.heartbeatDueMs = peer.get("heartbeatDueMs").asLong();
      // leader的间隔设置
      remote.leaderDueMs = peer.get("leaderDueMs").asLong();
      // 推荐leader的设置
      remote.voteFor = peer.get("voteFor").asText();
      
      if (remote.state != RaftPeer.State.LEADER) {
          //如果远程的状态不是leader，那么直接抛出错误交给springMVC拦截
          Loggers.RAFT.info("[RAFT] invalid state from master, state: {}, remote peer: {}", remote.state,
                  JacksonUtils.toJson(remote));
          throw new IllegalArgumentException("invalid state from master, state: " + remote.state);
      }
      
      if (local.term.get() > remote.term.get()) {
          //如果本地届数 比远程届数高，那么抛出错误
          Loggers.RAFT
                  .info("[RAFT] out of date beat, beat-from-term: {}, beat-to-term: {}, remote peer: {}, and leaderDueMs: {}",
                          remote.term.get(), local.term.get(), JacksonUtils.toJson(remote), local.leaderDueMs);
          throw new IllegalArgumentException(
                  "out of date beat, beat-from-term: " + remote.term.get() + ", beat-to-term: " + local.term.get());
      }
      
      if (local.state != RaftPeer.State.FOLLOWER) {
          //本地的状态如果不是跟随，那么编程跟随，推荐人改变
          Loggers.RAFT.info("[RAFT] make remote as leader, remote peer: {}", JacksonUtils.toJson(remote));
          // mk follower
          local.state = RaftPeer.State.FOLLOWER;
          local.voteFor = remote.ip;
      }
      
      final JsonNode beatDatums = beat.get("datums");
      //重新设置心跳间隔和leader间隔
      local.resetLeaderDue();
      local.resetHeartbeatDue();
      // 把当前leader修改的消息发送给集群，并且在自己本地更新
      peers.makeLeader(remote);
      
      
      //有点长，暂时放置
      if (!switchDomain.isSendBeatOnly()) {
          
          Map<String, Integer> receivedKeysMap = new HashMap<>(datums.size());
          
          for (Map.Entry<String, Datum> entry : datums.entrySet()) {
              receivedKeysMap.put(entry.getKey(), 0);
          }
          
          // now check datums
          List<String> batch = new ArrayList<>();
          
          int processedCount = 0;
          if (Loggers.RAFT.isDebugEnabled()) {
              Loggers.RAFT
                      .debug("[RAFT] received beat with {} keys, RaftCore.datums' size is {}, remote server: {}, term: {}, local term: {}",
                              beatDatums.size(), datums.size(), remote.ip, remote.term, local.term);
          }
          for (Object object : beatDatums) {
              processedCount = processedCount + 1;
              
              JsonNode entry = (JsonNode) object;
              String key = entry.get("key").asText();
              final String datumKey;
              
              if (KeyBuilder.matchServiceMetaKey(key)) {
                  datumKey = KeyBuilder.detailServiceMetaKey(key);
              } else if (KeyBuilder.matchInstanceListKey(key)) {
                  datumKey = KeyBuilder.detailInstanceListkey(key);
              } else {
                  // ignore corrupted key:
                  continue;
              }
              
              long timestamp = entry.get("timestamp").asLong();
              
              receivedKeysMap.put(datumKey, 1);
              
              try {
                  if (datums.containsKey(datumKey) && datums.get(datumKey).timestamp.get() >= timestamp
                          && processedCount < beatDatums.size()) {
                      continue;
                  }
                  
                  if (!(datums.containsKey(datumKey) && datums.get(datumKey).timestamp.get() >= timestamp)) {
                      batch.add(datumKey);
                  }
                  
                  if (batch.size() < 50 && processedCount < beatDatums.size()) {
                      continue;
                  }
                  
                  String keys = StringUtils.join(batch, ",");
                  
                  if (batch.size() <= 0) {
                      continue;
                  }
                  
                  Loggers.RAFT.info("get datums from leader: {}, batch size is {}, processedCount is {}"
                                  + ", datums' size is {}, RaftCore.datums' size is {}", getLeader().ip, batch.size(),
                          processedCount, beatDatums.size(), datums.size());
                  
                  // update datum entry
                  String url = buildUrl(remote.ip, API_GET) + "?keys=" + URLEncoder.encode(keys, "UTF-8");
                  HttpClient.asyncHttpGet(url, null, null, new AsyncCompletionHandler<Integer>() {
                      @Override
                      public Integer onCompleted(Response response) throws Exception {
                          if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                              return 1;
                          }
                          
                          List<JsonNode> datumList = JacksonUtils
                                  .toObj(response.getResponseBody(), new TypeReference<List<JsonNode>>() {
                                  });
                          
                          for (JsonNode datumJson : datumList) {
                              Datum newDatum = null;
                              OPERATE_LOCK.lock();
                              try {
                                  
                                  Datum oldDatum = getDatum(datumJson.get("key").asText());
                                  
                                  if (oldDatum != null && datumJson.get("timestamp").asLong() <= oldDatum.timestamp
                                          .get()) {
                                      Loggers.RAFT
                                              .info("[NACOS-RAFT] timestamp is smaller than that of mine, key: {}, remote: {}, local: {}",
                                                      datumJson.get("key").asText(),
                                                      datumJson.get("timestamp").asLong(), oldDatum.timestamp);
                                      continue;
                                  }
                                  
                                  if (KeyBuilder.matchServiceMetaKey(datumJson.get("key").asText())) {
                                      Datum<Service> serviceDatum = new Datum<>();
                                      serviceDatum.key = datumJson.get("key").asText();
                                      serviceDatum.timestamp.set(datumJson.get("timestamp").asLong());
                                      serviceDatum.value = JacksonUtils
                                              .toObj(datumJson.get("value").toString(), Service.class);
                                      newDatum = serviceDatum;
                                  }
                                  
                                  if (KeyBuilder.matchInstanceListKey(datumJson.get("key").asText())) {
                                      Datum<Instances> instancesDatum = new Datum<>();
                                      instancesDatum.key = datumJson.get("key").asText();
                                      instancesDatum.timestamp.set(datumJson.get("timestamp").asLong());
                                      instancesDatum.value = JacksonUtils
                                              .toObj(datumJson.get("value").toString(), Instances.class);
                                      newDatum = instancesDatum;
                                  }
                                  
                                  if (newDatum == null || newDatum.value == null) {
                                      Loggers.RAFT.error("receive null datum: {}", datumJson);
                                      continue;
                                  }
                                  
                                  raftStore.write(newDatum);
                                  
                                  datums.put(newDatum.key, newDatum);
                                  notifier.addTask(newDatum.key, ApplyAction.CHANGE);
                                  
                                  local.resetLeaderDue();
                                  
                                  if (local.term.get() + 100 > remote.term.get()) {
                                      getLeader().term.set(remote.term.get());
                                      local.term.set(getLeader().term.get());
                                  } else {
                                      local.term.addAndGet(100);
                                  }
                                  
                                  raftStore.updateTerm(local.term.get());
                                  
                                  Loggers.RAFT.info("data updated, key: {}, timestamp: {}, from {}, local term: {}",
                                          newDatum.key, newDatum.timestamp, JacksonUtils.toJson(remote), local.term);
                                  
                              } catch (Throwable e) {
                                  Loggers.RAFT
                                          .error("[RAFT-BEAT] failed to sync datum from leader, datum: {}", newDatum,
                                                  e);
                              } finally {
                                  OPERATE_LOCK.unlock();
                              }
                          }
                          TimeUnit.MILLISECONDS.sleep(200);
                          return 0;
                      }
                  });
                  
                  batch.clear();
                  
              } catch (Exception e) {
                  Loggers.RAFT.error("[NACOS-RAFT] failed to handle beat entry, key: {}", datumKey);
              }
              
          }
          
          List<String> deadKeys = new ArrayList<>();
          for (Map.Entry<String, Integer> entry : receivedKeysMap.entrySet()) {
              if (entry.getValue() == 0) {
                  deadKeys.add(entry.getKey());
              }
          }
          
          for (String deadKey : deadKeys) {
              try {
                  deleteDatum(deadKey);
              } catch (Exception e) {
                  Loggers.RAFT.error("[NACOS-RAFT] failed to remove entry, key={} {}", deadKey, e);
              }
          }
          
      }
      
      return local;
  }

makeLeader

public RaftPeer makeLeader(RaftPeer candidate) {
    //只有当前leader不等于cadicate，才进行更新
      if (!Objects.equals(leader, candidate)) {
          leader = candidate;
          ApplicationUtils.publishEvent(new MakeLeaderEvent(this, leader, local()));
          Loggers.RAFT
                  .info("{} has become the LEADER, local: {}, leader: {}", leader.ip, JacksonUtils.toJson(local()),
                          JacksonUtils.toJson(leader));
      }
      
      for (final RaftPeer peer : peers.values()) {
          // 获取集群同伴，设置发送请求的url，异步发送请求
          Map<String, String> params = new HashMap<>(1);
          if (!Objects.equals(peer, candidate) && peer.state == RaftPeer.State.LEADER) {
              //只会给以前状态为 leader的同伴推leader修改信息
              try {
                  String url = RaftCore.buildUrl(peer.ip, RaftCore.API_GET_PEER);
                  HttpClient.asyncHttpGet(url, null, params, new AsyncCompletionHandler<Integer>() {
                      @Override
                      public Integer onCompleted(Response response) throws Exception {
                          if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                              Loggers.RAFT
                                      .error("[NACOS-RAFT] get peer failed: {}, peer: {}", response.getResponseBody(),
                                              peer.ip);
                              peer.state = RaftPeer.State.FOLLOWER;
                              return 1;
                          }
                          //就是放入本地缓存
                          update(JacksonUtils.toObj(response.getResponseBody(), RaftPeer.class));
                          
                          return 0;
                      }
                  });
              } catch (Exception e) {
                  peer.state = RaftPeer.State.FOLLOWER;
                  Loggers.RAFT.error("[NACOS-RAFT] error while getting peer from peer: {}", peer.ip);
              }
          }
      }
      
      return update(candidate);
  }

集群更新

 @GetMapping("/peer")
  public JsonNode getPeer(HttpServletRequest request, HttpServletResponse response) {
      List<RaftPeer> peers = raftCore.getPeers();
      RaftPeer peer = null;

      for (RaftPeer peer1 : peers) {
          // 把本地的peer对象拿到
          if (StringUtils.equals(peer1.ip, NetUtils.localServer())) {
              peer = peer1;
          }
      }

      if (peer == null) {
          peer = new RaftPeer();
          peer.ip = NetUtils.localServer();
      }
//然后返回
      return JacksonUtils.transferToJsonNode(peer);
  }

RaftCore

package com.alibaba.nacos.naming.consistency.persistent.raft;
核心实现
@DependsOn("ProtocolManager")
@Component
public class RaftCore {
    
    public static final String API_VOTE = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/vote";
    
    public static final String API_BEAT = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/beat";
    
    public static final String API_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
    
    public static final String API_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
    
    public static final String API_GET = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum";
    
    public static final String API_ON_PUB = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
    
    public static final String API_ON_DEL = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/datum/commit";
    
    public static final String API_GET_PEER = UtilsAndCommons.NACOS_NAMING_CONTEXT + "/raft/peer";
    
    
      private final ScheduledExecutorService executor = ExecutorFactory.Managed
            .newSingleScheduledExecutorService(ClassUtils.getCanonicalName(NamingApp.class),
                    new NameThreadFactory("com.alibaba.nacos.naming.raft.notifier"));
    
    
}

init()

@PostConstruct //这个注解会在bean被填充后，执行到一个后置处理器时执行
  public void init() throws Exception {
      
      Loggers.RAFT.info("initializing Raft sub-system");
      // 时间发布器
      executor.submit(notifier);
      
      final long start = System.currentTimeMillis();
      
      //加载配置
      raftStore.loadDatums(notifier, datums);
      // 设置届数
      setTerm(NumberUtils.toLong(raftStore.loadMeta().getProperty("term"), 0L));
      
      Loggers.RAFT.info("cache loaded, datum count: {}, current term: {}", datums.size(), peers.getTerm());
      
      while (true) {
          if (notifier.tasks.size() <= 0) {
              break;
          }
          Thread.sleep(1000L);
      }
      
      initialized = true;
      
      Loggers.RAFT.info("finish to load data from disk, cost: {} ms.", (System.currentTimeMillis() - start));
      //注册两个 执行体
      /*
      
       public static void registerMasterElection(Runnable runnable) {
      NAMING_TIMER_EXECUTOR.scheduleAtFixedRate(runnable, 0, TICK_PERIOD_MS, TimeUnit.MILLISECONDS);
  }
          public static void registerHeartbeat(Runnable runnable) {
      NAMING_TIMER_EXECUTOR.scheduleWithFixedDelay(runnable, 0, TICK_PERIOD_MS, TimeUnit.MILLISECONDS);
  }
      */
      GlobalExecutor.registerMasterElection(new MasterElection());
      GlobalExecutor.registerHeartbeat(new HeartBeat());
      
      Loggers.RAFT.info("timer started: leader timeout ms: {}, heart-beat timeout ms: {}",
              GlobalExecutor.LEADER_TIMEOUT_MS, GlobalExecutor.HEARTBEAT_INTERVAL_MS);
  }

Notifier

public class Notifier implements Runnable {
       
       private ConcurrentHashMap<String, String> services = new ConcurrentHashMap<>(10 * 1024);
       
       private BlockingQueue<Pair> tasks = new LinkedBlockingQueue<>(1024 * 1024);
       
       /**
        * Add notify task.
        *
        * @param datumKey datum key
        * @param action   action of datum
        */
       public void addTask(String datumKey, ApplyAction action) {
           
           if (services.containsKey(datumKey) && action == ApplyAction.CHANGE) {
               return;
           }
           if (action == ApplyAction.CHANGE) {
               services.put(datumKey, StringUtils.EMPTY);
           }
           
           Loggers.RAFT.info("add task {}", datumKey);
           
           tasks.add(Pair.with(datumKey, action));
       }
       
       public int getTaskSize() {
           return tasks.size();
       }
       
       @Override
       public void run() {
           Loggers.RAFT.info("raft notifier started");
           
           while (true) {
               try {
                   
                   Pair pair = tasks.take();
                   
                   if (pair == null) {
                       continue;
                   }
                   
                   String datumKey = (String) pair.getValue0();
                   ApplyAction action = (ApplyAction) pair.getValue1();
                   
                   services.remove(datumKey);
                   
                   Loggers.RAFT.info("remove task {}", datumKey);
                   
                   int count = 0;
                   
                   if (listeners.containsKey(KeyBuilder.SERVICE_META_KEY_PREFIX)) {
                       
                       if (KeyBuilder.matchServiceMetaKey(datumKey) && !KeyBuilder.matchSwitchKey(datumKey)) {
                           
                           for (RecordListener listener : listeners.get(KeyBuilder.SERVICE_META_KEY_PREFIX)) {
                               try {
                                   if (action == ApplyAction.CHANGE) {
                                       listener.onChange(datumKey, getDatum(datumKey).value);
                                   }
                                   
                                   if (action == ApplyAction.DELETE) {
                                       listener.onDelete(datumKey);
                                   }
                               } catch (Throwable e) {
                                   Loggers.RAFT
                                           .error("[NACOS-RAFT] error while notifying listener of key: {}", datumKey,
                                                   e);
                               }
                           }
                       }
                   }
                   
                   if (!listeners.containsKey(datumKey)) {
                       continue;
                   }
                   
                   for (RecordListener listener : listeners.get(datumKey)) {
                       
                       count++;
                       
                       try {
                           if (action == ApplyAction.CHANGE) {
                               listener.onChange(datumKey, getDatum(datumKey).value);
                               continue;
                           }
                           
                           if (action == ApplyAction.DELETE) {
                               listener.onDelete(datumKey);
                               continue;
                           }
                       } catch (Throwable e) {
                           Loggers.RAFT.error("[NACOS-RAFT] error while notifying listener of key: {}", datumKey, e);
                       }
                   }
                   
                   if (Loggers.RAFT.isDebugEnabled()) {
                       Loggers.RAFT.debug("[NACOS-RAFT] datum change notified, key: {}, listener count: {}", datumKey,
                               count);
                   }
               } catch (Throwable e) {
                   Loggers.RAFT.error("[NACOS-RAFT] Error while handling notifying task", e);
               }
           }
       }
   }

MasterElection

public class MasterElection implements Runnable {
      
      @Override
      public void run() {
          try {
              //没准备号就return
              if (!peers.isReady()) {
                  return;
              }
             
              RaftPeer local = peers.local();
              local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;
               // 做时间校验
              if (local.leaderDueMs > 0) {
                  return;
              }
              // 超时了，重新设置心跳和 推荐发送
              // reset timeout
              local.resetLeaderDue();
              local.resetHeartbeatDue();
              
              sendVote();
          } catch (Exception e) {
              Loggers.RAFT.warn("[RAFT] error while master election {}", e);
          }
          
      }
      
      private void sendVote() {
          
          RaftPeer local = peers.get(NetUtils.localServer());
          Loggers.RAFT.info("leader timeout, start voting,leader: {}, term: {}", JacksonUtils.toJson(getLeader()),
                  local.term);
          
          peers.reset();
          //届数加加
          local.term.incrementAndGet();
          local.voteFor = local.ip;//投自己
          local.state = RaftPeer.State.CANDIDATE;
          
          Map<String, String> params = new HashMap<>(1);
          params.put("vote", JacksonUtils.toJson(local));
          for (final String server : peers.allServersWithoutMySelf()) {
              final String url = buildUrl(server, API_VOTE);
              try {
                  /vote  投票api
                  HttpClient.asyncHttpPost(url, null, params, new AsyncCompletionHandler<Integer>() {
                      @Override
                      public Integer onCompleted(Response response) throws Exception {
                          if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                              Loggers.RAFT
                                      .error("NACOS-RAFT vote failed: {}, url: {}", response.getResponseBody(), url);
                              return 1;
                          }
                          
                          RaftPeer peer = JacksonUtils.toObj(response.getResponseBody(), RaftPeer.class);
                          
                          Loggers.RAFT.info("received approve from peer: {}", JacksonUtils.toJson(peer));
                          // 决定leader，看有没有半数同伴响应投自己
                          
                          /*
                          
                          由于上面的届数加加1
                          说明如果没有超时，那么不会投，也不会加加
                          一旦超时，同伴有可能超时，同伴也会加加，只要同伴原子加加成功，那么届数一样不会进行处理
                          
                          */
                          peers.decideLeader(peer);
                          
                          return 0;
                      }
                  });
              } catch (Exception e) {
                  Loggers.RAFT.warn("error while sending vote to server: {}", server);
              }
          }
      }
  }

HeartBeat

public class HeartBeat implements Runnable {
     
     @Override
     public void run() {
         try {
             
             if (!peers.isReady()) {
                 return;
             }
             
             RaftPeer local = peers.local();
             local.heartbeatDueMs -= GlobalExecutor.TICK_PERIOD_MS;
             if (local.heartbeatDueMs > 0) {
                 return;
             }
             
             local.resetHeartbeatDue();
             
             sendBeat();
         } catch (Exception e) {
             Loggers.RAFT.warn("[RAFT] error while sending beat {}", e);
         }
         
     }
     
     private void sendBeat() throws IOException, InterruptedException {
         RaftPeer local = peers.local();
         if (ApplicationUtils.getStandaloneMode() || local.state != RaftPeer.State.LEADER) {
             //单机或者不是leader不要发送心跳
             return;
         }
         if (Loggers.RAFT.isDebugEnabled()) {
             Loggers.RAFT.debug("[RAFT] send beat with {} keys.", datums.size());
         }
         
         local.resetLeaderDue();
         //下面 一大堆压缩操作，构建数据
         // build data
         ObjectNode packet = JacksonUtils.createEmptyJsonNode();
         packet.replace("peer", JacksonUtils.transferToJsonNode(local));
         
         ArrayNode array = JacksonUtils.createEmptyArrayNode();
         
         if (switchDomain.isSendBeatOnly()) {
             Loggers.RAFT.info("[SEND-BEAT-ONLY] {}", String.valueOf(switchDomain.isSendBeatOnly()));
         }
         
         if (!switchDomain.isSendBeatOnly()) {
             for (Datum datum : datums.values()) {
                 
                 ObjectNode element = JacksonUtils.createEmptyJsonNode();
                 
                 if (KeyBuilder.matchServiceMetaKey(datum.key)) {
                     element.put("key", KeyBuilder.briefServiceMetaKey(datum.key));
                 } else if (KeyBuilder.matchInstanceListKey(datum.key)) {
                     element.put("key", KeyBuilder.briefInstanceListkey(datum.key));
                 }
                 element.put("timestamp", datum.timestamp.get());
                 
                 array.add(element);
             }
         }
         
         packet.replace("datums", array);
         // broadcast
         Map<String, String> params = new HashMap<String, String>(1);
         params.put("beat", JacksonUtils.toJson(packet));
         
         String content = JacksonUtils.toJson(params);
         
         ByteArrayOutputStream out = new ByteArrayOutputStream();
         GZIPOutputStream gzip = new GZIPOutputStream(out);
         gzip.write(content.getBytes(StandardCharsets.UTF_8));
         gzip.close();
         
         byte[] compressedBytes = out.toByteArray();
         String compressedContent = new String(compressedBytes, StandardCharsets.UTF_8);
         
         if (Loggers.RAFT.isDebugEnabled()) {
             Loggers.RAFT.debug("raw beat data size: {}, size of compressed data: {}", content.length(),
                     compressedContent.length());
         }
         
         for (final String server : peers.allServersWithoutMySelf()) {
             try {
                 final String url = buildUrl(server, API_BEAT);
                 if (Loggers.RAFT.isDebugEnabled()) {
                     Loggers.RAFT.debug("send beat to server " + server);
                 }
                 HttpClient.asyncHttpPostLarge(url, null, compressedBytes, new AsyncCompletionHandler<Integer>() {
                     @Override
                     public Integer onCompleted(Response response) throws Exception {
                         if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                             Loggers.RAFT.error("NACOS-RAFT beat failed: {}, peer: {}", response.getResponseBody(),
                                     server);
                             MetricsMonitor.getLeaderSendBeatFailedException().increment();
                             return 1;
                         }
                         //更新
                         peers.update(JacksonUtils.toObj(response.getResponseBody(), RaftPeer.class));
                         if (Loggers.RAFT.isDebugEnabled()) {
                             Loggers.RAFT.debug("receive beat response from: {}", url);
                         }
                         return 0;
                     }
                     
                     @Override
                     public void onThrowable(Throwable t) {
                         Loggers.RAFT.error("NACOS-RAFT error while sending heart-beat to peer: {} {}", server, t);
                         MetricsMonitor.getLeaderSendBeatFailedException().increment();
                     }
                 });
             } catch (Exception e) {
                 Loggers.RAFT.error("error while sending heart-beat to peer: {} {}", server, e);
                 MetricsMonitor.getLeaderSendBeatFailedException().increment();
             }
         }
         
     }
 }

日志备份

publish

@PostMapping("/datum")
  public String publish(HttpServletRequest request, HttpServletResponse response) throws Exception {

      response.setHeader("Content-Type", "application/json; charset=" + getAcceptEncoding(request));
      response.setHeader("Cache-Control", "no-cache");
      response.setHeader("Content-Encode", "gzip");

      String entity = IoUtils.toString(request.getInputStream(), "UTF-8");
      String value = URLDecoder.decode(entity, "UTF-8");
      JsonNode json = JacksonUtils.toObj(value);

      String key = json.get("key").asText();
      if (KeyBuilder.matchInstanceListKey(key)) {
          raftConsistencyService.put(key, JacksonUtils.toObj(json.get("value").toString(), Instances.class));
          return "ok";
      }

      if (KeyBuilder.matchSwitchKey(key)) {
          raftConsistencyService.put(key, JacksonUtils.toObj(json.get("value").toString(), SwitchDomain.class));
          return "ok";
      }

      if (KeyBuilder.matchServiceMetaKey(key)) {
          raftConsistencyService.put(key, JacksonUtils.toObj(json.get("value").toString(), Service.class));
          return "ok";
      }

      throw new NacosException(NacosException.INVALID_PARAM, "unknown type publish key: " + key);
  }

signalPublish

public void signalPublish(String key, Record value) throws Exception {
     
     if (!isLeader()) {
         //不是leader，消息发送给leader
         ObjectNode params = JacksonUtils.createEmptyJsonNode();
         params.put("key", key);
         params.replace("value", JacksonUtils.transferToJsonNode(value));
         Map<String, String> parameters = new HashMap<>(1);
         parameters.put("key", key);
         
         final RaftPeer leader = getLeader();
         
         raftProxy.proxyPostLarge(leader.ip, API_PUB, params.toString(), parameters);
         return;
     }
     
     try {
         OPERATE_LOCK.lock();
         final long start = System.currentTimeMillis();
         final Datum datum = new Datum();
         datum.key = key;
         datum.value = value;
         if (getDatum(key) == null) {
             datum.timestamp.set(1L);
         } else {
             datum.timestamp.set(getDatum(key).timestamp.incrementAndGet());
         }
         
         ObjectNode json = JacksonUtils.createEmptyJsonNode();
         json.replace("datum", JacksonUtils.transferToJsonNode(datum));
         json.replace("source", JacksonUtils.transferToJsonNode(peers.local()));
         
         //发布，并自己刷盘
         onPublish(datum, peers.local());
         
         final String content = json.toString();
         
         final CountDownLatch latch = new CountDownLatch(peers.majorityCount());
         for (final String server : peers.allServersIncludeMyself()) {
             if (isLeader(server)) {
                 //减去自己同意的一票
                 latch.countDown();
                 continue;
             }
             final String url = buildUrl(server, API_ON_PUB);
             HttpClient.asyncHttpPostLarge(url, Arrays.asList("key=" + key), content,
                     new AsyncCompletionHandler<Integer>() {
                         @Override
                         public Integer onCompleted(Response response) throws Exception {
                             //commit操作
                             if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                                 Loggers.RAFT
                                         .warn("[RAFT] failed to publish data to peer, datumId={}, peer={}, http code={}",
                                                 datum.key, server, response.getStatusCode());
                                 return 1;
                             }
                             // 成功一个就--
                             latch.countDown();
                             return 0;
                         }
                         
                         @Override
                         public STATE onContentWriteCompleted() {
                             return STATE.CONTINUE;
                         }
                     });
             
         }
         //一半通过返回
         if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) {
             // only majority servers return success can we consider this update success
             Loggers.RAFT.error("data publish failed, caused failed to notify majority, key={}", key);
             throw new IllegalStateException("data publish failed, caused failed to notify majority, key=" + key);
         }
         
         long end = System.currentTimeMillis();
         Loggers.RAFT.info("signalPublish cost {} ms, key: {}", (end - start), key);
     } finally {
         OPERATE_LOCK.unlock();
     }
 }

commit

 @PostMapping("/datum/commit")
  public String onPublish(HttpServletRequest request, HttpServletResponse response) throws Exception {

      response.setHeader("Content-Type", "application/json; charset=" + getAcceptEncoding(request));
      response.setHeader("Cache-Control", "no-cache");
      response.setHeader("Content-Encode", "gzip");

      String entity = IoUtils.toString(request.getInputStream(), "UTF-8");
      String value = URLDecoder.decode(entity, "UTF-8");

      JsonNode jsonObject = JacksonUtils.toObj(value);
      String key = "key";

      RaftPeer source = JacksonUtils.toObj(jsonObject.get("source").toString(), RaftPeer.class);
      JsonNode datumJson = jsonObject.get("datum");

      Datum datum = null;
      if (KeyBuilder.matchInstanceListKey(datumJson.get(key).asText())) {
          datum = JacksonUtils.toObj(jsonObject.get("datum").toString(), new TypeReference<Datum<Instances>>() {
          });
      } else if (KeyBuilder.matchSwitchKey(datumJson.get(key).asText())) {
          datum = JacksonUtils.toObj(jsonObject.get("datum").toString(), new TypeReference<Datum<SwitchDomain>>() {
          });
      } else if (KeyBuilder.matchServiceMetaKey(datumJson.get(key).asText())) {
          datum = JacksonUtils.toObj(jsonObject.get("datum").toString(), new TypeReference<Datum<Service>>() {
          });
      }
// 因为不是leader所以只提交，不发布
      raftConsistencyService.onPut(datum, source);
      return "ok";
  }

赏

谢谢你请我吃糖果

支付宝

微信

• 本文作者： 忘忧症
• 本文链接： https://NepenthesZGW.github.io/2020/09/08/framework/SpringCloud/Nacos/Nacos源码/
• 版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！