zookeeper(12)源码分析-请求处理链(2)

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;
    }

核心方法

1、run

@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!");
    }

2、flush

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();
        }
    }

3、shutdown

函数用于关闭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();
        }
    }

文章名称:zookeeper(12)源码分析-请求处理链(2)
链接地址:http://scyanting.com/article/jdciep.html