HDFS中reportWrittenBlock函數的作用是什么

本篇文章為大家展示了HDFS中reportWrittenBlock函數的作用是什么，內容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細介紹希望你能有所收獲。

/**

     * The client can report in a set written blocks that it wrote.

     * These blocks are reported via the client instead of the datanode

     * to prevent weird heartbeat race conditions.

     */

    public void reportWrittenBlock(LocatedBlock lb) throws IOException {

        Block b = lb.getBlock();//獲取完成的這個Block信息

        DatanodeInfo targets[] = lb.getLocations();//獲取節點信息

        for (int i = 0; i < targets.length; i++) {

            namesystem.blockReceived(b, targets[i].getName());//對于每個DataNode來說，都要調用一次此函數

        }

}

C1：2014-12-19 18:26:00    C2:2014-12-19 18:59:00            C3:2014-12-19 19:03:00

=========================

那么，接下來就是理解 namesystem.blockReceived(b, targets[i].getName());了。

       /**

     * The given node is reporting that it received a certain block.

     */

    public synchronized void blockReceived(Block block, UTF8 name) {

        DatanodeInfo node = (DatanodeInfo) datanodeMap.get(name);//獲取對應的datanode

        if (node == null) {//為空可不行

            throw new IllegalArgumentException("Unexpected exception.  Got blockReceived message from node " + name + ", but there is no info for " + name);

        }

        //

        // Modify the blocks->datanode map

        // 

        addStoredBlock(block, node);//下面兩行是來執行block和node的一個映射。

        //

        // Supplement node's blockreport

        //

        node.addBlock(block);//同上

    }

C1：2014-12-19 19:11:00       C2:2014-12-19 19:11:00      C3:2014-12-19 19:12:00

===============那么接下來還有2個函數需要攻破，分別是addStoredBlock和node.addBlock(block);

后面一個函數非常簡單，不細講，所以就剩下最后一個函數了！

addStoredBlock(block, node);的執行過程如下：

synchronized void addStoredBlock(Block block, DatanodeInfo node) {

        TreeSet containingNodes = (TreeSet) blocksMap.get(block);//獲取當前block已經存在的datanode信息

        if (containingNodes == null) {//這里保證肯定存在datanode集合，不保證一定有節點在內

            containingNodes = new TreeSet();

            blocksMap.put(block, containingNodes);

        }

        if (! containingNodes.contains(node)) {//根據需要決定是否加入此datanode信息

            containingNodes.add(node);

        } else {

            LOG.info("Redundant addStoredBlock request received for block " + block + " on node " + node);

        }

//接下來的邏輯是確定是否需要重新備份

        synchronized (neededReplications) {//鎖定neededReplications

            if (dir.isValidBlock(block)) {//不懂這一句

                if (containingNodes.size() >= this.desiredReplication) {//如果已經超過最大備份個數

                    neededReplications.remove(block);//刪除此block

                    pendingReplications.remove(block);//刪除此block

                } else if (containingNodes.size() < this.desiredReplication) {

                    if (! neededReplications.contains(block)) {

                        neededReplications.add(block);//否則表示需要重新備份，這代碼寫的真夠差的。。。

                    }

                }

                //

                // Find how many of the containing nodes are "extra", if any.

                // If there are any extras, call chooseExcessReplicates() to

                // mark them in the excessReplicateMap.

                //

//也有可能一個block存儲的datanode節點數太多了，同樣要刪除這些block

                Vector nonExcess = new Vector();//構造一個空的Vector

                for (Iterator it = containingNodes.iterator(); it.hasNext(); ) {

                    DatanodeInfo cur = (DatanodeInfo) it.next();//對于當前節點來說

                    TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getName());//取到當前節點的多余塊信息

                    if (excessBlocks == null || ! excessBlocks.contains(block)) {//如果之前沒有標志在這個節點的多余塊信息里

                        nonExcess.add(cur);//則表明當前節點存儲了這個block

                    }

                }

                if (nonExcess.size() > this.maxReplication) {//如果超過了最大備份數

                    chooseExcessReplicates(nonExcess, block, this.maxReplication);//選擇若干來消除塊    

                }

            }

        }

}

void chooseExcessReplicates(Vector nonExcess, Block b, int maxReps) {

        while (nonExcess.size() - maxReps > 0) {//如果還有需要

            int chosenNode = r.nextInt(nonExcess.size());//隨機選擇一個節點

            DatanodeInfo cur = (DatanodeInfo) nonExcess.elementAt(chosenNode);

            nonExcess.removeElementAt(chosenNode);//獲取這個節點

            TreeSet excessBlocks = (TreeSet) excessReplicateMap.get(cur.getName());

            if (excessBlocks == null) {

                excessBlocks = new TreeSet();

                excessReplicateMap.put(cur.getName(), excessBlocks);

            }

            excessBlocks.add(b);//加入此block到excessReplicateMap

            //

            // The 'excessblocks' tracks blocks until we get confirmation

            // that the datanode has deleted them; the only way we remove them

            // is when we get a "removeBlock" message.  

            //

            // The 'invalidate' list is used to inform the datanode the block 

            // should be deleted.  Items are removed from the invalidate list

            // upon giving instructions to the namenode.

            //

            Vector invalidateSet = (Vector) recentInvalidateSets.get(cur.getName());

            if (invalidateSet == null) {

                invalidateSet = new Vector();

                recentInvalidateSets.put(cur.getName(), invalidateSet);

            }

            invalidateSet.add(b);//同樣的，更新recentInvalidateSets，沒啥好解釋的

        }

    }

上述內容就是HDFS中reportWrittenBlock函數的作用是什么，你們學到知識或技能了嗎？如果還想學到更多技能或者豐富自己的知識儲備，歡迎關注億速云行業資訊頻道。