簡(jiǎn)介

Active Namenode與StandBy Namenode之間的綠色區域就是JournalNode,當然數量不一定只有1個(gè),作用相當于NFS共享文件系統.Active Namenode往里寫(xiě)editlog數據,StandBy再從里面讀取數據進(jìn)行同步.

JournalNodede 在hdfs架構中的角色：

源碼解析

解讀JournalNodede的原理。JN的核心功能主要包含下面幾個(gè)：

• JN啟動(dòng)
• 讀寫(xiě)editLog。
• JN之間editLog數據同步

JN 啟動(dòng)

JN的啟動(dòng)入口類(lèi)是JournalNode.java， 啟動(dòng)函數是main。在啟動(dòng)階段主要啟動(dòng)了兩個(gè)核心部件：

• JournalNodeHttpServer，主要是jn的http服務(wù)端。讀取editlog使用。
• JournalNodeRpcServer， 主要是jn的rpc服務(wù)端。主要是寫(xiě)入editlog使用當前協(xié)議。

JournalNodeHttpServer

當前類(lèi)主要是jn的http服務(wù)端，在啟動(dòng)階段最關(guān)鍵的功能是初始化http的Servlet：GetJournalEditServlet。

httpServer.addInternalServlet("getJournal", "/getJournal", GetJournalEditServlet.class, true);
httpServer.start();

核心是讀取editLog，沒(méi)有其他特別的功能。

JournalNodeRpcServer

當前類(lèi)主要是jn的rpc服務(wù)端，核心在于啟動(dòng)關(guān)鍵的rpc服務(wù)，通過(guò)下面函數初始化rpc服務(wù)端。

 this.server = new RPC.Builder(confCopy)
    .setProtocol(QJournalProtocolPB.class)
    .setInstance(service)
    .setBindAddress(bindHost)
    .setPort(addr.getPort())
    .setNumHandlers(this.handlerCount)
    .setVerbose(false)
    .build();

啟動(dòng)服務(wù)端定義的rpc協(xié)議主要包含QJournalProtocol.java，核心函數例如：

public interface QJournalProtocol {
  public static final long versionID = 1L;

  boolean isFormatted(String journalId,
                      String nameServiceId) throws IOException;

  GetJournalStateResponseProto getJournalState(String journalId,
                                               String nameServiceId)
      throws IOException;
  
  void format(String journalId, String nameServiceId,
      NamespaceInfo nsInfo, boolean force) throws IOException;

  NewEpochResponseProto newEpoch(String journalId,
                                        String nameServiceId,
                                        NamespaceInfo nsInfo,
                                        long epoch) throws IOException;
  
  public void journal(RequestInfo reqInfo,
                      long segmentTxId,
                      long firstTxnId,
                      int numTxns,
                      byte[] records) throws IOException;

  
  public void heartbeat(RequestInfo reqInfo) throws IOException;
  
  public void startLogSegment(RequestInfo reqInfo,
      long txid, int layoutVersion) throws IOException;

  public void finalizeLogSegment(RequestInfo reqInfo,
      long startTxId, long endTxId) throws IOException;

  public void purgeLogsOlderThan(RequestInfo requestInfo, long minTxIdToKeep)
      throws IOException;
  
  GetEditLogManifestResponseProto getEditLogManifest(String jid,
                                                     String nameServiceId,
                                                     long sinceTxId,
                                                     boolean inProgressOk)
      throws IOException;


  GetJournaledEditsResponseProto getJournaledEdits(String jid,
      String nameServiceId, long sinceTxId, int maxTxns) throws IOException;


  public PrepareRecoveryResponseProto prepareRecovery(RequestInfo reqInfo,
      long segmentTxId) throws IOException;

  public void acceptRecovery(RequestInfo reqInfo,
      SegmentStateProto stateToAccept, URL fromUrl) throws IOException;

  void doPreUpgrade(String journalId) throws IOException;

  public void doUpgrade(String journalId, StorageInfo sInfo) throws IOException;

  void doFinalize(String journalId,
                         String nameServiceid) throws IOException;

  Boolean canRollBack(String journalId, String nameServiceid,
                      StorageInfo storage, StorageInfo prevStorage,
                      int targetLayoutVersion) throws IOException;

  void doRollback(String journalId,
                         String nameServiceid) throws IOException;


  @Idempotent
  void discardSegments(String journalId,
                       String nameServiceId,
                       long startTxId)
      throws IOException;

  Long getJournalCTime(String journalId,
                       String nameServiceId) throws IOException;
}

讀寫(xiě)editLog

JN之間editLog數據同步