深入浅出负载均衡
作者：vivo互联网团队-Zhang Peng
一、负载均衡简介
应用集群：将同一应用部署到多台机器上，组成处理集群，接收负载均衡设备分发的请求，进行处理，并返回相应数据。 负载均衡：将用户访问请求，通过某种算法，分发到集群中的节点。
高并发：负载均衡通过算法调整负载，尽力均匀的分配应用集群中各节点的工作量，以此提高应用集群的并发处理能力（吞吐量）。 伸缩性：添加或减少服务器数量，然后由负载均衡进行分发控制。这使得应用集群具备伸缩性。 高可用：负载均衡器可以监控候选服务器，当服务器不可用时，自动跳过，将请求分发给可用的服务器。这使得应用集群具备高可用的特性。 安全防护：有些负载均衡软件或硬件提供了安全性功能，如：黑白名单处理、防火墙，防 DDos 攻击等。
二、负载均衡的分类
功能强大：支持全局负载均衡并提供较全面的、复杂的负载均衡算法。 性能强悍：硬件负载均衡由于是在专用处理器上运行，因此吞吐量大，可支持单机百万以上的并发。 安全性高：往往具备防火墙，防 DDos 攻击等安全功能。
成本昂贵：购买和维护硬件负载均衡的成本都很高。 扩展性差：当访问量突增时，超过限度不能动态扩容。
LVS 可以作为四层负载均衡器。其负载均衡的性能要优于 Nginx。 HAProxy 可以作为 HTTP 和 TCP 负载均衡器。 Nginx、HAProxy 可以作为四层或七层负载均衡器。
扩展性好：适应动态变化，可以通过添加软件负载均衡实例，动态扩展到超出初始容量的能力。 成本低廉：软件负载均衡可以在任何标准物理设备上运行，降低了购买和运维的成本。
性能略差：相比于硬件负载均衡，软件负载均衡的性能要略低一些。
DNS 重定向 HTTP 重定向 反向代理
修改 IP 地址 修改 MAC 地址
使用简单：负载均衡工作，交给 DNS 服务器处理，省掉了负载均衡服务器维护的麻烦 提高性能：可以支持基于地址的域名解析，解析成距离用户最近的服务器地址（类似 CDN 的原理），可以加快访问速度，改善性能；
可用性差：DNS 解析是多级解析，新增/修改 DNS 后，解析时间较长；解析过程中，用户访问网站将失败； 扩展性低：DNS 负载均衡的控制权在域名商那里，无法对其做更多的改善和扩展； 维护性差：也不能反映服务器的当前运行状态；支持的算法少；不能区分服务器的差异（不能根据系统与服务的状态来判断负载）。
性能较差：每次访问需要两次请求服务器，增加了访问的延迟。 降低搜索排名：使用重定向后，搜索引擎会视为 SEO 作弊。 如果负载均衡器宕机，就无法访问该站点。
正向代理：发生在 客户端，是由用户主动发起的。翻墙软件就是典型的正向代理，客户端通过主动访问代理服务器，让代理服务器获得需要的外网数据，然后转发回客户端。 反向代理：发生在 服务端，用户不知道代理的存在。
1) 多种负载均衡算法：支持多种负载均衡算法，以应对不同的场景需求。 2) 可以监控服务器：基于 HTTP 协议，可以监控转发服务器的状态，如：系统负载、响应时间、是否可用、连接数、流量等，从而根据这些数据调整负载均衡的策略。
1) 额外的转发开销：反向代理的转发操作本身是有性能开销的，可能会包括创建连接，等待连接响应，分析响应结果等操作。
2) 增加系统复杂度：反向代理常用于做分布式应用的水平扩展，但反向代理服务存在以下问题，为了解决以下问题会给系统整体增加额外的复杂度和运维成本：
反向代理服务如果自身宕机，就无法访问站点，所以需要有 高可用 方案，常见的方案有：主备模式（一主一备）、双主模式（互为主备）。 反向代理服务自身也存在性能瓶颈，随着需要转发的请求量不断攀升，需要有 可扩展 方案。
客户端请求 192.168.137.10，由负载均衡服务器接收到报文。 负载均衡服务器根据算法选出一个服务节点 192.168.0.1，然后将报文请求地址改为该节点的 IP。 真实服务节点收到请求报文，处理后，返回响应数据到负载均衡服务器。 负载均衡服务器将响应数据的源地址改负载均衡服务器地址，返回给客户端。
当用户访问 www.sina.com.cn 时，用户数据通过层层网络，最后通过交换机进入 LVS 服务器网卡，并进入内核网络层。
进入 PREROUTING 后经过路由查找，确定访问的目的 VIP 是本机 IP 地址，所以数据包进入到 INPUT 链上
IPVS 是工作在 INPUT 链上，会根据访问的 vip+port 判断请求是否 IPVS 服务，如果是则调用注册的 IPVS HOOK 函数，进行 IPVS 相关主流程，强行修改数据包的相关数据，并将数据包发往 POSTROUTING 链上。
POSTROUTING 上收到数据包后，根据目标 IP 地址（后端服务器），通过路由选路，将数据包最终发往后端的服务器上。
三、负载均衡算法
根据负载均衡算法在候选服务器列表选出一个服务器; 将请求数据发送到该服务器上。
public interface LoadBalance<N extends Node> {N select(List<N> nodes, String ip);}
public abstract class BaseLoadBalance<N extends Node> implements LoadBalance<N> {public N select(List<N> nodes, String ip) {if (CollectionUtil.isEmpty(nodes)) {return null;}// 如果 nodes 列表中仅有一个 node，直接返回即可，无需进行负载均衡if (nodes.size() == 1) {return nodes.get(0);}return doSelect(nodes, ip);}protected abstract N doSelect(List<N> nodes, String ip);}
public class Node implements Comparable<Node> {protected String url;protected Integer weight;protected Integer active;// ...}
public class RandomLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {private final Random random = new Random();protected N doSelect(List<N> nodes, String ip) {// 在列表中随机选取一个节点int index = random.nextInt(nodes.size());return nodes.get(index);}}
public class WeightRandomLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {private final Random random = ThreadLocalRandom.current();protected N doSelect(List<N> nodes, String ip) {int length = nodes.size();AtomicInteger totalWeight = new AtomicInteger(0);for (N node : nodes) {Integer weight = node.getWeight();totalWeight.getAndAdd(weight);}if (totalWeight.get() > 0) {int offset = random.nextInt(totalWeight.get());for (N node : nodes) {// 让随机值 offset 减去权重值offset -= node.getWeight();if (offset < 0) {// 返回相应的 Nodereturn node;}}}// 直接随机返回一个return nodes.get(random.nextInt(length));}}
public class RoundRobinLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {private final AtomicInteger position = new AtomicInteger(0);protected N doSelect(List<N> nodes, String ip) {int length = nodes.size();// 如果位置值已经等于节点数，重置为 0position.compareAndSet(length, 0);N node = nodes.get(position.get());position.getAndIncrement();return node;}}
public class WeightRoundRobinLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {/*** 60秒*/private static final int RECYCLE_PERIOD = 60000;/*** Node hashcode 到 WeightedRoundRobin 的映射关系*/private ConcurrentMap<Integer, WeightedRoundRobin> weightMap = new ConcurrentHashMap<>();/*** 原子更新锁*/private AtomicBoolean updateLock = new AtomicBoolean();protected N doSelect(List<N> nodes, String ip) {int totalWeight = 0;long maxCurrent = Long.MIN_VALUE;// 获取当前时间long now = System.currentTimeMillis();N selectedNode = null;WeightedRoundRobin selectedWRR = null;// 下面这个循环主要做了这样几件事情：// 1. 遍历 Node 列表，检测当前 Node 是否有相应的 WeightedRoundRobin，没有则创建// 2. 检测 Node 权重是否发生了变化，若变化了，则更新 WeightedRoundRobin 的 weight 字段// 3. 让 current 字段加上自身权重，等价于 current += weight// 4. 设置 lastUpdate 字段，即 lastUpdate = now// 5. 寻找具有最大 current 的 Node，以及 Node 对应的 WeightedRoundRobin，// 暂存起来，留作后用// 6. 计算权重总和for (N node : nodes) {int hashCode = node.hashCode();WeightedRoundRobin weightedRoundRobin = weightMap.get(hashCode);int weight = node.getWeight();if (weight < 0) {weight = 0;}// 检测当前 Node 是否有对应的 WeightedRoundRobin，没有则创建if (weightedRoundRobin == null) {weightedRoundRobin = new WeightedRoundRobin();// 设置 Node 权重weightedRoundRobin.setWeight(weight);// 存储 url 唯一标识 identifyString 到 weightedRoundRobin 的映射关系weightMap.putIfAbsent(hashCode, weightedRoundRobin);weightedRoundRobin = weightMap.get(hashCode);}// Node 权重不等于 WeightedRoundRobin 中保存的权重，说明权重变化了，此时进行更新if (weight != weightedRoundRobin.getWeight()) {weightedRoundRobin.setWeight(weight);}// 让 current 加上自身权重，等价于 current += weightlong current = weightedRoundRobin.increaseCurrent();// 设置 lastUpdate，表示近期更新过weightedRoundRobin.setLastUpdate(now);// 找出最大的 currentif (current > maxCurrent) {maxCurrent = current;// 将具有最大 current 权重的 Node 赋值给 selectedNodeselectedNode = node;// 将 Node 对应的 weightedRoundRobin 赋值给 selectedWRR，留作后用selectedWRR = weightedRoundRobin;}// 计算权重总和totalWeight += weight;}// 对 weightMap 进行检查，过滤掉长时间未被更新的节点。// 该节点可能挂了，nodes 中不包含该节点，所以该节点的 lastUpdate 长时间无法被更新。// 若未更新时长超过阈值后，就会被移除掉，默认阈值为60秒。if (!updateLock.get() && nodes.size() != weightMap.size()) {if (updateLock.compareAndSet(false, true)) {try {// 遍历修改，即移除过期记录weightMap.entrySet().removeIf(item -> now - item.getValue().getLastUpdate() > RECYCLE_PERIOD);} finally {updateLock.set(false);}}}if (selectedNode != null) {// 让 current 减去权重总和，等价于 current -= totalWeightselectedWRR.decreaseCurrent(totalWeight);// 返回具有最大 current 的 Nodereturn selectedNode;}// should not happen herereturn nodes.get(0);}protected static class WeightedRoundRobin {// 服务提供者权重private int weight;// 当前权重private AtomicLong current = new AtomicLong(0);// 最后一次更新时间private long lastUpdate;public long increaseCurrent() {// current = current + weight；return current.addAndGet(weight);}public long decreaseCurrent(int total) {// current = current - total;return current.addAndGet(-1 * total);}public int getWeight() {return weight;}public void setWeight(int weight) {this.weight = weight;// 初始情况下，current = 0current.set(0);}public AtomicLong getCurrent() {return current;}public void setCurrent(AtomicLong current) {this.current = current;}public long getLastUpdate() {return lastUpdate;}public void setLastUpdate(long lastUpdate) {this.lastUpdate = lastUpdate;}}}
特点：根据候选服务器当前的请求连接数，动态分配。 场景：适用于对系统负载较为敏感或请求连接时长相差较大的场景。
public class LeastActiveLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {private final Random random = new Random();protected N doSelect(List<N> nodes, String ip) {int length = nodes.size();// 最小的活跃数int leastActive = -1;// 具有相同“最小活跃数”的服务者提供者（以下用 Node 代称）数量int leastCount = 0;// leastIndexs 用于记录具有相同“最小活跃数”的 Node 在 nodes 列表中的下标信息int[] leastIndexs = new int[length];int totalWeight = 0;// 第一个最小活跃数的 Node 权重值，用于与其他具有相同最小活跃数的 Node 的权重进行对比，// 以检测是否“所有具有相同最小活跃数的 Node 的权重”均相等int firstWeight = 0;boolean sameWeight = true;// 遍历 nodes 列表for (int i = 0; i < length; i++) {N node = nodes.get(i);// 发现更小的活跃数，重新开始if (leastActive == -1 || node.getActive() < leastActive) {// 使用当前活跃数更新最小活跃数 leastActiveleastActive = node.getActive();// 更新 leastCount 为 1leastCount = 1;// 记录当前下标值到 leastIndexs 中leastIndexs[0] = i;totalWeight = node.getWeight();firstWeight = node.getWeight();sameWeight = true;// 当前 Node 的活跃数 node.getActive() 与最小活跃数 leastActive 相同} else if (node.getActive() == leastActive) {// 在 leastIndexs 中记录下当前 Node 在 nodes 集合中的下标leastIndexs[leastCount++] = i;// 累加权重totalWeight += node.getWeight();// 检测当前 Node 的权重与 firstWeight 是否相等，// 不相等则将 sameWeight 置为 falseif (sameWeight && i > 0&& node.getWeight() != firstWeight) {sameWeight = false;}}}// 当只有一个 Node 具有最小活跃数，此时直接返回该 Node 即可if (leastCount == 1) {return nodes.get(leastIndexs[0]);}// 有多个 Node 具有相同的最小活跃数，但它们之间的权重不同if (!sameWeight && totalWeight > 0) {// 随机生成一个 [0, totalWeight) 之间的数字int offsetWeight = random.nextInt(totalWeight);// 循环让随机数减去具有最小活跃数的 Node 的权重值，// 当 offset 小于等于0时，返回相应的 Nodefor (int i = 0; i < leastCount; i++) {int leastIndex = leastIndexs[i];// 获取权重值，并让随机数减去权重值offsetWeight -= nodes.get(leastIndex).getWeight();if (offsetWeight <= 0) {return nodes.get(leastIndex);}}}// 如果权重相同或权重为0时，随机返回一个 Nodereturn nodes.get(leastIndexs[random.nextInt(leastCount)]);}}
特点：保证特定用户总是请求到相同的服务器，若服务器宕机，会话会丢失。
public class IpHashLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {protected N doSelect(List<N> nodes, String ip) {if (StrUtil.isBlank(ip)) {ip = "127.0.0.1";}int length = nodes.size();int index = hash(ip) % length;return nodes.get(index);}public int hash(String text) {return HashUtil.fnvHash(text);}}
用户 ID 请求方 IP 请求服务名称，参数列表构成的串
注意：因为 一致性哈希分区 的这些缺点，一些分布式系统采用 虚拟槽 对 一致性哈希 进行改进，比如 Dynamo 系统。
public class ConsistentHashLoadBalance<N extends Node> extends BaseLoadBalance<N> implements LoadBalance<N> {private final ConcurrentMap<String, ConsistentHashSelector<?>> selectors = new ConcurrentHashMap<>();("unchecked")protected N doSelect(List<N> nodes, String ip) {// 分片数，这里设为节点数的 4 倍Integer replicaNum = nodes.size() * 4;// 获取 nodes 原始的 hashcodeint identityHashCode = System.identityHashCode(nodes);// 如果 nodes 是一个新的 List 对象，意味着节点数量发生了变化// 此时 selector.identityHashCode != identityHashCode 条件成立ConsistentHashSelector<N> selector = (ConsistentHashSelector<N>) selectors.get(ip);if (selector == null || selector.identityHashCode != identityHashCode) {// 创建新的 ConsistentHashSelectorselectors.put(ip, new ConsistentHashSelector<>(nodes, identityHashCode, replicaNum));selector = (ConsistentHashSelector<N>) selectors.get(ip);}// 调用 ConsistentHashSelector 的 select 方法选择 Nodereturn selector.select(ip);}/*** 一致性哈希选择器*/private static final class ConsistentHashSelector<N extends Node> {/*** 存储虚拟节点*/private final TreeMap<Long, N> virtualNodes;private final int identityHashCode;/*** 构造器** @param nodes 节点列表* @param identityHashCode hashcode* @param replicaNum 分片数*/ConsistentHashSelector(List<N> nodes, int identityHashCode, Integer replicaNum) {this.virtualNodes = new TreeMap<>();this.identityHashCode = identityHashCode;// 获取虚拟节点数，默认为 100if (replicaNum == null) {replicaNum = 100;}for (N node : nodes) {for (int i = 0; i < replicaNum / 4; i++) {// 对 url 进行 md5 运算，得到一个长度为16的字节数组byte[] digest = md5(node.getUrl());// 对 digest 部分字节进行 4 次 hash 运算，得到四个不同的 long 型正整数for (int j = 0; j < 4; j++) {// h = 0 时，取 digest 中下标为 0 ~ 3 的4个字节进行位运算// h = 1 时，取 digest 中下标为 4 ~ 7 的4个字节进行位运算// h = 2, h = 3 时过程同上long m = hash(digest, j);// 将 hash 到 node 的映射关系存储到 virtualNodes 中，// virtualNodes 需要提供高效的查询操作，因此选用 TreeMap 作为存储结构virtualNodes.put(m, node);}}}}public N select(String key) {// 对参数 key 进行 md5 运算byte[] digest = md5(key);// 取 digest 数组的前四个字节进行 hash 运算，再将 hash 值传给 selectForKey 方法，// 寻找合适的 Nodereturn selectForKey(hash(digest, 0));}private N selectForKey(long hash) {// 查找第一个大于或等于当前 hash 的节点Map.Entry<Long, N> entry = virtualNodes.ceilingEntry(hash);// 如果 hash 大于 Node 在哈希环上最大的位置，此时 entry = null，// 需要将 TreeMap 的头节点赋值给 entryif (entry == null) {entry = virtualNodes.firstEntry();}// 返回 Nodereturn entry.getValue();}}/*** 计算 hash 值*/public static long hash(byte[] digest, int number) {return (((long) (digest[3 + number * 4] & 0xFF) << 24)| ((long) (digest[2 + number * 4] & 0xFF) << 16)| ((long) (digest[1 + number * 4] & 0xFF) << 8)| (digest[number * 4] & 0xFF))& 0xFFFFFFFFL;}/*** 计算 MD5 值*/public static byte[] md5(String value) {MessageDigest md5;try {md5 = MessageDigest.getInstance("MD5");} catch (NoSuchAlgorithmException e) {throw new IllegalStateException(e.getMessage(), e);}md5.reset();byte[] bytes = value.getBytes(StandardCharsets.UTF_8);md5.update(bytes);return md5.digest();}}
四、参考资料
END
猜你喜欢
评论