腾讯云MongoDB多机房部署场景下就近访问原理详解

try {    ScopedDbConnection conn(ConnectionString(ns.host), socketTimeoutSecs);    bool ignoredOutParam = false;    Timer timer;    //开始计时    if (conn->isStillConnected()) {        conn->isMaster(ignoredOutParam, &reply);    //执行isMaster命令获取副本集状态    } else {        log() << "Connection to " << ns.host.toString() << " is closed";        reply = BSONObj();    }    pingMicros = timer.micros();    //统计本次耗时    conn.done();} catch (const DBException& ex) {    ...}

if (reply.latencyMicros >= 0) {    if (latencyMicros == unknownLatency) {        latencyMicros = reply.latencyMicros;    //第一次更新，直接赋值    } else {        // update latency with smoothed moving average (1/4th the delta)        latencyMicros += (reply.latencyMicros - latencyMicros) / 4; //根据当前延迟和历史统计，进行平滑更新    }}

case ReadPreference::SecondaryOnly:case ReadPreference::Nearest: {    BSONForEach(tagElem, criteria.tags.getTagBSON()) {        uassert(16358, "Tags should be a BSON object", tagElem.isABSONObj());        BSONObj tag = tagElem.Obj();
        std::vector<const Node*> matchingNodes;        for (size_t i = 0; i < nodes.size(); i++) {    // 如果是SecondaryOnly模式，需要进行过滤            if (nodes[i].matches(criteria.pref) && nodes[i].matches(tag)) {                matchingNodes.push_back(&nodes[i]);            }        }
        // don't do more complicated selection if not needed        if (matchingNodes.empty())            continue;        if (matchingNodes.size() == 1)            return matchingNodes.front()->host;
        // order by latency and don't consider hosts further than a threshold from the        // closest.        // 对候选节点按延迟进行排序        std::sort(matchingNodes.begin(), matchingNodes.end(), compareLatencies);        for (size_t i = 1; i < matchingNodes.size(); i++) {            int64_t distance =                matchingNodes[i]->latencyMicros - matchingNodes[0]->latencyMicros;            if (distance >= latencyThresholdMicros) {                // this node and all remaining ones are too far away        // 剔除延迟超过阈值（默认15ms，可配置）的节点                matchingNodes.erase(matchingNodes.begin() + i, matchingNodes.end());                break;            }        }
        // of the remaining nodes, pick one at random (or use round-robin)        if (ReplicaSetMonitor::useDeterministicHostSelection) {            // only in tests            return matchingNodes[roundRobin++ % matchingNodes.size()]->host;        } else {            // normal case            // 从剩余的候选节点中，随机选取一个返回            return matchingNodes[rand.nextInt32(matchingNodes.size())]->host;        };    }
    return HostAndPort();}

replication:   localPingThresholdMs: <int>

for {    if loop {        time.Sleep(delay)   // 每隔一段时间（默认15秒）采集一次    }    op := op    socket, _, err := server.AcquireSocket(0, delay)    if err == nil {        start := time.Now()        _, _ = socket.SimpleQuery(&op)  // 执行ping命令        delay := time.Now().Sub(start)  // 并统计耗时
        server.pingWindow[server.pingIndex] = delay        server.pingIndex = (server.pingIndex + 1) % len(server.pingWindow)        server.pingCount++        var max time.Duration        for i := 0; i < len(server.pingWindow) && uint32(i) < server.pingCount; i++ {            if server.pingWindow[i] > max {                max = server.pingWindow[i]  // 统计最近6次（默认）采集的最大值            }        }        socket.Release()        server.Lock()        if server.closed {            loop = false        }        server.pingValue = max  // 将最大值作为网络延迟统计，作为后续选择节点时的评估依据        server.Unlock()        logf("Ping for %s is %d ms", server.Addr, max/time.Millisecond)    } else if err == errServerClosed {        return    }    if !loop {        return    }}

// BestFit returns the best guess of what would be the most interesting// server to perform operations on at this point in time.func (servers *mongoServers) BestFit(mode Mode, serverTags []bson.D) *mongoServer {    var best *mongoServer    for _, next := range servers.slice {    // 遍历每一个候选节点        if best == nil {            best = next            best.RLock()            if serverTags != nil && !next.info.Mongos && !best.hasTags(serverTags) {                best.RUnlock()                best = nil            }            continue        }        next.RLock()        swap := false        switch {        case serverTags != nil && !next.info.Mongos && !next.hasTags(serverTags):            // Must have requested tags.        case next.info.Master != best.info.Master && mode != Nearest:            // Prefer slaves, unless the mode is PrimaryPreferred.            swap = (mode == PrimaryPreferred) != best.info.Master        case absDuration(next.pingValue-best.pingValue) > 15*time.Millisecond:            // Prefer nearest server.            // 如果找到一个明显更近(默认15ms为判断依据)的节点，则用更近的            swap = next.pingValue < best.pingValue        case len(next.liveSockets)-len(next.unusedSockets) < len(best.liveSockets)-len(best.unusedSockets):            // Prefer servers with less connections.            // 如果延迟没有明显差距，则用当前连接数比较小的节点            swap = true        }        if swap {            best.RUnlock()            best = next        } else {            next.RUnlock()        }    }    if best != nil {        best.RUnlock()    }    return best}

now := time.Now()    // 开始统计耗时
// 去对应的节点上执行isMaster命令isMasterCmd := &command.IsMaster{Compressors: s.cfg.compressionOpts}isMaster, err := isMasterCmd.RoundTrip(ctx, conn)
...
delay := time.Since(now)    // 得到耗时统计desc = description.NewServer(s.address, isMaster).SetAverageRTT(s.updateAverageRTT(delay))    // 进行平滑统计

func (s *Server) updateAverageRTT(delay time.Duration) time.Duration {    if !s.averageRTTSet {        s.averageRTT = delay    // 如果是第一次统计，则直接赋值    } else {        alpha := 0.2        // 进行平滑处理，新数据和历史数据的比重为 1 : 4        s.averageRTT = time.Duration(alpha*float64(delay) + (1-alpha)*float64(s.averageRTT))    }    return s.averageRTT}

readSelect := description.CompositeSelector([]description.ServerSelector{  // 复合选择器    description.ReadPrefSelector(rp),   // 1.根据readPreference设置，选择主从节点    description.LatencySelector(db.client.localThreshold),  //2.根据延迟选择最优节点})

func (ls *latencySelector) SelectServer(t Topology, candidates []Server) ([]Server, error) {    if ls.latency < 0 {        return candidates, nil    }
    switch len(candidates) {    case 0, 1:        return candidates, nil    default:        min := time.Duration(math.MaxInt64)        for _, candidate := range candidates {            if candidate.AverageRTTSet {    // 计算所有候选节点的最小延迟                if candidate.AverageRTT < min {                    min = candidate.AverageRTT                }            }        }
        if min == math.MaxInt64 {            return candidates, nil        }        // 用最小延迟加阈值配置（默认15ms）作为最大容忍延迟        max := min + ls.latency
        var result []Server        for _, candidate := range candidates {            if candidate.AverageRTTSet {                if candidate.AverageRTT <= max {                    // 返回所有符合延迟标准（最大容忍延迟）的节点                    result = append(result, candidate)                }            }        }
        return result, nil    }}

for {    // 根据前面介绍的“复合选择器”，得到候选节点列表    suitable, err := t.selectServer(ctx, sub.C, ss, ssTimeoutCh)    if err != nil {        return nil, err    }
    // 随机选择一个作为目标节点    selected := suitable[rand.Intn(len(suitable))]    selectedS, err := t.FindServer(selected)    switch {    case err != nil:        return nil, err    case selectedS != nil:        return selectedS, nil    default:        // We don't have an actual server for the provided description.        // This could happen for a number of reasons, including that the        // server has since stopped being a part of this topology, or that        // the server selector returned no suitable servers.    }}