CircuitBreaker(熔断器) Node.js 实现

最近业务需求, 需要对服务进行限流. 查阅了一些资料, 发现没有现成好用的. 所以打算自己撸个熔断器.

我们的 Web 服务是用 Node.js 写的. 不了解熔断器是啥子东西的同学, 可以谷百3搜下. 下面我说下思路:

首先定义一个 CircuitBreaker, 下面是它的属性:

  1. state: CircuitBreaker 当前状态, 有三种状态:
    • CloseState: 闭合状态, 此时正常提供服务;
    • OpenState: 断开状态, 此时拒绝服务请求;
    • HalfOpenState: 半开状态, 是 CloseState 和 OpenState 的中间态, 可提供服务, 但会根据服务请求的数量, 向上述两种状态转换
  2. counter: 计数器, 存储请求的次数, CircuitBreaker 根据请求的次数去切换状态
  3. thresholdForOpen: CircuitBreaker 切换到 OpenState 的阈值
  4. idleTimeForOpen: CircuitBreaker 切换到 OpenState 保持的时间
  5. thresholdForHalfOpen: CircuitBreaker 切换到 HalfOpenState 的阈值

下面我们首先来实现一个单机版的熔断器

我们先来看 CircuitBreaker 的代码:

// circuit_breaker.js
const logger = require('../util/logger');
const {CloseState, HalfOpenState, OpenState} = require('./state');
const {Counter} = require('./counter');


class CircuitBreaker {

    /**
     * @param thresholdForOpen {string} format: '600/60'
     *              '600/60' for instance, it means maximum allowable request is 600 times per 60 seconds, or the breaker will switch to OpenState
     * @param idleTimeForOpen {number} unit: second
     *              600 for instance, it means if the breaker switched to OpenState, it will keep for 600 seconds
     * @param thresholdForHalfOpen {string} format: '300/60'
     *              '300/60' for instance, it means the breaker will switch to OpenState if the maximum number of requests exceeds 300 per 60 seconds,
     *              or the breaker switch to CloseState
     */
    constructor(thresholdForOpen = '600/60', idleTimeForOpen = 5 * 60, thresholdForHalfOpen = '300/60') {
        this.idleTimeForOpen = idleTimeForOpen;
        this.thresholdForOpen = thresholdForOpen.split('/');
        this.thresholdForHalfOpen = thresholdForHalfOpen.split('/');
        this.counter = new Counter();   // max times for each 60s
        this.state = new CloseState();  // default state
    }

    getState() {
        return this.state;
    }

    setState(state) {
        logger.info(`switch state from ${this.getState().getName()} to ${state.getName()}`);
        this.state = state;
    }

    reset() {
        this.counter.reset();
    }

    canPass() {
        return this.getState().canPass(this);
    }

    count() {
        // 计数器 +1, 同时让 当前的 state 去做条件校验
        this.counter.increase();
        this.getState().checkout(this);
    }

    getCount() {
        return this.counter.get();
    }
}


module.exports = CircuitBreaker;

CircuitBreaker 中有个 Counter, 下面是其实现. 由于是单机版的, 我们存到本机内存就好了, 如果是多台机器, 可以实现这个 Counter, 将数值存在 redis 就好了

// counter.js
class Counter {
    constructor(num = 0) {
        this.num = num;
    }

    get() {
        return this.num;
    }

    increase() {
        this.num += 1;
    }

    reset() {
        this.num = 0;
    }
}

module.exports = {
    Counter: Counter,
};

最后我们看下 state 相关代码

// state.js
const logger = require('../util/logger');

/**
 * 抽象 state 父类
 */
class AbstractState {

    constructor(time = Date.now()) {
        this.startTime = time;
        logger.info(`${this.getName()} --> startTime = ${this.startTime / 1000}`);
    }

    getName() {
        return this.constructor.name;
    }

    canPass() {
        return true;
    }

    checkout(breaker) {
    }
}


/**
 * 闭合
 */
class CloseState extends AbstractState {
    constructor() {
        super();
    }

    canPass(breaker) {
        return true;
    }

    checkout(breaker) {
        let period = breaker.thresholdForOpen[1] * 1000;
        let now = Date.now();
        if (now >= this.startTime + period) { // 过了这段校验时间, 清零等待重新开始
            this.startTime = Date.now();
            breaker.reset();
        }

        logger.info('checkout -->  = ', breaker.getCount());
        if (breaker.getCount() >= breaker.thresholdForOpen[0]) { // 在这段校验时间内, 超过断路阈值, 切换到 `OpenState`
            breaker.reset();
            breaker.setState(new OpenState())
        }
    }
}


/**
 * 半开
 */
class HalfOpenState extends AbstractState {
    constructor() {
        super();
    }

    canPass(breaker) {
        let limit = breaker.thresholdForHalfOpen[0];
        return breaker.getCount() <= limit;
    }

    checkout(breaker) {
        logger.info('checkout --> count = ', breaker.getCount());
        let period = breaker.thresholdForHalfOpen[1] * 1000;
        let now = Date.now();
        if (now >= this.startTime + period) {
            breaker.reset();
            if (breaker.getCount() > breaker.thresholdForHalfOpen[0]) { // 依然超过断路阈值, 切到 `OpenState`
                breaker.setState(new OpenState());
            } else { // 低于断路阈值, 切到 `CloseState`
                breaker.setState(new CloseState());
            }
        }
    }

}

/**
 * 断路
 */
class OpenState extends AbstractState {
    constructor() {
        super();
    }

    canPass() {
        return false;
    }

    checkout(breaker) {
        let period = breaker.idleTimeForOpen * 1000;
        let now = Date.now();
        if (now >= this.startTime + period) { // 过了这段校验时间, 切换到 `HalfOpenState`
            breaker.reset();
            breaker.setState(new HalfOpenState());
        }
    }
}


module.exports = {
    AbstractState: AbstractState,
    CloseState: CloseState,
    HalfOpenState: HalfOpenState,
    OpenState: OpenState,
};

那么上面我们就实现了一个基本的熔断器. 下面我们来写个 Web 服务测试下, 我就用 restify 写了.

// app2.js
const AbstractServer = require('./abstract_server');
const CircuitBreaker = require('./breaker/circuit_breaker');
const logger = require('./util/logger');


class App extends AbstractServer {

    constructor(name, version, port) {
        super(name, version, port);
        // 为方便测试, 观察效果, 我们使用较小的阈值
        this.breaker = new CircuitBreaker('10/10', 20, '5/10');
    }

    checkFlow(req, res, next) {
        logger.debug('come into check flow');

        // 调用真实的业务逻辑之前, 先让熔断器计数并校验
        this.breaker.count();
        if (this.breaker.canPass()) {
            next();
        } else {
            res.end('reject\n');
        }
    }

    setPlugin() {
        super.setPlugin();
        this.server.use(this.checkFlow.bind(this));
    }


    initApi() {
        this.server.get('/test', this.test.bind(this));
    }

    test(req, res, next) {
        logger.debug('come into test');
        res.end('ok\n');
    }
}

let app = new App('test_server', '1.0.0', 6666);
app.run();

上面的代码在本地开启了一个服务, 提供了一个 get 接口, 可通过 curl http://localhost:6666/test 访问.
看日志也不是很直观, 我描述下上述测试的现象. 当服务开启后, 执行 curl http://localhost:6666/test

  1. 开始我可以正常访问, 返回 'ok';
  2. 加快访问速度, 当 10s 内的请求参数超过 10次 后,访问就会被拒绝, 返回 'reject', 熔断器会切换到 OpenState;
  3. 20s 后, 我可以正常访问, 此时熔断器切换到 HalfOpenState;
    • 在 HalfOpenState 我过量访问, 使其超过 5次/10s, 熔断器重新切换到 OpenState;
    • 在 HalfOpenState 我访问频率低于 5次/10s, 熔断器就切换到 CloseState;

上面会对所有请求限流, 我的需求实际上是想要针对部分调用者限流, 所以会针对不同的调用者, 有不同的熔断器

const AbstractServer = require('./abstract_server');
const CircuitBreaker = require('./breaker/circuit_breaker');
const logger = require('./util/logger');

let globalBreakers = new Map();

// setInterval(() => {
//     console.log('global -->  = ', globalBreakers);
// }, 3000);


class App extends AbstractServer {

    constructor(name, version, port) {
        super(name, version, port);
    }

    checkFlow(req, res, next) {
        logger.debug('come into check flow');
        let appid = req.query['appid'];
        if (!appid) {
            logger.error('no appid');
            res.end('no appid\n');
            return;
        }
        logger.debug('checkFlow --> appid = ' + appid);

        let breaker;
        if (globalBreakers.has(appid)) {
            breaker = globalBreakers.get(appid);
        } else {
            breaker = new CircuitBreaker('10/10', 20, '5/10');
            globalBreakers.set(appid, breaker);
        }

        breaker.count();
        if (breaker.canPass()) {
            next();
        } else {
            res.end('reject\n');
        }
    }

    setPlugin() {
        super.setPlugin();
        this.server.use(this.checkFlow.bind(this));
    }


    initApi() {
        this.server.get('/test', this.test.bind(this));
    }

    test(req, res, next) {
        logger.debug('come into test');
        res.end('ok\n');
    }
}

let app = new App('test_server', '1.0.0', 6666);
app.run();

上面代码中, 可以通过 curl http://localhost:6666/test?appid=myid 做测试, 不同的 appid, 会生成不同的 CircuitBreaker, 然后存到一个全局的 map 中, 这样就做到了针对不同 appid 的限流了.

以上思路安全不是自己所想, 抄自 线上防雪崩利器——熔断器设计原理与实现, 感谢作者. 不过原文是 Java 版本的, 而且文中的代码都是图片截屏, 本猿开始不太理解思路, 所以抄了一遍, 连同 Node.js 版本, 放到了 这里. 原作者是用来防雪崩的, 我整理了下思路, 改成 Node.js 版本, 既能用来限流, 又能用来防雪崩.

当然, 上面的代码写得非常的 Java, 本猿虽然现在主要用动态类语言, 但是依然用 Java 那一套约束自己. 因为当我想要看一个新项目的代码的时候, 只有 Java 能快速让我找到真相. 即便是动态语言里面的一些牛逼的开源库, 你想要读下源码, 依然会因为动态语言的特性, 让人困惑不已.
比如说 一个 constructor(options){}, 这是 Node.js 和 python 里面比较常见的构造器方式, 但是如果没有文档, 鬼知道 options 里面有哪些可用属性. 但是如果在 Java 里面, instance.setXxx(), 一切那么显而易见. 对.

以上, 算不上黑动态语言, 但是想要一堆人一起把一个动态语言的项目写得漂亮, 还是略难的. 本猿写的时候, 通常会遵循一些之前写 Java 的套路, 这样感觉又丧失了动态语言的活力.

既如此, 便不敢多想, 只是多写多看, 和和稀泥, 如果有幸找到法门, 一切自然会通透吧.

©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • Spring Cloud
    Spring Cloud为开发人员提供了快速构建分布式系统中一些常见模式的工具（例如配置管理，服务发现，断路器，智...
    卡卡罗2017阅读 135,292评论 19赞 139
  • 90年前后出生的人，都混成什么样子了？
    01 听他们说原来同班的高中同学今天结婚了。 我听完相当诧异，我说才多大啊，这么早就结婚了？自己还是个孩子，怎么承...
    不穿长袍的孔已己阅读 2,761评论 0赞 1
  • 烧纸钱
    大明是我的好朋友。 有福同享，有难同当这句话用在我们身上最合适不过。他有福一定与我同享，我有难也一定与他同当。 大...
    汪伦居然追上了李白阅读 4,075评论 0赞 1
  • 如何对比JS中两个对象是否相等
    ① 方法一：通过JSON.stringify(obj)来判断两个对象转后的字符串是否相等 优点：用法简单，对于顺序...
    木A木阅读 48,594评论 14赞 20
  • 时间不够用
    前些日子他们在朋友圈发80后对应的岁数，我看到我的，不由一惊，原来我都这么老了 不忍直视的数字： 1980—39岁...
    牛羊妈阅读 1,274评论 0赞 0