[深入浅出LB]手把手带你实现一个负载均衡器

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 健康监测步骤（七层）

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

TCP健康检查步骤（四层）

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

UDP健康检查步骤（四层）

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

四、VIP技术

Vrtual IP

虚拟IP

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

虚拟IP原理

1. ARP是地址解析协议，作用为将一个IP地址转换为MAC地址。
2. 每台主机都有ARP高速缓存，存储同一个网络内IP地址与MAC地址的映射关系，主机发送数据会先从这个缓存中查询目标IP对应MAC地址，向这个MAC地址发送数据。操作系统自动维护这个缓存。
3. 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

1. 初始化负载均衡统计对象balanceDataBase

   • balanceDataBase是一个DataBase类实例，用于统计负载均衡数据（后续会讲到）.

2. 运行均衡器

   • 多进程模型，提供并发能力。

3. 运行后端服务节点

   • 多线程+多进程模型，运行多个服务节点并提供并发能力。

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函数）

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

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函数）

1. 创建线程

2. 解析需要监听的端口

3. 向子线程通信，发送需要监听的端口

4. 通过线程通信，监听子线程事件

   • 监听连接事件，并触发处理函数。

   • 监听断开连接事件并触发处理函数。

   • 用于统计负载均衡分布和实时连接数。

5. 监听异常退出并重新创建，线程守护。

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

1. 获取getURL工具函数
2. 监听请求并代理
   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

1. 通过进程通信，向父进程发送获取消息。
2. 同时开始监听父进程通信消息，接收后使用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. 根据配置文件，创建相应数量服务节点。
   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）

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

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类

• 成员：

1. status：任务队列状态
2. urlCollect：数据统计对象（提供给各算法使用 / 展示数据）
   1. count：处理请求数
   2. costTime：响应时间
   3. connection：实时连接数
3. add方法
   1. 增加连接数和实时连接数
4. sub方法
   1. 减少实时连接数
5. 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多进程为什么可以监听一个端口?

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

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

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

多进程 VS 多线程

多进程

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

多线程

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

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

欢迎留言讨论

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

• Node.js 做 Web 后端优势为什么这么大？ - 知乎 (zhihu.com)

八、参考资料

1. 健康检查概述 - 负载均衡
2. 《深入浅出Node.js》
3. Node.js (nodejs.cn)
4. 深入理解Node.js 中的进程与线程

掘金：前端LeBron

知乎：前端LeBron

持续分享技术博文，关注微信公众号👇🏻