Hdfs有两个抽象类提供对EditLog的读出和写回：EditLogOutputStream(以下简称ELOS)和EditLogInputStream(以下简称ELIS)。同时还有一个JournalManager接口，负责管理EditLog的可靠存取。它的实现包括QJM(QuorumJournalManager)和BKJM(BookKeeperJournalManager)。

写日志

对于hdfs而言，主节点写的每一个日志对象为BK的entry，entry的集合组成一个ledger，每一个日志段对应一个ledger，相同日志段追加edits即为向ledger追加entry。Ledger有一个递增的ledgerId，entry也有递增的entryId，每个entryId对应一个txId。

ELOS使用write()将FSEditLogOp往外写，对应的BookKeeperEditLogOutputStream的实现为：

@Overridepublic void write(FSEditLogOp op) throws IOException {writer.writeOp(op);if (bufCurrent.getLength() > transmissionThreshold) {transmit();}}

BookKeeperEditLogOutputStream内部有一个buffer，每次调用write()写FSEditLogOp的时候，会由一个Writer将此次FSEditLogOp写入buffer，当buffer长度达到门槛值时，进行transmit操作：把buffer里的editLog发送到BK上，代码如下：

/*** Transmit the current buffer to bookkeeper.* Synchronised at the FSEditLog level. #write() and #setReadyToFlush()* are never called at the same time.*/private void transmit() throws IOException {if (!transmitResult.compareAndSet(BKException.Code.OK,BKException.Code.OK)) {throw new IOException("Trying to write to an errored stream;"+ " Error code : (" + transmitResult.get()+ ") " + BKException.getMessage(transmitResult.get()));}if (bufCurrent.getLength() > 0) {byte[] entry = Arrays.copyOf(bufCurrent.getData(),bufCurrent.getLength());lh.asyncAddEntry(entry, this, null);bufCurrent.reset();outstandingRequests.incrementAndGet();}}

lh为BK的LedgerHandle，asyncAddEntry方法异步将entry写往一个open状态的ledger。这就是一个简单的把Editlog写往BK的过程。

BKJM简单写的代码如下：

public void testSimpleWrite() throws Exception {NamespaceInfo nsi = newNSInfo();BookKeeperJournalManager bkjm = new BookKeeperJournalManager(conf,BKJMUtil.createJournalURI("/hdfsjournal-simplewrite"), nsi);bkjm.format(nsi);EditLogOutputStream out = bkjm.startLogSegment(1);for (long i = 1 ; i <= 100; i++) {FSEditLogOp op = FSEditLogTestUtil.getNoOpInstance();op.setTransactionId(i);out.write(op);}out.close();bkjm.finalizeLogSegment(1, 100);String zkpath = bkjm.finalizedLedgerZNode(1, 100);}

BKJM的startLogSegment(txId)将产生一个新的ledger，对应一个新的日志段，该日志段状态为接收写入日志的状态。创建ledger之前有一些校验工作

if (txId <= maxTxId.get()) {throw new IOException("We've already seen " + txId+ ". A new stream cannot be created with it");}try {String existingInprogressNode = ci.read();if (null != existingInprogressNode&& zkc.exists(existingInprogressNode, false) != null) {throw new IOException("Inprogress node already exists");}if (currentLedger != null) {// bookkeeper errored on last stream, clean up ledgercurrentLedger.close();}currentLedger = bkc.createLedger(ensembleSize, quorumSize,BookKeeper.DigestType.MAC,digestpw.getBytes());} catch (BKException bke) {throw new IOException("Error creating ledger", bke);} catch (KeeperException ke) {throw new IOException("Error in zookeeper while creating ledger", ke);} catch (InterruptedException ie) {Thread.currentThread().interrupt();throw new IOException("Interrupted creating ledger", ie);}

Ledger的创建还对应一个新的EditLogLedgerMetadata，该类记录这个日志段的元信息，包括zkPath，ledgerId，开始和结束txId等，在读取ledger里的日志内容的时候需要这些元数据信息。

BKJM的finalizeLogSegment()将文件由正在写入日志的状态转化为不接收写日志的状态。BKJM会create ledger，delete ledger，open ledger，这里的ledger即LedgerHandler类，它对每个ledger entry进行读写操作。

写日志总体流程

ZK作为BK的元数据服务器，里面存储了哪些bookie服务是可用的，同时也记录了目前系统有哪些ledger，及其ledger相关信息，如该ledger数据存储在哪些机器上，及其该ledger起始，结束entryid等。Bookie节点存储实际的数据，及其数据的读写服务。

写操作由主节点来完成，当主节点调用setReadyToFlush操作，会调用RPC同时向N（N=quorums）个bookie节点写，flush异步等待响应。

主节点对bk的操作，其实就是对ledger的操作，在开始向bk服务写数据前，首先需要打开ledger，打开ledger就会与配置的所有bookie节点建立连接；打开连接后，数据以entry为单位以RR算法选择向N（N=quorums）个bookie节点写entry数据，并且异步地等待结果返回，有任何一个bookie写入失败，则需要重新选择一个bookie写入失败的副本。

当bookie服务端接收到写入数据后，首先会写日志，然后根据同步或者异步算法将数据同步到磁盘上。写入数据过程中，首先会写入log文件，写入的内容包含ledgerid，entryid，EntrySize，LastConfirmed，及其真实数据内容。然后在相应ledger文件中记录下entryid，及其该entry所在的日志文件，偏移量等。

读日志相比写日志过程，相对简单一些。同样，读日志过程也支持高可用。BKJM通过selectInputStreams方法读出一个范围内的ELIS集合，每个ELIS是BookKeeperEditLogInputStream类，new BookKeeperEditLogInputStream需要得到一个EditLogLedgerMetadata，并打开对应的ledger。具体BookKeeperEditLogInputStream类里的内容就不详细说明了。

@Overridepublic void selectInputStreams(Collection<EditLogInputStream> streams,long fromTxId, boolean inProgressOk, boolean forReading)throws IOException {List<EditLogLedgerMetadata> currentLedgerList = getLedgerList(fromTxId,inProgressOk);try {BookKeeperEditLogInputStream elis = null;for (EditLogLedgerMetadata l : currentLedgerList) {long lastTxId = l.getLastTxId();if (l.isInProgress()) {lastTxId = recoverLastTxId(l, false);}// Check once again, required in case of InProgress and is case of any// gap.if (fromTxId >= l.getFirstTxId() && fromTxId <= lastTxId) {LedgerHandle h;if (l.isInProgress()) { // we don't want to fence the current journalh = bkc.openLedgerNoRecovery(l.getLedgerId(),BookKeeper.DigestType.MAC, digestpw.getBytes());} else {h = bkc.openLedger(l.getLedgerId(), BookKeeper.DigestType.MAC,digestpw.getBytes());}elis = new BookKeeperEditLogInputStream(h, l);elis.skipTo(fromTxId);} else {// If mismatches then there might be some gap, so we should not check// further.return;}streams.add(elis);if (elis.getLastTxId() == HdfsConstants.INVALID_TXID) {return;}fromTxId = elis.getLastTxId() + 1;}} catch (BKException e) {throw new IOException("Could not open ledger for " + fromTxId, e);} catch (InterruptedException ie) {Thread.currentThread().interrupt();throw new IOException("Interrupted opening ledger for " + fromTxId, ie);}}

首先选择日志文件，建立输入流。从节点触发消化日志后，首先会查询ZK，获取到主节点写入ZK的edits元数据信息（不包含inprocess状态的edits元数据），这个元数据包含日志段的startTxid，lastTxid，ledgerID，同时也会打开相应的ledger，并获取其元数据，如ledger的quorumSize，ensembleSize，lastEntryId等，同时按照txid先后顺序对ledger进行排序，放入输入流集合。需要强调的是，当打开ledger时，会检查其entry副本之间的一致性，如果不一致需恢复。

准备好输入流以后，开始消化日志，依次操作输入流集合的ledgers，读取每个ledger内的entry：

1. 通过查询ledger元数据，同时通过RR算法确定该entry存储在哪几个bookies；
2. 尝试从bookies集合的第一个bookie服务读取entry，如果成功，该entry就读取成功，如果失败，转入第3步；
3. 尝试从bookies集合的第二个bookie服务读取entry，如果成功，该entry就读取成功，如果失败，依次类推，如果尝试读取完所有的bookies均失败，则该entry读取失败；

恢复

BKJM还有恢复机制，相关接口有recoverUnfinalizedSegments()，recoverLastTxId()。Bookie数据恢复检查通过定时或者人工发起，集群数据修复流程：

1. 通过zk查询到ledger元数据；
2. 通过元数据，查询相关bookie中存储的ledger的entry是否完整；
3. 如果查询到存储在某bookie上的entry不完整，则需要进入数据恢复流程；
4. 首先从bk服务端读取到ledger相关的entry，然后将其写到需要恢复entry的某bookie服务端；
5. Ledger数据恢复完成后，需要更新ledger的segment相关元数据。