负载均衡，含义就是根据一定算法将负载（工作任务）进行平衡，分摊到多个操作单元上运行、执行，常见的为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
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
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;- 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技术
虚拟IP
- 在TCP / IP架构下，所有想上网的电脑，不论以何种形式连上网络，都不需要有一个唯一的IP地址。事实上IP地址是主机硬件物理地址的一种抽象。
简单来说地址分为两种
- MAC物理地址
- IP逻辑地址
虚拟IP是一个未分配给真实主机的IP，也就是说对外提供的服务器的主机除了有一个真实IP还有一个虚IP，这两个IP中的任意一个都可以连接到这台主机。
- 通过虚拟IP对应真实主机的MAC地址实现
- 虚拟IP一般用作达到高可用的目的，比如让所有项目中的数据库链接配置都是这个虚拟IP，当主服务器发生故障无法对外提供服务时，动态将这个虚IP切换到备用服务器。
虚拟IP原理
- ARP是地址解析协议，作用为将一个IP地址转换为MAC地址。
- 每台主机都有ARP高速缓存，存储同一个网络内IP地址与MAC地址的映射关系，主机发送数据会先从这个缓存中查3目标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 实现一个简单的负载均衡
预期效果
- 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工具函数
监听请求并代理
- 获取需要传入getURL工具函数的参数。
- 通过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(302, 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
主进程流程
根据配置文件，创建相应数量服务节点。
- 创建进程
- 监听父线程消息（服务节点监听端口），并转发给子进程。
- 监听子进程消息，并转发给父线程（建立连接、断开连接事件）。
- 监听异常退出并重新建立。
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：数据统计对象（提供给各算法使用 / 展示数据）
- count：处理请求数
- costTime：响应时间
- connection：实时连接数
add方法
- 增加连接数和实时连接数
sub方法
- 减少实时连接数
updateCostTime方法
- 更新响应时间
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;
}
}最终效果
小作业
六、知识扩展
cluster多进程为什么可以监听一个端口?
- 通过cluster.isMaster判断是否为主进程，主进程不负责任务处理，只负责管理和调度工作子进程。
master主进程启动了一个TCP服务器，真正监听端口的只有这个TCP服务器。请求触发了这个TCP服务器的
connection事件后，通过句柄转发（IPC）给工作进程处理。- 句柄转发可转发TCP服务器、TCP套接字、UDP套接字、IPC管道
- IPC只支持传输字符串，不支持传输对象（可序列化）。
- 转发流程：父进程发送 -> stringfy && send(fd) -> IPC -> get(fd) && parse -> 子进程接收
- fd为句柄文件描述符。
如何选择工作进程?
- cluster模块内置了RoundRobin算法，轮询选择工作进程。
为什么不直接用cluster进行负载均衡?
- 手动实现可根据不同场景选择不同的负载均衡算法。
Node怎么实现进程间通信的？
常见的进程间通信方式
管道通信
- 匿名管道
- 命名管道
- 信号量
- 共享内存
- Socket
- 消息队列
- Node中实现IPC通道是依赖于libuv。Windows下由命名管道实现，*nix系统则采用Domain Socket实现。
- 表现在应用层上的进程间通信只有简单的message事件和send()方法，接口十分简洁和消息化。
IPC管道是如何建立的？
- 父进程先通过环境变量告知子进程管道的文件描述符
- 父进程创建子进程
- 子进程启动，通过文件描述符连接已存在的IPC管道，与父进程建立连接。
多进程 VS 多线程
多进程
- 数据共享复杂，需要IPC。数据是分开的，同步简单。
- 占用内存多，CPU利用率低。
- 创建销毁复杂，速度慢
- 进程独立运行，不会相互影响
- 可用于多机多核分布式，易于扩展
多线程
- 共享进程数据，数据共享简单，同步复杂。
- 占用内存少，CPU利用率高。
- 创建销毁简单，速度快。
- 线程同呼吸共命运。
- 只能用于多核分布式。
七、由本次分享产生的一些想法
- Node.js非阻塞异步I/O速度快，前端扩展服务端业务？
企业实践，说明Node还是可靠的？
- 阿里Node中台架构
- 腾讯CloudBase云开发Node
- 大量Node.js全栈工程师岗位
Node计算密集型不友好？
- Serverless盛行，计算密集型用C++/Go/Java编写，以Faas / RPC 的方式调用。
Node生态不如其他成熟的语言
- 阿里输出了Java生态
- 是不是可以看准趋势，打造Node生态以增强团队影响力。
讨论
八、参考资料
- 健康检查概述 - 负载均衡
- 《深入浅出Node.js》
- Node.js (nodejs.cn)
- 深入理解Node.js 中的进程与线程