ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 自动化运维 > > BRPC详解（五）——负载均衡之Round Robin

BRPC详解（五）——负载均衡之Round Robin

作者：浪花小雨滴 时间：2020-05-25 15:42:22 0 删除 编辑

一、负载均衡概述

brpc中LoadBalancer从多个服务节点中选择一个节点，目前的实现见负载均衡。

Load balancer最重要的是如何让不同线程中的负载均衡不互斥，解决这个问题的技术是DoublyBufferedData。DoublyBufferedData的相关内容已经在上一篇文章中提到了，感兴趣的小伙伴可以去翻看一下

被包养的程序猿丶：BRPC详解（四）——负载均衡之DoublyBufferedData​zhuanlan.zhihu.com图标

和NamingService类似，我们使用字符串来指代一个load balancer，在global.cpp中注册：

// Load Balancers


LoadBalancerExtension()->RegisterOrDie("rr", &g_ext->rr_lb); 


LoadBalancerExtension()->RegisterOrDie("wrr", &g_ext->wrr_lb); 


LoadBalancerExtension()->RegisterOrDie("random", &g_ext->randomized_lb); 


LoadBalancerExtension()->RegisterOrDie("la", &g_ext->la_lb); 


LoadBalancerExtension()->RegisterOrDie("c_murmurhash", &g_ext->ch_mh_lb); 


LoadBalancerExtension()->RegisterOrDie("c_md5", &g_ext->ch_md5_lb); 


LoadBalancerExtension()->RegisterOrDie("c_ketama", &g_ext->ch_ketama_lb); 


LoadBalancerExtension()->RegisterOrDie("_dynpart", &g_ext->dynpart_lb);

二、brpc中实现的负载均衡算法

1、Round Robin

总是选择列表中的下一台服务器，结尾的下一台是开头，无需其他设置。比如有3台机器a,b,c，那么brpc会依次向a, b, c, a, b, c, ...发送请求。注意这个算法的前提是服务器的配置，网络条件，负载都是类似的

2、Weighted Round Robin

根据服务器列表配置的权重值来选择服务器。服务器被选到的机会正比于其权重值，并且该算法能保证同一服务器被选到的结果较均衡的散开

3、Random

随机从列表中选择一台服务器，无需其他设置。和round robin类似，这个算法的前提也是服务器都是类似的

4、La

locality-aware，优先选择延时低的下游，直到其延时高于其他机器，无需其他设置。

5、c_md5或者c_murmurhash

一致性哈希，与简单hash的不同之处在于增加或删除机器时不会使分桶结果剧烈变化，特别适合cache类服务

三、负载均衡算法源码

所有的负载均衡算法均继承自基类LoadBalancer，在该类中主要定义了三个结构体（SelectIn、SelectOut、CallInfo）、增加（批量）/删除（批量）Server的接口、选择Server的接口（核心逻辑）等。这里指的一提的是所有的负载均衡算法为了保证不同线程之间的负载均衡不互斥，都使用了DoublyBufferedData这种数据结构

1、Round Robin

在Round Robin算法中，需要循环的依次选择不同的server，这里主要是在头文件中定义一个单独的结构体TLS，主要负责存储偏移量和步长。

struct TLS {
        TLS() : stride(), offset() { }
        uint32_t stride;
        uint32_t offset;
    };

在步长为0的时候，需要初始化一个固定的步长，即调用GenRandomStride（）函数，得到一个初始的步长，在成功选择一个server之后，就将修改之后的TLS结构存入DoublyBufferedData中，下一次选择Server的时候直接从DoublyBufferedData中提取TLS数据即可确定上一次的偏移量和步长。具体源代码如下所示

int RoundRobinLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
    butil::DoublyBufferedData<Servers, TLS>::ScopedPtr s;
    if (_db_servers.Read(&s) != ) {
        return ENOMEM;
    }
    const size_t n = s->server_list.size();
    if (n == ) {
        return ENODATA;
    }
    if (_cluster_recover_policy && _cluster_recover_policy->StopRecoverIfNecessary()) {
        if (_cluster_recover_policy->DoReject(s->server_list)) {
            return EREJECT;
        }
    }
    TLS tls = s.tls();
    if (tls.stride == ) {
        tls.stride = GenRandomStride();
        tls.offset = ;
    }

    for (size_t i = ; i < n; ++i) {
        tls.offset = (tls.offset + tls.stride) % n;
        const SocketId id = s->server_list[tls.offset].id;
        if (((i + 1) == n  // always take last chance
             || !ExcludedServers::IsExcluded(in.excluded, id))
            && Socket::Address(id, out->ptr) == 
            && (*out->ptr)->IsAvailable()) {
            s.tls() = tls;
            return ;
        }
    }
    if (_cluster_recover_policy) {
        _cluster_recover_policy->StartRecover();
    }
    s.tls() = tls;
    return EHOSTDOWN;
}

2、Weighted Round Robin

类似于Round Robin，但是在Weighted Round Robin中，对TLS的定义有一些不同，当服务器列表发生变化的时候，需要重新计算步长，具体定义如下：

struct TLS {
        size_t position = ;
        uint64_t stride = ;
        Server remain_server;
        // If server list changed, we need caculate a new stride.
        bool IsNeededCaculateNewStride(const uint64_t curr_weight_sum, 
                                       const size_t curr_servers_num) {
            if (curr_weight_sum != weight_sum 
                || curr_servers_num != servers_num) {
                weight_sum = curr_weight_sum;
                servers_num = curr_servers_num;
                return true;
            }
            return false;
        }
    private:
        uint64_t weight_sum = ;
        size_t servers_num = ;
    };



const std::vector<uint64_t> prime_stride = {
2,3,5,11,17,29,47,71,107,137,163,251,307,379,569,683,857,1289,1543,1949,2617,
2927,3407,4391,6599,9901,14867,22303,33457,50207,75323,112997,169501,254257,
381389,572087,849083,1273637,1910471,2865727,4298629,6447943,9671923,14507903,
21761863,32642861,48964297,73446469,110169743,165254623,247881989,371822987,
557734537,836601847,1254902827,1882354259,2823531397,4235297173,6352945771,
9529418671};

bool IsCoprime(uint64_t num1, uint64_t num2) {
    uint64_t temp;
    if (num1 < num2) {
        temp = num1;
        num1 = num2;
        num2 = temp;
    }
    while (true) {
        temp = num1 % num2;
        if (temp == ) {
            break;
        } else {
            num1 = num2;
            num2 = temp;
        }
    }
    return num2 == 1;
}

// Get a reasonable stride according to weights configured of servers.
uint64_t GetStride(const uint64_t weight_sum, const size_t num) {
    if (weight_sum == 1) {
      return 1;
    }
    uint32_t average_weight = weight_sum / num;
    auto iter = std::lower_bound(prime_stride.begin(), prime_stride.end(),
                                 average_weight);
    while (iter != prime_stride.end()
           && !IsCoprime(weight_sum, *iter)) {
        ++iter;
    }
    CHECK(iter != prime_stride.end()) << "Failed to get stride";
    return *iter > weight_sum ? *iter % weight_sum : *iter;
}



int WeightedRoundRobinLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
    butil::DoublyBufferedData<Servers, TLS>::ScopedPtr s;
    if (_db_servers.Read(&s) != ) {
        return ENOMEM;
    }
    if (s->server_list.empty()) {
        return ENODATA;
    }
    TLS& tls = s.tls();
    if (tls.IsNeededCaculateNewStride(s->weight_sum, s->server_list.size())) {
      if (tls.stride == ) {
          tls.position = butil::fast_rand_less_than(s->server_list.size());
      }
      tls.stride = GetStride(s->weight_sum, s->server_list.size());
    }
    // If server list changed, the position may be out of range.
    tls.position %= s->server_list.size();
    // Check whether remain server was removed from server list.
    if (tls.remain_server.weight >  &&
        tls.remain_server.id != s->server_list[tls.position].id) {
        tls.remain_server.weight = ;
    }
    // The servers that can not be choosed.
    std::unordered_set<SocketId> filter;
    TLS tls_temp = tls;
    uint64_t remain_weight = s->weight_sum;
    size_t remain_servers = s->server_list.size();
    while (remain_servers > ) {
        SocketId server_id = GetServerInNextStride(s->server_list, filter, tls_temp);
        if (!ExcludedServers::IsExcluded(in.excluded, server_id)
            && Socket::Address(server_id, out->ptr) == 
            && (*out->ptr)->IsAvailable()) {
            // update tls.
            tls.remain_server = tls_temp.remain_server;
            tls.position = tls_temp.position;
            return ;
        } else {
            // Skip this invalid server. We need calculate a new stride for server selection.
            if (--remain_servers == ) {
                break;
            }
            filter.emplace(server_id);
            remain_weight -= (s->server_list[s->server_map.at(server_id)]).weight;
            // Select from begining status.
            tls_temp.stride = GetStride(remain_weight, remain_servers);
            tls_temp.position = tls.position;
            tls_temp.remain_server = tls.remain_server;
        }
    }
    return EHOSTDOWN;
}

3、Random

随机选择算法的逻辑就比较简单了，首先生成一个随机数用作偏移量，然后从server列表中选择指定偏移量的server，对其进行检查（包括是否需要在负载均衡极端被排除、SocketId对应的Socket是否可用），如果检查没有问题则成功，如果检查发现有问题，则需要重新指定偏移量，这里新的偏移量为旧偏移量加上一个步长，这个步长由一个专门的函数生成，可以保证当前for训练过程中检查有问题的server不再被检索到，具体源代码如下所示

int RandomizedLoadBalancer::SelectServer(const SelectIn& in, SelectOut* out) {
    butil::DoublyBufferedData<Servers>::ScopedPtr s;
    if (_db_servers.Read(&s) != ) {
        return ENOMEM;
    }
    size_t n = s->server_list.size();
    if (n == ) {
        return ENODATA;
    }
    if (_cluster_recover_policy && _cluster_recover_policy->StopRecoverIfNecessary()) {
        if (_cluster_recover_policy->DoReject(s->server_list)) {
            return EREJECT;
        }
    }
    uint32_t stride = ;
    size_t offset = butil::fast_rand_less_than(n);
    for (size_t i = ; i < n; ++i) {
        const SocketId id = s->server_list[offset].id;
        if (((i + 1) == n  // always take last chance
             || !ExcludedServers::IsExcluded(in.excluded, id))
            && Socket::Address(id, out->ptr) == 
            && (*out->ptr)->IsAvailable()) {
            // We found an available server
            return ;
        }
        if (stride == ) {
            stride = GenRandomStride();
        }
        // If `Address' failed, use `offset+stride' to retry so that
        // this failed server won't be visited again inside for
        offset = (offset + stride) % n;
    }
    if (_cluster_recover_policy) {
        _cluster_recover_policy->StartRecover();
    }
    // After we traversed the whole server list, there is still no
    // available server
    return EHOSTDOWN;
}

感兴趣的小伙伴可以关注专栏：

独立团丶​zhuanlan.zhihu.com图标

感兴趣的小伙伴也可以关注公众号：

来自 “ ITPUB博客 ” ，链接：https://blog.itpub.net/31537575/viewspace-3123317/ ，如需转载，请注明出处，否则将追究法律责任。

0
0
上一篇： 利用docker部署深度学习模型的一个最佳实践
下一篇： 数据库主键一定要自增吗？有哪些场景不建议自增？
请登录后发表评论 登录
全部评论
浪花小雨滴

注册时间：2018-05-16

  • 博文量
    4259
  • 访问量
    40116

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593