Load Balance 负载均衡 🌡

简介

负载均衡，含义就是根据一定算法将负载（工作任务）进行平衡，分摊到多个操作单元上运行、执行，常见的为 Web 服务器、企业核心应用服务器和其他主要任务服务器等，从而协同完成工作任务。负载均衡在原有的网络结构上提供了一种透明且有效的的方法扩展服务器和网络设备的带宽、加强网络数据处理能力、增加吞吐量、提高网络的可用性和灵活性，同时承受住更大的并发量级。

简单来说就是将大量的并发请求处理转发给多个后端节点处理，减少工作响应时间。

避免资源浪费
避免服务不可用

一、分类

四层（传输层）

四层即 OSI 七层模型中的传输层，有 TCP、UDP 协议，这两种协议中包含源 IP、目标 IP 以外，还包含源端口号及目标端口号。四层负载均衡在接收到客户端请求后，通过修改报文的地址信息（IP + PORT）将流量转发到应用服务器。

七层（应用层）

代理负载均衡

七层即 OSI 七层模型中的应用层，应用层协议较多，常用的为 HTTP/HTTPS。七层负载均衡可以给予这些协议来负载。这些应用层协议中会包含很多有意义的内容。比如同一个 Web 服务器的负载均衡，除了根据 IP + PORT 进行负载均衡，还可以根据七层的 URL、Cookie、浏览器类别、语言、请求类型来决定。

四层负载均衡的本质是转发，七层负载均衡的本质是内容交换和代理。

	四层负载均衡	七层负载均衡
基于	IP + PORT	URL 或主机 IP
类似	路由器	代理服务器
复杂度	低	高
性能	高，无需解析内容	中，需算法识别 URL Header、Cookie 等
安全性	低，无法识别 DDoS 攻击	高，可防御 SYN Flood 攻击
扩展功能	无	内容缓存、图片防盗链等

二、常见算法

前置数据结构

interface urlObj{
  url:string,
  weight:number // 仅在权重轮询时生效
}
urlDesc: urlObj[]

interface urlCollectObj{
  count: number, // 连接数
  costTime: number, // 响应时间
  connection: number, // 实时连接数
}
urlCollect: urlCollectObj[]

Random

随机

const Random = (urlDesc) => {
  let urlCollect = [];

  //  收集url
  urlDesc.forEach((val) => {
    urlCollect.push(val.url);
  });

  return () => {
    //  生成随机数下标返回相应URL
    const pos = parseInt(Math.random() * urlCollect.length);
    return urlCollect[pos];
  };
};

module.exports = Random;

Weighted Round Robin

权重轮询算法

const WeiRoundRobin = (urlDesc) => {
  let pos = 0,
    urlCollect = [],
    copyUrlDesc = JSON.parse(JSON.stringify(urlDesc));

  // 根据权重收集url
  while (copyUrlDesc.length > 0) {
    for (let i = 0; i < copyUrlDesc.length; i++) {
      urlCollect.push(copyUrlDesc[i].url);
      copyUrlDesc[i].weight--;
      if (copyUrlDesc[i].weight === 0) {
        copyUrlDesc.splice(i, 1);
        i--;
      }
    }
  }
  // 轮询获取URL函数
  return () => {
    const res = urlCollect[pos++];
    if (pos === urlCollect.length) {
      pos = 0;
    }
    return res;
  };
};

module.exports = WeiRoundRobin;

IP Hash & URL Hash

源 IP / URL Hash

const { Hash } = require('../util');

const IpHash = (urlDesc) => {
  let urlCollect = [];

  for (const key in urlDesc) {
    // 收集url
    urlCollect.push(urlDesc[key].url);
  }

  return (sourceInfo) => {
    // 生成Hash十进制数值
    const hashInfo = Hash(sourceInfo);
    // 取余为下标
    const urlPos = Math.abs(hashInfo) % urlCollect.length;
    // 返回
    return urlCollect[urlPos];
  };
};

module.exports = IpHash;

Consistent Hash

一致性 Hash

const { Hash } = require('../util');

const ConsistentHash = (urlDesc) => {
  let urlHashMap = {},
    hashCollect = [];

  for (const key in urlDesc) {
    // 收集urlHash进数组和生成HashMap
    const { url } = urlDesc[key];
    const hash = Hash(url);
    urlHashMap[hash] = url;
    hashCollect.push(hash);
  }
  // 将hash数组从小到大排序
  hashCollect = hashCollect.sort((a, b) => a - b);

  return (sourceInfo) => {
    // 生成Hash十进制数值
    const hashInfo = Hash(sourceInfo);
    // 遍历hash数组找到第一个比源信息hash值大的，并通过hashMap返回url
    hashCollect.forEach((val) => {
      if (val >= hashInfo) {
        return urlHashMap[val];
      }
    });
    // 没找大则返回最大的
    return urlHashMap[hashCollect[hashCollect.length - 1]];
  };
};

module.exports = ConsistentHash;

Least Connections

最小连接数

const leastConnections = () => {
  return (urlCollect) => {
    let min = Number.POSITIVE_INFINITY,
      url = '';

    // 遍历对象找到最少连接数的地址
    for (let key in urlCollect) {
      const val = urlCollect[key].connection;
      if (val < min) {
        min = val;
        url = key;
      }
    }
    // 返回
    return url;
  };
};

module.exports = leastConnections;

注：urlCollect 为负载均属数据统计对象，有以下属性

connection 实时连接数
count 处理请求次数
costTime 响应时间。

FAIR

最小响应时间

const Fair = () => {
  return (urlCollect) => {
    let min = Number.POSITIVE_INFINITY,
      url = '';

    // 找到耗时最少的url
    for (const key in urlCollect) {
      const urlObj = urlCollect[key];
      if (urlObj.costTime < min) {
        min = urlObj.costTime;
        url = key;
      }
    }
    // 返回
    return url;
  };
};

module.exports = Fair;

看到这里是不是感觉算法都挺简单的 🥱

期待一下模块四的实现吧 😏

三、健康监测

健康监测即对应用服务器的健康监测，为防止把请求转发到异常的应用服务器上，应使用健康监测策略。应对不同的业务敏感程度，可相应调整策略和频率。

HTTP / HTTPS 健康监测步骤（七层）

负载均衡节点向应用服务器发送 HEAD 请求。
应用服务器接收到 HEAD 请求后根据情况返回相应状态码。
若在超时时间内未收到返回的状态码，则判断为超时，健康检查失败。
若在超时时间内收到返回的状态码，负载均衡节点进行比对，判断健康检查是否成功。

TCP 健康检查步骤（四层）

负载均衡节点向内网应用服务器 IP + PORT 发 TCP SYN 请求数据包。
内网应用服务器收到请求后，若在正常监听，则返回 SYN + ACK 数据包。
若在超时时间内未收到返回的数据包，则判断服务无响应、健康检查失败，并向内网应用服务器发送 RST 数据包中断 TCP 连接。
若在超时时间内收到返回的数据包，则判定服务健康运行，发起 RST 数据包中断 TCP 连接。

UDP 健康检查步骤（四层）

负载均衡节点向内网应用服务器 IP + PORT 发送 UDP 报文。
若内网应用服务器未正常监听，则返回PORT XX unreachable的 ICMP 报错信息，反之为正常。
若在超时时间内收到了报错信息，则判断服务异常，健康检查失败。
若在超时时间内未收到报错信息，则判断服务健康运行。

四、VIP 技术

Vrtual IP

虚拟 IP

在 TCP / IP 架构下，所有想上网的电脑，不论以何种形式连上网络，都不需要有一个唯一的 IP 地址。事实上 IP 地址是主机硬件物理地址的一种抽象。
简单来说地址分为两种
- MAC 物理地址
- IP 逻辑地址
虚拟 IP 是一个未分配给真实主机的 IP，也就是说对外提供的服务器的主机除了有一个真实 IP 还有一个虚 IP，这两个 IP 中的任意一个都可以连接到这台主机。
- 通过虚拟 IP 对应真实主机的 MAC 地址实现
虚拟 IP 一般用作达到高可用的目的，比如让所有项目中的数据库链接配置都是这个虚拟 IP，当主服务器发生故障无法对外提供服务时，动态将这个虚 IP 切换到备用服务器。

虚拟 IP 原理

ARP 是地址解析协议，作用为将一个 IP 地址转换为 MAC 地址。
每台主机都有 ARP 高速缓存，存储同一个网络内 IP 地址与 MAC 地址的映射关系，主机发送数据会先从这个缓存中查询目标 IP 对应 MAC 地址，向这个 MAC 地址发送数据。操作系统自动维护这个缓存。
Linux 下可用 ARP 命令操作 ARP 高速缓存

比如存在主机 A（192.168.1.6）和主机 B（192.168.1.8）。A 作为对外服务的主服务器，B 作为备份机器，两台服务器之间通过 HeartBeat 通信。
即主服务器会定时给备份服务器发送数据包，告知主服务器正常，当备份服务器在规定时间内没有收到主服务器的 HeartBeat，会认为主服务器宕机。
此时备份服务器就升级为主服务器。
- 服务器 B 将自己的 ARP 缓存发送出去，告知路由器修改路由表，告知虚拟 IP 地址应该指向 192.168.1.8.
- 这时外接再次访问虚拟 IP 的时候，机器 B 就会变成主服务器，而 A 降级为备份服务器。
- 这样就完成了主从机器的切换，这一切对外都是无感知、透明的。

五、基于 nodejs 实现一个简单的负载均衡

预期效果

编辑 config.js 后npm run start即可启动均衡器和后端服务节点

urlDesc：后端服务节点配置对象，weight 仅在 WeightRoundRobin 算法时起作用
port：均衡器监听端口
algorithm：算法名称（模块二中的算法均已实现）
workerNum：后端服务端口开启进程数，提供并发能力。
balancerNum：均衡器端口开启进程数，提供并发能力。
workerFilePath：后端服务节点执行文件，推荐使用绝对路径。

const {ALGORITHM, BASE_URL} = require("./constant");

module.exports = {
    urlDesc: [
        {
            url: `${BASE_URL}:${16666}`,
            weight: 6,
        },
        {
            url: `${BASE_URL}:${16667}`,
            weight: 1,
        },
        {
            url: `${BASE_URL}:${16668}`,
            weight: 1,
        },
        {
            url: `${BASE_URL}:${16669}`,
            weight: 1,
        },
        {
            url: `${BASE_URL}:${16670}`,
            weight: 2,
        },
        {
            url: `${BASE_URL}:${16671}`,
            weight: 1,
        },
        {
            url: `${BASE_URL}:${16672}`,
            weight: 4,
        },
    ],
    port: 8080,
    algorithm: ALGORITHM.RANDOM,
    workerNum: 5，
    balancerNum: 5，
    workerFilePath:path.resolve(__dirname, "./worker.js")
}

架构设计图

先来看看主流程 main.js

初始化负载均衡统计对象 balanceDataBase
- balanceDataBase 是一个 DataBase 类实例，用于统计负载均衡数据（后续会讲到）.
运行均衡器
- 多进程模型，提供并发能力。
运行后端服务节点
- 多线程+多进程模型，运行多个服务节点并提供并发能力。

const { urlDesc, balancerNum } = require('./config');
const cluster = require('cluster');
const path = require('path');
const cpusLen = require('os').cpus().length;
const { DataBase } = require('./util');
const { Worker } = require('worker_threads');

const runWorker = () => {
  // 防止监听端口数 > CPU核数
  const urlObjArr = urlDesc.slice(0, cpusLen);
  // 初始化创建子线程
  for (let i = 0; i < urlObjArr.length; i++) {
    createWorkerThread(urlObjArr[i].url);
  }
};

const runBalancer = () => {
  // 设置子进程执行文件
  cluster.setupMaster({ exec: path.resolve(__dirname, './balancer.js') });
  // 初始化创建子进程
  let max;
  if (balancerNum) {
    max = balancerNum > cpusLen ? cpusLen : balancerNum;
  } else {
    max = 1;
  }
  for (let i = 0; i < max; i++) {
    createBalancer();
  }
};

// 初始化负载均衡数据统计对象
const balanceDataBase = new DataBase(urlDesc);
// 运行均衡器
runBalancer();
// 运行后端服务节点
runWorker();

创建均衡器（createBalancer 函数）

创建进程
监听进程通信消息
- 监听更新响应时间事件并执行更新函数
  - 用于 FAIR 算法（最小响应时间）。
- 监听获取统计对象事件并返回
监听异常退出并重新创建，进程守护。

const createBalancer = () => {
  // 创建进程
  const worker = cluster.fork();
  worker.on('message', (msg) => {
    // 监听更新响应时间事件
    if (msg.type === 'updateCostTime') {
      balanceDataBase.updateCostTime(msg.URL, msg.costTime);
    }
    // 监听获取url统计对象事件并返回
    if (msg.type === 'getUrlCollect') {
      worker.send({
        type: 'getUrlCollect',
        urlCollect: balanceDataBase.urlCollect,
      });
    }
  });
  // 监听异常退出事件并重新创建进程
  worker.on('exit', () => {
    createBalancer();
  });
};

创建后端服务节点（createWorkerThread 函数）

创建线程
解析需要监听的端口
向子线程通信，发送需要监听的端口
通过线程通信，监听子线程事件
- 监听连接事件，并触发处理函数。
- 监听断开连接事件并触发处理函数。
- 用于统计负载均衡分布和实时连接数。
监听异常退出并重新创建，线程守护。

const createWorkerThread = (listenUrl) => {
  // 创建线程
  const worker = new Worker(path.resolve(__dirname, './workerThread.js'));
  // 获取监听端口
  const listenPort = listenUrl.split(':')[2];
  // 向子线程发送要监听的端口号
  worker.postMessage({ type: 'port', port: listenPort });

  // 接收子线程消息统计进程被访问次数
  worker.on('message', (msg) => {
    // 监听连接事件并触发计数事件
    if (msg.type === 'connect') {
      balanceDataBase.add(msg.port);
    }
    // 监听断开连接事件并触发计数事件
    else if (msg.type === 'disconnect') {
      balanceDataBase.sub(msg.port);
    }
  });
  // 监听异常退出事件并重新创建进程
  worker.on('exit', () => {
    createWorkerThread(listenUrl);
  });
};

再来看看均衡器工作流程 balancer.js

获取 getURL 工具函数
监听请求并代理
1. 获取需要传入 getURL 工具函数的参数。
2. 通过 getURL 工具函数获取均衡代理目的地址 URL
3. 记录请求开始时间
4. 处理跨域
5. 返回响应
6. 通过进程通信，触发响应时间更新事件。

注 1：LoadBalance 函数即通过算法名称返回不同的 getURL 工具函数，各算法实现见模块二：常见算法

注 2：getSource 函数即处理参数并返回，getURL 为上面讲到的获取 URL 工具函数。

const cpusLen = require('os').cpus().length;
const LoadBalance = require('./algorithm');
const express = require('express');
const axios = require('axios');
const app = express();
const { urlFormat, ipFormat } = require('./util');
const { ALGORITHM, BASE_URL } = require('./constant');
const { urlDesc, algorithm, port } = require('./config');

const run = () => {
  // 获取转发URL工具函数
  const getURL = LoadBalance(urlDesc.slice(0, cpusLen), algorithm);
  // 监听请求并均衡代理
  app.get('/', async (req, res) => {
    // 获取需要传入的参数
    const source = await getSource(req);
    // 获取URL
    const URL = getURL(source);
    // res.redirect(301, URL) 重定向负载均衡
    // 记录请求开始时间
    const start = Date.now();
    // 代理请求
    axios.get(URL).then(async (response) => {
      // 获取负载均衡统计对象并返回
      const urlCollect = await getUrlCollect();
      // 处理跨域
      res.setHeader('Access-Control-Allow-Origin', '*');
      response.data.urlCollect = urlCollect;
      // 返回数据
      res.send(response.data);
      // 记录相应时间并更新
      const costTime = Date.now() - start;
      process.send({ type: 'updateCostTime', costTime, URL });
    });
  });
  // 负载均衡服务器开始监听请求
  app.listen(port, () => {
    console.log(`Load Balance Server Running at ${BASE_URL}:${port}`);
  });
};

run();

const getSource = async (req) => {
  switch (algorithm) {
    case ALGORITHM.IP_HASH:
      return ipFormat(req);
    case ALGORITHM.URL_HASH:
      return urlFormat(req);
    case ALGORITHM.CONSISTENT_HASH:
      return urlFormat(req);
    case ALGORITHM.LEAST_CONNECTIONS:
      return await getUrlCollect();
    case ALGORITHM.FAIR:
      return await getUrlCollect();
    default:
      return null;
  }
};

如何在均衡器中获取负载均衡统计对象 getUrlCollect

通过进程通信，向父进程发送获取消息。
同时开始监听父进程通信消息，接收后使用 Promise resovle 返回。

// 获取负载均衡统计对象
const getUrlCollect = () => {
  return new Promise((resolve, reject) => {
    try {
      process.send({ type: 'getUrlCollect' });
      process.on('message', (msg) => {
        if (msg.type === 'getUrlCollect') {
          resolve(msg.urlCollect);
        }
      });
    } catch (e) {
      reject(e);
    }
  });
};

如何实现服务节点并发 workerThread.js

使用多线程+多进程模型，为每个服务节点提供并发能力。

主进程流程

根据配置文件，创建相应数量服务节点。
1. 创建进程
2. 监听父线程消息（服务节点监听端口），并转发给子进程。
3. 监听子进程消息，并转发给父线程（建立连接、断开连接事件）。
4. 监听异常退出并重新建立。

const cluster = require('cluster');
const cpusLen = require('os').cpus().length;
const { parentPort } = require('worker_threads');
const { workerNum, workerFilePath } = require('./config');

if (cluster.isMaster) {
  // 创建工作进程函数
  const createWorker = () => {
    // 创建进程
    const worker = cluster.fork();
    // 监听父线程消息，并转发给子进程。
    parentPort.on('message', (msg) => {
      if (msg.type === 'port') {
        worker.send({ type: 'port', port: msg.port });
      }
    });
    // 监听子进程消息并转发给父线程
    worker.on('message', (msg) => {
      parentPort.postMessage(msg);
    });
    // 监听进程异常退出并重新创建
    worker.on('exit', () => {
      createWorker();
    });
  };
  // 按配置创建进程，但不可大于CPU核数
  let max;
  if (workerNum) {
    max = workerNum > cpusLen ? cpusLen : workerNum;
  } else {
    max = 1;
  }
  for (let i = 0; i < max; i++) {
    createWorker();
  }
} else {
  // 后端服务执行文件
  require(workerFilePath);
}

子进程流程 worker.js（config.workerFilePath）

通过进程间通信，向父进程发送消息，触发建立连接事件。
返回相应。
通过进程间通信，向父进程发送消息，触发断开连接事件。

var express = require('express');
var app = express();
let port = null;

app.get('/', (req, res) => {
  // 触发连接事件
  process.send({ type: 'connect', port });
  // 打印信息
  console.log('HTTP Version: ' + req.httpVersion);
  console.log('Connection PORT Is ' + port);

  const msg = 'Hello My PORT is ' + port;

  // 返回响应
  res.send({ msg });
  // 触发断开连接事件
  process.send({ type: 'disconnect', port });
});

// 接收主进通信消息中的端口口并监听
process.on('message', (msg) => {
  if (msg.type === 'port') {
    port = msg.port;
    app.listen(port, () => {
      console.log('Worker Listening ' + port);
    });
  }
});

最后来看看 DataBase 类

成员：

status：任务队列状态
urlCollect：数据统计对象（提供给各算法使用 / 展示数据）
1. count：处理请求数
2. costTime：响应时间
3. connection：实时连接数
add 方法
1. 增加连接数和实时连接数
sub 方法
1. 减少实时连接数
updateCostTime 方法
1. 更新响应时间

class DataBase {
  urlCollect = {};

  // 初始化
  constructor(urlObj) {
    urlObj.forEach((val) => {
      this.urlCollect[val.url] = {
        count: 0,
        costTime: 0,
        connection: 0,
      };
    });
  }

  //增加连接数和实时连接数
  add(port) {
    const url = `${BASE_URL}:${port}`;
    this.urlCollect[url].count++;
    this.urlCollect[url].connection++;
  }

  // 减少实时连接数
  sub(port) {
    const url = `${BASE_URL}:${port}`;
    this.urlCollect[url].connection--;
  }

  // 更新响应时间
  updateCostTime(url, time) {
    this.urlCollect[url].costTime = time;
  }
}

最终效果

做了个可视化图表来看均衡效果(Random)✔️

看起来均衡效果还不错 🧐

小作业

想手动实现一下负载均衡器 / 看看源码的同学都可以看看 👉🏻 代码仓库

六、知识扩展

cluster 多进程为什么可以监听一个端口?

通过 cluster.isMaster 判断是否为主进程，主进程不负责任务处理，只负责管理和调度工作子进程。
master 主进程启动了一个 TCP 服务器，真正监听端口的只有这个 TCP 服务器。请求触发了这个 TCP 服务器的connection事件后，通过句柄转发（IPC）给工作进程处理。
1. 句柄转发可转发 TCP 服务器、TCP 套接字、UDP 套接字、IPC 管道
2. IPC 只支持传输字符串，不支持传输对象（可序列化）。
3. 转发流程：父进程发送 -> stringfy && send(fd) -> IPC -> get(fd) && parse -> 子进程接收
4. fd 为句柄文件描述符。
如何选择工作进程?
1. cluster 模块内置了 RoundRobin 算法，轮询选择工作进程。
为什么不直接用 cluster 进行负载均衡?
1. 手动实现可根据不同场景选择不同的负载均衡算法。

Node 怎么实现进程间通信的？

常见的进程间通信方式
1. 管道通信
  1. 匿名管道
  2. 命名管道
2. 信号量
3. 共享内存
4. Socket
5. 消息队列
Node 中实现 IPC 通道是依赖于 libuv。Windows 下由命名管道实现，*nix 系统则采用 Domain Socket 实现。
表现在应用层上的进程间通信只有简单的 message 事件和 send()方法，接口十分简洁和消息化。
IPC 管道是如何建立的？
1. 父进程先通过环境变量告知子进程管道的文件描述符
2. 父进程创建子进程
3. 子进程启动，通过文件描述符连接已存在的 IPC 管道，与父进程建立连接。

多进程 VS 多线程

多进程

数据共享复杂，需要 IPC。数据是分开的，同步简单。
占用内存多，CPU 利用率低。
创建销毁复杂，速度慢
进程独立运行，不会相互影响
可用于多机多核分布式，易于扩展

多线程

共享进程数据，数据共享简单，同步复杂。
占用内存少，CPU 利用率高。
创建销毁简单，速度快。
线程同呼吸共命运。
只能用于多核分布式。

七、由本次分享产生的一些想法

欢迎留言讨论

Node.js 非阻塞异步 I/O 速度快，前端扩展服务端业务？
企业实践，说明 Node 还是可靠的？
1. 阿里 Node 中台架构
2. 腾讯 CloudBase 云开发 Node
3. 大量 Node.js 全栈工程师岗位
Node 计算密集型不友好？
1. Serverless 盛行，计算密集型用 C++/Go/Java 编写，以 Faas 的方式调用。
Node 生态不如其他成熟的语言
1. 阿里输出了 Java 生态
2. 是不是可以看准趋势，打造 Node 生态以增强团队影响力。
未雨绸缪，将 Node & 服务端业务知识加入学习计划 / 规划专题分享？
讨论