189 8069 5689
SyncRequestProcessor，该处理器将请求存入磁盘，其将请求批量的存入磁盘以提高效率，请求在写入磁盘之前是不会被转发到下个处理器的。
创新互联为您提适合企业的网站设计 让您的网站在搜索引擎具有高度排名，让您的网站具备超强的网络竞争力！结合企业自身，进行网站设计及把握，最后结合企业文化和具体宗旨等，才能创作出一份性化解决方案。从网站策划到网站设计、成都网站制作， 我们的网页设计师为您提供的解决方案。
SyncRequestProcessor维护了ZooKeeperServer实例，其用于获取ZooKeeper的数据库和其他信息；维护了一个处理请求的队列，其用于存放请求；维护了一个处理快照的线程，用于处理快照；维护了一个running标识，标识SyncRequestProcessor是否在运行；同时还维护了一个等待被刷新到磁盘的请求队列。
// Zookeeper服务器
private final ZooKeeperServer zks;
// 请求队列
private final LinkedBlockingQueue queuedRequests =
new LinkedBlockingQueue();
// 下个处理器
private final RequestProcessor nextProcessor;
// 快照处理线程
private Thread snapInProcess = null;
// 是否在运行中
volatile private boolean running;
/**
* Transactions that have been written and are waiting to be flushed to
* disk. Basically this is the list of SyncItems whose callbacks will be
* invoked after flush returns successfully.
*/
// 等待被刷新到磁盘的请求队列
private final LinkedList toFlush = new LinkedList();
// 随机数生成器
private final Random r = new Random();
/**
* The number of log entries to log before starting a snapshot
*/
// 快照个数
private static int snapCount = ZooKeeperServer.getSnapCount();
// 结束请求标识
private final Request requestOfDeath = Request.requestOfDeath; 构造函数首先会调用父类的构造函数，然后根据构造函数参数给类的属性赋值，其中会确定下个处理器，并会设置该处理器正在运行的标识。
public SyncRequestProcessor(ZooKeeperServer zks,
RequestProcessor nextProcessor) {
super("SyncThread:" + zks.getServerId(), zks
.getZooKeeperServerListener());
this.zks = zks;
this.nextProcessor = nextProcessor;
running = true;
}@Override
public void run() {
try {
// 写日志数量初始化为0
int logCount = 0;
// we do this in an attempt to ensure that not all of the servers
// in the ensemble take a snapshot at the same time
// 防止集群中所有机器在同一时刻进行数据快照,对是否进行数据快照增加随机因素
int randRoll = r.nextInt(snapCount/2);
while (true) {
Request si = null;
// 没有需要刷新到磁盘的请求
if (toFlush.isEmpty()) {
// 从请求队列中取出一个请求，若queuedRequests队列为空会阻塞
si = queuedRequests.take();
} else {
// 从请求队列中取出一个请求，若queuedRequests队列为空，则返回空，不会阻塞
si = queuedRequests.poll();
// 取出的请求为空
if (si == null) {
// 刷新数据磁盘
flush(toFlush);
continue;
}
}
// 在关闭处理器之后，会添加requestOfDeath请求到queuedRequests队列，表示关闭后不再处理请求
if (si == requestOfDeath) {
break;
}
// 请求不为空，处理请求
if (si != null) {
// track the number of records written to the log
// 将写请求添加至事务日志文件 FileTxnSnapLog.append(si)
if (zks.getZKDatabase().append(si)) {
// 日志写入，logCount加1
logCount++;
//确定是否需要进行数据快照
if (logCount > (snapCount / 2 + randRoll)) {
randRoll = r.nextInt(snapCount/2);
// roll the log
// 滚动日志，从当前日志文件滚到下一个日志文件，不是回滚
zks.getZKDatabase().rollLog();
// take a snapshot
if (snapInProcess != null && snapInProcess.isAlive()) { // 正在进行快照
LOG.warn("Too busy to snap, skipping");
} else {
// 创建线程来处理快照
snapInProcess = new ZooKeeperThread("Snapshot Thread") {
public void run() {
try {
// 进行快照
zks.takeSnapshot();
} catch(Exception e) {
LOG.warn("Unexpected exception", e);
}
}
};
// 开始快照线程处理
snapInProcess.start();
}
// 重置为0
logCount = 0;
}
} else if (toFlush.isEmpty()) {// 读请求会走到这里，查看此时toFlush是否为空，如果为空，说明近段时间读多写少，直接响应
// optimization for read heavy workloads
// iff this is a read, and there are no pending
// flushes (writes), then just pass this to the next
// processor
if (nextProcessor != null) {
// 下个处理器开始处理请求
nextProcessor.proce***equest(si);
// 处理器是Flushable的，刷新数据到磁盘
if (nextProcessor instanceof Flushable) {
((Flushable)nextProcessor).flush();
}
}
continue;
}
// 将请求添加至被刷新至磁盘队列
toFlush.add(si);
if (toFlush.size() > 1000) {// 队列大小大于1000，直接刷新到磁盘
flush(toFlush);
}
}
}
} catch (Throwable t) {
handleException(this.getName(), t);
} finally{
running = false;
}
LOG.info("SyncRequestProcessor exited!");
}flush将toFlush队列中的请求刷新到磁盘中。
private void flush(LinkedList toFlush)
throws IOException, RequestProcessorException
{
if (toFlush.isEmpty())
return;
// 提交事务至ZK数据库
zks.getZKDatabase().commit();
while (!toFlush.isEmpty()) {
// 从队列移除请求
Request i = toFlush.remove();
// 下个处理器开始处理请求
if (nextProcessor != null) {
nextProcessor.proce***equest(i);
}
}
if (nextProcessor != null && nextProcessor instanceof Flushable) {
((Flushable)nextProcessor).flush();
}
} 函数用于关闭SyncRequestProcessor处理器，其首先会在queuedRequests队列中添加一个结束请求requestOfDeath，然后再判断SyncRequestProcessor是否还在运行，若是，则会等待其结束；之后判断toFlush队列是否为空，若不为空，则刷新到磁盘中
public void shutdown() {
LOG.info("Shutting down");
// 添加结束请求请求至队列
queuedRequests.add(requestOfDeath);
try {
// 还在运行
if(running){
this.join();// 等待该线程终止
}
if (!toFlush.isEmpty()) {// 队列不为空，刷新到磁盘
flush(toFlush);
}
} catch(InterruptedException e) {
LOG.warn("Interrupted while wating for " + this + " to finish");
} catch (IOException e) {
LOG.warn("Got IO exception during shutdown");
} catch (RequestProcessorException e) {
LOG.warn("Got request processor exception during shutdown");
}
if (nextProcessor != null) {
nextProcessor.shutdown();
}
}