Hadoop NameNode启动之FSDirectiry

前言

作为Hadoop NameNode类启动分析篇的起始篇章。
我们先来了解一下FSDirectiry做了哪些功能点。

一切都因下面这段代码开始(一切的是命运之门的选择, 滑稽)

进入主题

我们先来看看NameNode的构造方法吧。
可以发现会创建一个FSNamesystem对象。

/**
     * Create a NameNode at the default location
     */
    public NameNode(Configuration conf) throws IOException {
        this(getDir(conf), // 获取namenode存放路径文件夹
             DataNode.createSocketAddr // 获取datanode端口信息
             (conf.get("fs.default.name", "local")).getPort(), conf);
    }

    /**
     * Create a NameNode at the specified location and start it.
     */
    public NameNode(File dir, int port, Configuration conf) throws IOException {
        this.namesystem = new FSNamesystem(dir, conf);
        this.handlerCount = conf.getInt("dfs.namenode.handler.count", 10); // NameNode用来处理来自DataNode的RPC请求的线程数量 
        this.server = RPC.getServer(this, port, handlerCount, false, conf);
        this.server.start();
    }

之后我们进入FSNamesystem观察

public FSNamesystem(File dir, Configuration conf) throws IOException {
        this.dir = new FSDirectory(dir); // fsimg和edits的初始化
        this.hbthread = new Daemon(new HeartbeatMonitor()); // 心跳监控
        this.lmthread = new Daemon(new LeaseMonitor()); // 租约监控
        hbthread.start(); // 启动心跳线程
        lmthread.start(); // 启动租约监控线程
        this.systemStart = System.currentTimeMillis();
        this.conf = conf;
        
        this.desiredReplication = conf.getInt("dfs.replication", 3); // 备份数为3
        this.maxReplication = desiredReplication;
        this.maxReplicationStreams = conf.getInt("dfs.max-repl-streams", 2);
        this.minReplication = 1; // 最小备份数量
        this.heartBeatRecheck= 1000; // 1秒发送一次心跳
    }

现在我们只关注FSDirectory类做了哪些功能即可。
为了观察方便, 我把需要研究的代码抽离出来了，并且写成一个测试类方便进行测试

public FSDirectory(File dir) throws IOException {
        // configuration中配置name的dir路径
        File fullimage = new File(dir, "image");
        // 如果不存在则表示没有对NameNode进行过初始化
        if (! fullimage.exists()) {
          throw new IOException("NameNode not formatted: " + dir);
        }
        
        // edits文件
        File edits = new File(dir, "edits");
        // 加载和保存
        if (loadFSImage(fullimage, edits)) {
            saveFSImage(fullimage, edits);
        }

        synchronized (this) {
            this.ready = true;
            this.notifyAll();
            this.editlog = new DataOutputStream(new FileOutputStream(edits));
        }
    }

public class FSDirectorTest {

    class INode {
        public String name;
        public INode parent;
        public TreeMap children = new TreeMap();
        public Block blocks[];

        /**
         */
        INode(String name, INode parent, Block blocks[]) {
            this.name = name;
            this.parent = parent;
            this.blocks = blocks;
        }
        
        /**
         * 添加文件块, 其实就是把blk_id写入到children中(到时候在saveImage中使用)
         * @param target
         * @param blks
         * @return
         */
        INode addNode(String target, Block blks[]) {
            if (getNode(target) != null) {
                return null;
            } else {
                String parentName = DFSFile.getDFSParent(target);
                if (parentName == null) {
                    return null;
                }

                INode parentNode = getNode(parentName);
                if (parentNode == null) {
                    return null;
                } else {
                    // 读取的fsimag文件数据, 这里的targetName其实就是我们HDFS上的文件名
                    // blks就是存储在DataNode上的blk文件名字
                    // parentNode是我们的父节点(头结点)
                    String targetName = new File(target).getName();
                    INode newItem = new INode(targetName, parentNode, blks);
                    // 之后会被saveImage进行使用
                    parentNode.children.put(targetName, newItem);
                    return newItem;
                }
            }
        }

        int numItemsInTree() {
            int total = 0;
            for (Iterator it = children.values().iterator(); it.hasNext();) {
                INode child = (INode) it.next();
                total += child.numItemsInTree();
            }
            return total + 1;
        }

        /**
         * This is the external interface
         */
        INode getNode(String target) {
            if (!target.startsWith("/") || target.length() == 0) {
                return null;
            } else if (parent == null && "/".equals(target)) {
                return this;
            } else {
                Vector components = new Vector();
                int start = 0;
                int slashid = 0;
                while (start < target.length() && (slashid = target.indexOf('/', start)) >= 0) {
                    components.add(target.substring(start, slashid));
                    start = slashid + 1;
                }
                if (start < target.length()) {
                    components.add(target.substring(start));
                }
                return getNode(components, 0);
            }
        }

        /**
         */
        INode getNode(Vector components, int index) {
            if (!name.equals((String) components.elementAt(index))) {
                return null;
            }
            if (index == components.size() - 1) {
                return this;
            }

            // Check with children
            INode child = (INode) children.get(components.elementAt(index + 1));
            if (child == null) {
                return null;
            } else {
                return child.getNode(components, index + 1);
            }
        }

        /**
         * 通过递归调用把元数据信息写入到fsimage.new文件中
         * 其实就是把loadFSImage方法加载到内存的数据读取出来然后写入进去
         */
        void saveImage(String parentPrefix, DataOutputStream out) throws IOException {
            String fullName = "";
            if (parent != null) {
                fullName = parentPrefix + "/" + name;
                new UTF8(fullName).write(out);
                if (blocks == null) {
                    out.writeInt(0);
                } else {
                    out.writeInt(blocks.length);
                    for (int i = 0; i < blocks.length; i++) {
                        blocks[i].write(out);
                    }
                }
            }
            for (Iterator it = children.values().iterator(); it.hasNext();) {
                INode child = (INode) it.next();
                child.saveImage(fullName, out);
            }
        }
    }

    static String FS_IMAGE = "fsimage";
    static String NEW_FS_IMAGE = "fsimage.new";
    static String OLD_FS_IMAGE = "fsimage.old";

    // 即我们的头结点
    INode rootDir = new INode("", null, null);
    TreeSet activeBlocks = new TreeSet();

    public boolean unprotectedAddFile(UTF8 name, Block blocks[]) {
        synchronized (rootDir) {
            if (blocks != null) {
                // Add file->block mapping
                for (int i = 0; i < blocks.length; i++) {
                    activeBlocks.add(blocks[i]);
                }
            }
            return (rootDir.addNode(name.toString(), blocks) != null);
        }
    }

    public boolean loadFSImage(File fsdir, File edits) throws IOException {
        File curFile = new File(fsdir, FS_IMAGE);
        File newFile = new File(fsdir, NEW_FS_IMAGE);
        File oldFile = new File(fsdir, OLD_FS_IMAGE);

        // 这里的判断挺有意思的
        // saveFSImage中途失败还没从命名或者一些已经命名好了的文件,通过不同判断进行修复

        if (oldFile.exists() && curFile.exists()) {
            oldFile.delete();
            if (edits.exists()) {
                edits.delete();
            }
        } else if (oldFile.exists() && newFile.exists()) {
            newFile.renameTo(curFile);
            oldFile.delete();
        } else if (curFile.exists() && newFile.exists()) {
            newFile.delete();
        }

        if (curFile.exists()) {
            DataInputStream in = new DataInputStream(new BufferedInputStream(new FileInputStream(curFile)));
            try {
                int numFiles = in.readInt();
                for (int i = 0; i < numFiles; i++) {
                    UTF8 name = new UTF8();
                    name.readFields(in);
                    int numBlocks = in.readInt();
                    if (numBlocks == 0) {
                        unprotectedAddFile(name, null);
                    } else {
                        Block blocks[] = new Block[numBlocks];
                        for (int j = 0; j < numBlocks; j++) {
                            blocks[j] = new Block();
                            blocks[j].readFields(in);
                        }
                        unprotectedAddFile(name, blocks);
                    }
                }
            } finally {
                in.close();
            }
        }

        return true;

        // if (edits.exists() && loadFSEdits(edits) > 0) {
        // return true;
        // } else {
        // return false;
        // }

    }

    public void saveFSImage(File fullimage, File edits) throws IOException {
        File curFile = new File(fullimage, FS_IMAGE);
        File newFile = new File(fullimage, NEW_FS_IMAGE);
        File oldFile = new File(fullimage, OLD_FS_IMAGE);

        //
        // Write out data
        //
        DataOutputStream out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(newFile)));
        try {
            out.writeInt(rootDir.numItemsInTree() - 1);
            rootDir.saveImage("", out);
        } finally {
            out.close();
        }

        //
        // Atomic move sequence(这里说的原子操作我之前想成是类似事务的功能了，所在自己抛出一个异常，看看是否会回退,结果并没有)

        // 1-4步：把当前fsimage修改为old文件，并将saveImage方法写入数据的文件重命名为fsimage，之后删除edits和old文件
        // 1. Move cur to old
        curFile.renameTo(oldFile);
        
//        int i = 1 / 0; // 测试原子性

        // 2. Move new to cur
        newFile.renameTo(curFile);

        // 3. Remove pending-edits file (it's been integrated with newFile)
        edits.delete();

        // 4. Delete old
        oldFile.delete();
    }

    public static void main(String[] args) throws IOException {
        File dir = new File("tmp/hadoopx/dfs/name");
        File fullimage = new File(dir, "image");
        File edits = new File(dir, "edits");
        FSDirectorTest fsDirectorTest = new FSDirectorTest();
        boolean loadFSImage = fsDirectorTest.loadFSImage(fullimage, edits);
        System.out.println(loadFSImage);
        if (loadFSImage) {
            fsDirectorTest.saveFSImage(fullimage, edits);
        }
    }
}

流程图?

也算不上是一个流程图，只不过把代码上一些内容以图片方式呈现。

写入到.new的文件后然后重命名为fsimage之后再删除.old的文件这样就算加载完毕了。