2021-12-03 DataNode注册与心跳

2021-12-03 DataNode注册与心跳

本文基于hadoop-3.3.0

目录

1 概述

2 注册过程

2.1 注册前置过程

2.2 注册过程

2.3 心跳发送

---

1 概述

dn在完成启动（源码阅读）之后，需要向nn进行注册，才能提供后续的数据存储服务，注册完成后定时进行心跳检测，确保dn的存活。

而注册的流程是在DataNode的构造函数中的startDataNode方法中调用refreshNamenode方法完成：

blockPoolManager.refreshNamenodes(getConf());

2 注册过程 2.1 注册前置过程

先看refreshNamenodes方法，获取到新的NS和NNLifeRpcAddress后调用doRefreshNamenodes完成后续 *** 作

void refreshNamenodes(Configuration conf)
    throws IOException {
    LOG.info("Refresh request received for nameservices: " +
             conf.get(DFSConfigKeys.DFS_NAMESERVICES));

    Map> newAddressMap = null;
    Map> newLifelineAddressMap = null;

    try {
        newAddressMap =
            DFSUtil.getNNServiceRpcAddressesForCluster(conf);
        newLifelineAddressMap =
            DFSUtil.getNNLifelineRpcAddressesForCluster(conf);
    } catch (IOException ioe) {
        LOG.warn("Unable to get NameNode addresses.");
    }

    if (newAddressMap == null || newAddressMap.isEmpty()) {
        throw new IOException("No services to connect, missing NameNode " +
                              "address.");
    }

    synchronized (refreshNamenodesLock) {
        doRefreshNamenodes(newAddressMap, newLifelineAddressMap);
    }
}

doRefreshNamenodes方法中主要包含5个步骤：

1. 对于每个新的名称服务，确定它是对现有 NS 的一组 NN 的更新，还是全新的名称服务

2. 我们目前拥有但不再存在的任何名称服务都需要删除

3. 开启新的ns

上面三步是在同步代码块中

4. 删除过时的ns，不在同步代码块，可能触发删除等 *** 作，耗时多

5. 更新ns

在第三步的时候通过BPOfferService完成dn的注册和心跳（下一节）

private void doRefreshNamenodes(
    Map> addrMap,
    Map> lifelineAddrMap)
    throws IOException {
    assert Thread.holdsLock(refreshNamenodesLock);

    Set toRefresh = Sets.newlinkedHashSet();
    Set toAdd = Sets.newlinkedHashSet();
    Set toRemove;

    synchronized (this) {
        // Step 1. For each of the new nameservices, figure out whether
        // it's an update of the set of NNs for an existing NS,
        // or an entirely new nameservice.
        for (String nameserviceId : addrMap.keySet()) {
            if (bpByNameserviceId.containsKey(nameserviceId)) {
                toRefresh.add(nameserviceId);
            } else {
                toAdd.add(nameserviceId);
            }
        }

        // Step 2. Any nameservices we currently have but are no longer present
        // need to be removed.
        toRemove = Sets.newHashSet(Sets.difference(
            bpByNameserviceId.keySet(), addrMap.keySet()));

        assert toRefresh.size() + toAdd.size() ==
            addrMap.size() :
        "toAdd: " + Joiner.on(",").useForNull("").join(toAdd) +
            "  toRemove: " + Joiner.on(",").useForNull("").join(toRemove) +
            "  toRefresh: " + Joiner.on(",").useForNull("").join(toRefresh);


        // Step 3. Start new nameservices
        if (!toAdd.isEmpty()) {
            LOG.info("Starting BPOfferServices for nameservices: " +
                     Joiner.on(",").useForNull("").join(toAdd));

            for (String nsToAdd : toAdd) {
                Map nnIdToAddr = addrMap.get(nsToAdd);
                Map nnIdToLifelineAddr =
                    lifelineAddrMap.get(nsToAdd);
                ArrayList addrs =
                    Lists.newArrayListWithCapacity(nnIdToAddr.size());
                ArrayList nnIds =
                    Lists.newArrayListWithCapacity(nnIdToAddr.size());
                ArrayList lifelineAddrs =
                    Lists.newArrayListWithCapacity(nnIdToAddr.size());
                for (String nnId : nnIdToAddr.keySet()) {
                    addrs.add(nnIdToAddr.get(nnId));
                    nnIds.add(nnId);
                    lifelineAddrs.add(nnIdToLifelineAddr != null ?
                                      nnIdToLifelineAddr.get(nnId) : null);
                }
                // 根据传入的参数初始化BPOfferService，此方法是BlockPoolManager的成员函数，
                // 而BlockPoolManager对象是在初始化DataNode的时候创建（构造函数中创建，创建时传入了dn）
                BPOfferService bpos = createBPOS(nsToAdd, nnIds, addrs,
                                                 lifelineAddrs);
                bpByNameserviceId.put(nsToAdd, bpos);
                offerServices.add(bpos);
            }
        }
        startAll();
    }

    // Step 4. Shut down old nameservices. This happens outside
    // of the synchronized(this) lock since they need to call
    // back to .remove() from another thread
    if (!toRemove.isEmpty()) {
        LOG.info("Stopping BPOfferServices for nameservices: " +
                 Joiner.on(",").useForNull("").join(toRemove));

        for (String nsToRemove : toRemove) {
            BPOfferService bpos = bpByNameserviceId.get(nsToRemove);
            bpos.stop();
            bpos.join();
            // they will call remove on their own
        }
    }

    // Step 5. Update nameservices whose NN list has changed
    if (!toRefresh.isEmpty()) {
        LOG.info("Refreshing list of NNs for nameservices: " +
                 Joiner.on(",").useForNull("").join(toRefresh));

        for (String nsToRefresh : toRefresh) {
            BPOfferService bpos = bpByNameserviceId.get(nsToRefresh);
            Map nnIdToAddr = addrMap.get(nsToRefresh);
            Map nnIdToLifelineAddr =
                lifelineAddrMap.get(nsToRefresh);
            ArrayList addrs =
                Lists.newArrayListWithCapacity(nnIdToAddr.size());
            ArrayList lifelineAddrs =
                Lists.newArrayListWithCapacity(nnIdToAddr.size());
            ArrayList nnIds = Lists.newArrayListWithCapacity(
                nnIdToAddr.size());
            for (String nnId : nnIdToAddr.keySet()) {
                addrs.add(nnIdToAddr.get(nnId));
                lifelineAddrs.add(nnIdToLifelineAddr != null ?
                                  nnIdToLifelineAddr.get(nnId) : null);
                nnIds.add(nnId);
            }
            try {
                UserGroupInformation.getLoginUser()
                    .doAs(new PrivilegedExceptionAction