HDFS｜NameNode 支撑超高并发的秘诀

/**
 * 定义日志类型，每条操作日志就是一个 EditLog 对象
 */
class EditLog{
    private long txid ;
    private String context ;


    public EditLog(long txid, String context) {
        this.txid = txid;
        this.context = context;
    }


    //get 和 set 方法
    @Override
    public String toString() {
        return "EditLog{" +
                "txid=" + txid +
                ", context='" + context + '\'' +
                '}';
    }
}
复制

class DoubleBuffer{


    //执行日志写入操作的缓冲
    private LinkedList<EditLog> currentBufffer = new LinkedList<EditLog>();
    //执行日志同步刷盘的缓冲
    private LinkedList<EditLog> syncBufffer = new LinkedList<EditLog>();


    //get 和 set 方法


    /**
     * 向currentBuffer缓冲中写入操作日志
     * @param log
     */
    public void write(EditLog log){
        currentBufffer.add(log);
    }


    /**
     * 返回同步缓冲中的最大txid
     * @return
     */
    public long getLastTxid(){
        return syncBufffer.getLast().getTxid();
    }


    /**
     * 交换缓冲
     */
    public void setReadyToFlush(){
        LinkedList<EditLog> tmp = currentBufffer;
        currentBufffer=syncBufffer;
        syncBufffer=tmp;
    }


    /**
     * 将同步缓冲中的数据刷盘，这里象征性地打印一下
     */
    public void flush(){
        for (EditLog editLog : syncBufffer) {
            System.out.println("存入操作日志："+editLog);
        }
        //清空缓冲
        syncBufffer.clear();
    }
}
复制

public class DoubleBufferDemo {


    private long txid = 0L;
    //双缓冲操作对象
    private DoubleBuffer buffer = new DoubleBuffer();
    //标记当前是否正在往磁盘里面刷写数据
    private volatile Boolean isSyncRunning = false;
    //标记是否有线程正在等待同步
    private volatile Boolean isWaitSync = false;
    //初始化刷写的最大事务id
    private volatile Long syncMaxTxid = 0L;


    //存储每个线程内部的事务id
    private ThreadLocal<Long> localTxid=new ThreadLocal<Long>();
    ...


    /**
     * 写入操作日志
     * @param content
     */
    public void logEdit(String context){
        //加锁
        synchronized (this){
            //线程1：txid=1
            //线程2：txid=2
            //线程3：txid=3
            txid++;


            //线程1：localTxid=1
            //线程2：localTxid=2
            //线程3：localTxid=3
            localTxid.set(txid);


            EditLog editLog = new EditLog(txid, content);


            //将操作日志写入缓冲
            buffer.write(editLog);
        }//释放锁




        logSync();
    }


    /**
     * 将缓冲中的数据刷盘
     */
    public void logSync(){
        synchronized (this){
            //当第一个线程执行到这里，isSyncRunning 为false，不执行分支内的逻辑
            if(isSyncRunning){
                //线程二：2
                //线程三：3
                long txid = localTxid.get();
                //当线程2和线程3进来时，由于txid<=syncMaxTxid(也就是3)，说明线程1进行刷盘时，sync缓冲中
                //已经包含了线程2和线程3的操作日志，所以不需要线程2和线程3再进行刷盘，方法直接返回
                //当线程4进来时，该条件不满足，接着往下执行
                if(txid <= syncMaxTxid){
                    return;
                }
                //线程4执行到这里，isWaitSync=false，继续向下执行
                //线程5执行到这里，isWaitSync=true，方法返回
                if(isWaitSync){
                    //线程5直接返回
                    return;
                }
                //线程4将isWaitSync赋值为true
                isWaitSync = true;
                //线程4执行到这里，isSyncRunning=true，执行里面的逻辑
                while(isSyncRunning){
                    try {
                        //线程4就在这儿等：退出等待有两种情况
                        //1）等待时间到了 2）被人唤醒了
                        wait(2000);//wait操作会释放锁
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                //线程4结束等待，更新isWaitSync=false
                isWaitSync = false;
            }


            //进行内存交换
            //当第一个线程执行缓冲交换时，假设 currentBuffer             //中已经完成了线程1，线程2
            //中已经完成了线程1，线程2线程3 的操作日志写入
            buffer.setReadyToFlush();


            //如果同步缓冲中存在操作日志，获取其最大的事务id
            //这里由于假设线程1-3已经完成了日志写入，所以第一次获取到的最大事务id就是3
            if(buffer.getSyncBufffer().size() > 0) {
                syncMaxTxid = buffer.getLastTxid();
            }
            //更新标识，表明正在往磁盘中同步数据
            isSyncRunning = true;


        } //线程1 释放锁


        /**
         * 分段加锁
         * 线程1 刷写数据
         * 刷盘相对较慢
         */
        //这里耗时的刷盘操作并未加锁
        buffer.flush();


        //重新加锁
        synchronized (this) {
            //线程1刷写磁盘完成，将isSyncRunning 更新为false
            isSyncRunning = false;
            //唤醒等待线程。
            notifyAll();
        }
    }
}
复制