事务日志的持久化是在SyncRequestProcessor类中实现，并会按照一定的规则滚动日志（关闭当前文件，创建一个新文件），以及生成新的Snapshot。在持久化过程中，使用组提交(Group Commits)来优化磁盘io 操作。组提交是指将多个Request对象的事务作为一次写附加到磁盘上。使用这种方式可以在持久化多个事务的时候，只使用一次磁盘寻道(Disk Seek)的开销。Request对象只有在其中事务同步到磁盘后，才会传递到下一个处理器。
SyncRequestProcessor被用于下面三种不同的情景中：
在SyncRequestProcessor中有两个关键队列：
SyncRequestProcessor的run方法循环读取queuedRequests队列中的Request对象并进行持久化。
流程图如下：
如果toFlush队列为空，则调用queuedRequest队列的阻塞方法take()；如果toFlush队列不为空，则调用queuedRequest队列的非阻塞方法poll()。如果poll()方法返回null，则会立即将toFlush队列中所有Request对象中事务Flush到磁盘，并将Request对象传入到下一个处理器。这样可以避免增加请求处理的延时。如果queuedRequest.poll()方法返回不为Null或者queuedRequest.take()方法返回, 则将返回的Request对象si中的事务附加到事务日志文件中，并放入toFlush队列中。如果toFlush队列大小大于1000，则将队列中所有Request对象中事务Flush到磁盘，并将Request对象传入下一个处理器。这是可以避免在有大量请求的时候增加请求处理的延时。
Request对象附加到事务日志之后，会检查日志记录数logCount是否大于(snapCount / 2 + randRoll)。如果大于则滚动日志，并启动生成新Snapshot的线程。其中randRoll是一个随机数。这个随机数的使用可以避免Zookeeper集群里的所有机器同时构建Snapshot。
SyncRequestProcessor.run方法如下：
public void run() { try { int logCount = 0; //这个随机数randRoll的使用可以避免Zookeeper集群里的所有机器同时构建Snapshot setRandRoll(r.nextInt( snapCount/2)); while (true ) { Request si = null; //如果toFlush为空，则调用队列queuedRequests的阻塞方法take() if (toFlush .isEmpty()) { si = queuedRequests.take(); } //如果toFlush不为空，则调用队列queuedRequests的非阻塞方法poll() else { si = queuedRequests.poll(); //如果si为null, 说明queuedRequests为空，则调用flush()方法 if (si == null) { flush( toFlush); continue; } } //如果si是一个poison pill, 则退出循环 if (si == requestOfDeath ) { break; } if (si != null) { // track the number of records written to the log //将record的操作记到日志中 if (zks .getZKDatabase().append(si)) { logCount++; if (logCount > (snapCount / 2 + randRoll)) { randRoll = r .nextInt(snapCount/2); //滚动事务日志 zks.getZKDatabase().rollLog(); //构建snapshot if (snapInProcess != null && snapInProcess.isAlive()) { LOG.warn("Too busy to snap, skipping" ); } else { //生成snapshot线程 snapInProcess = new Thread("Snapshot Thread") { public void run() { try { zks.takeSnapshot(); } catch(Exception e) { LOG.warn("Unexpected exception", e); } } }; //启动snapInProcess snapInProcess.start(); } logCount = 0; } } else if (toFlush .isEmpty()) { // optimization for read heavy workloads //如果这是一个read, 并且没有pending的flushes(writes), 那么直接传递到下一个处理器 if (nextProcessor != null) { nextProcessor.processRequest(si); if (nextProcessor instanceof Flushable) { ((Flushable) nextProcessor).flush(); } } continue; } toFlush.add(si); //如果toFlush的大小大于1000, 则flush if (toFlush .size() > 1000) { flush( toFlush); } } } } catch (Throwable t) { LOG.error("Severe unrecoverable error, exiting" , t); running = false ; System. exit(11); } LOG.info("SyncRequestProcessor exited!" ); }
转载请附上原博客地址：http://blog.csdn.net/jeff_fangji/article/details/44046997
ZooKeeper源码分析：Log和Snapshot持久化（SyncRequestProcessor类)
原文地址：http://blog.csdn.net/jeff_fangji/article/details/44046997
