基于 Koa 的微服务 Node.js 框架设计思路与简单实现

发表于 3年以前 | 总阅读数：366 次

Koa 官网的介绍是这样介绍自己的：

Koa 是一个新的 web 框架，由 Express 幕后的原班人马打造，致力于成为 web 应用和 API 开发领域中的一个更小、更富有表现力、更健壮的基石。通过利用 async 函数，Koa 帮你丢弃回调函数，并有力地增强错误处理。Koa 并没有捆绑任何中间件，而是提供了一套优雅的方法，帮助您快速而愉快地编写服务端应用程序。

从上面的描述中我们可以知道，Koa 是一种简单好用的 Web 框架。它的特点是优雅、简洁、表达力强、自由度高。其本身代码只有 1000 多行，所有功能都可以通过插件的方式扩展，很符合 KISS 原则与 Unix 哲学。比较有名的 Node.js 业务框架 egg.js 就是是继承自Koa。

但 Koa 的劣势也很明显，就是太过自由，并没有内置过多的功能，比如常见的请求体解析、路由、模板渲染等功能都没有，需要加载第三方中间件来实现。另外 Koa 只支持 Http 服务，无法满足业务方对于 RPC 服务的需求。

本文将对基于 Koa 的微服务 Node.js 框架设计思路做一些思考与探究，并且对实现方面做一些简单补充。让我们先从 Koa 的核心思想与原理开始。

Koa的核心思想与最简实现

核心思想：AOP 面向切面编程

AOP技术的诞生并不算晚，早在1990年开始，来自Xerox Palo Alto Research Lab（即PARC）的研究人员就对面向对象思想的局限性进行了分析。他们研究出了一种新的编程思想，借助这一思想或许可以通过减少代码重复模块从而帮助开发人员提高工作效率。随着研究的逐渐深入，AOP也逐渐发展成一套完整的程序设计思想，各种应用AOP的技术也应运而生。

这个名词听起来很高大上，可能很多人都听过，但是又没有彻底搞懂，到底什么叫面向切面编程？这里先不解释 AOP 的具体含义，而是举个简单的例子。

农场的水果包装流水线一开始只有 采摘 - 清洗 - 贴标签

为了提高销量，想加上两道工序 分类 和 包装 但又不能干扰原有的流程，同时如果没增加收益可以随时撤销新增工序。

最后在流水线的中的空隙插上两个工人去处理，形成采摘 - 分类 - 清洗 - 包装 - 贴标签 的新流程，而且工人可以随时撤回。

上面所说的每一道工序，都可以看作是一个切面。

回到 AOP 的含义：就是在现有代码程序中，在程序的生命周期或横向流程中，加入或减去一个或多个功能，使原本功能不受影响。

核心原理：koa-compose + Node.js http

Koa 可以被拆解为如下公式：

Koa = Node.js原生http服务 + 中间件引擎koa-compose

通过把中间件用 Promise + async/await 的方式嵌套组合，Koa 实现了比 Express 的线性模型中间件多了一倍切面的洋葱模型中间件，所以 Koa 能非常方便地实现类似响应时间计算、日志打印、鉴权等等常用功能。

下面举一个 Koa 官网的 demo 例子，可以看到这些功能的具体实现是多么的简单：

const Koa = require('koa');
const app = new Koa();

// x-response-time
app.use(async (ctx, next) => {
  const start = Date.now();
  await next();
  const ms = Date.now() - start;
  ctx.set('X-Response-Time', `${ms}ms`);
});

// logger
app.use(async (ctx, next) => {
  const start = Date.now();
  await next();
  const ms = Date.now() - start;
  console.log(`${ctx.method} ${ctx.url} - ${ms}`);
});
// response
app.use(async ctx => {
  ctx.body = 'Hello World';
});

app.listen(3000);

Node.js 原生 http 不用多说，下面着重讲一下中间件引擎 koa-compose 的实现，源码非常精简，核心代码只有 30 行左右：

function compose(middleware) {
  // 如果middleware不是数组，或者元素不是函数，则抛异常
  if (!Array.isArray(middleware)) {
    throw new TypeError('Middleware stack must be an array!');
  }
  for (const fn of middleware) {
    if (typeof fn !== 'function') {
      throw new TypeError('Middleware must be composed of functions!');
    }
  }
  // 返回一个闭包函数
  return function (context, next) {
    // last called middleware #

    let index = -1;

    return dispatch(0);

    function dispatch(i) {
      if (i <= index) {
        return Promise.reject(new Error('next() called multiple times'));
      }

      index = i;

      let fn = middleware[i];

      if (i === middleware.length) {
        fn = next;
      }

      if (!fn) {
        return Promise.resolve();
      }

      try {
        // 将每一个 middleware 函数作为前一个函数的 next 参数

        return Promise.resolve(fn(context, dispatch.bind(null, i + 1)));
      } catch (err) {
        return Promise.reject(err);
      }
    }
  };
}

简化掉判断逻辑，compose执行后就是类似下面这样的结构：

// 这样就可能更好理解了。
// simpleKoaCompose
const [fn1, fn2, fn3] = stack;

const fnMiddleware = function (context) {
  return Promise.resolve(
    fn1(context, function next() {
      return Promise.resolve(
        fn2(context, function next() {
          return Promise.resolve(
            fn3(context, function next() {
              return Promise.resolve();
            })
          );
        })
      );
    })
  );
};

实际上 koa-compose 返回的是一个 Promise，从中间件（传入的数组）中取出第一个函数，传入context和第一个next函数来执行。

第一个 next 函数也返回一个 Promise，从中间件（传入的数组）中取出第二个函数，传入context和第二个next函数来执行。

第二个 next 函数也返回一个 Promise，从中间件（传入的数组）中取出第三个函数，传入context和第三个next函数来执行。

第三个...

以此类推。最后一个中间件中如果调用了 next 函数，则返回 Promise.resolve()。这样就把所有中间件串联起来了。类似栈的先进后出，每个中间件都有两个切面，这就是洋葱模型的实现原理。

Koa 最简实现


const http = require('http');

const Emitter = require('events');

const compose = require('koa-compose'); // 上面的 compose

/**
 * 通用上下文
 */

const context = {
  _body: null,

  get body() {
    return this._body;
  },

  set body(val) {
    this._body = val;

    this.res.end(this._body);
  },
};

class MiniKoa extends Emitter {
  constructor() {
    super();

    this.middleware = [];

    this.context = Object.create(context);
  }

  /**
    * 服务事件监听
    * @param {*} args
  */

  listen(...args) {
    const server = http.createServer(this.callback());

    return server.listen(...args);
  }

  /**
    * 注册使用中间件
    * @param {Function} fn

  */

  use(fn) {
    this.middleware.push(fn);
  }

  /**
    * 中间件总回调方法
    */

  callback() {
    if (this.listeners('error').length === 0) {
      this.on('error', this.onerror);
    }

    const handleRequest = (req, res) => {
      let context = this.createContext(req, res);
      let { middleware } = this;
      // 执行中间件
      compose(middleware)(context).catch(err => this.onerror(err));
    };

    return handleRequest;
  }

  /**
   * 异常处理监听
   * @param {EndOfStreamError} err
   */
  onerror(err) {
    console.log(err);
  }
  /**
    * 创建通用上下文
    * @param {Object} req
    * @param {Object} res
    */
  createContext(req, res) {
    let context = Object.create(this.context);
    context.req = req;
    context.res = res;
    return context;
  }
}

/**
 * 测试一下
 */
const app = new MiniKoa();
const PORT = 3001;
app.use(async ctx => {
  ctx.body = 'hello';
});
app.listen(PORT, () => {
  console.log(`started at port ${PORT}`);
});

基于 Koa 的微服务 Node.js 框架设计

然而，Koa只是一个HTTP框架，在实际的业务场景中，业务方除了要编写HTTP服务，还可能要编写其他类型的服务，比如 Thrift 服务、WebSocket 服务、消息队列的 Consumer 服务等等。应该如何设计这样一个不仅支持HTTP，还支持其他服务类型的微服务 Node.js 框架呢？我们先从这些服务的共性出发。

设计思想

HTTP、Thrift、WebSocket 等服务虽然应用层协议不同，但归根结底都是 C/S 结构的软件系统，其工作流程都可以划分为请求和响应两个阶段，如下图所示：

如果把整个客户端与服务端之间的交互过程看成是一个完整流水线的话，那么请求和响应自然就可以作为整个请求过程中的两个切面，因此 Koa 的洋葱模型也同样适用于除 HTTP 之外其他类型的服务。所以我们可以基于 Koa进行封装和改造，构造一个通用的服务中间件处理模型，这样我们就可以用 Koa 的形式来编写任意类型的服务程序。

框架的基本架构如下图所示：

image.png

简单实现

我们可以根据上述架构图做一个简单的实现（基于 AbstractServer 构建 HttpServer 与 ThriftServer ）：

某些方法的细节部分这里先不做展开，感兴趣的同学可以自行查阅更多资料。

AbstractServer

import compose from 'koa-compose';
import http from 'http';

export abstract class AbstractServer extends EventEmitter {
    public middlewares: any[];
    public context;
    public request;
    public response;

    /**
     * Initialize a new application.
     *
     * @constructor
     */
    constructor(options) {
        super();
        this.middlewares = [];
        this.context = Object.create(options.context);
        this.request = Object.create(options.request);
        this.response = Object.create(options.response);
    }
    /**
     * Listen to specific port.
     */
    public listen(...args) {
        const server = this.createServer(this.callback());
        return server.listen(...args);
    }

    /**
     * Use the given middleware `fn`.
     *
     * @param fn - middleware
     */
    public use(fn): this {
        if (typeof fn !== 'function') {
            throw new Error('middleware must be a function!');
        }
        this.middlewares.push(fn);
        return this;
    }

    /**
     * Return a request handler callback.
     */
    public callback() {
        const fn = compose(this.middlewares);
        return (req, res) => {
            const ctx = this.createContext(req, res);
            return this.handleRequest(ctx, fn);
        };
    }

    /**
     * Handle request in callback.
     *
     * @param ctx
     * @param fn
     */
    public handleRequest(ctx, fn): Promise<void> {
        return fn(ctx)
            .then(() => this.handleResponse(ctx))
            .catch((err) => ctx.onerror(err));
    }

    /**
     * Initialize a new context.
     *
     * @param {Object} req - request
     * @param {Object} res - response
     */
    public createContext(
        req,
        res,
    ) {
        const context = Object.create(this.context);
        const request = Object.create(this.request);
        const response = Object.create(this.response);
        context.app = this;
        context.request = request;
        context.response = response;
        context.req = req;
        context.res = res;
        context.state = {};

        request.app = this;
        request.ctx = context;
        request.req = req;
        request.res = res;
        request.response = response;

        response.app = this;
        response.ctx = context;
        response.req = req;
        response.res = res;
        response.request = request;
        return context;
    }

    /**
     * Default error handler
     *
     * @param err - error
     */
    public onerror(err: Error): void {
        const msg = err.stack || err.toString();
        console.error();
        console.error(msg.replace(/^/gm, '  '));
        console.error();
    }

    /**
     * Create server
     *
     * @param callback - server request callback
     */
    public abstract createServer(callback);
    /**
     * Handle response after all middlewares have been executed
     *
     * @param ctx - context
     */
    public abstract handleResponse(ctx): void;
}

HttpServer

export class HttpServer extends AbstractServer {
   /**
     * initialize http server
     * @param options
     */
  constructor(options) {
    // more detail...
    const { context, request, response } = options;
    super({
      context,
      request,
      response,
    });
  }
  /**
     * Handle request.
     *
     * @param ctx - context
     * @param fn - composed middleware
     */
  handleRequest(ctx, fn) {
    // more detail...

    return super.handleRequest(ctx, fn);
  }

  /**
     * Create context.
     *
     * @param req - raw request
     * @param res - raw response
     */
  createContext(req, res) {
    const context = super.createContext(req, res);
    // more detail...
    return context;
  }
  /**
     * Handle response after all middlewares have been executed.
     *
     * @param ctx
     */
  handleResponse(ctx) {
    let { body } = ctx;
    const { res } = ctx;
    const code = ctx.status;
    // more detail...
    body = JSON.stringify(body);
    return res.end(body);
  }

  /**
     * Error handler.
     *
     * @param err
     */

  onerror(err) {
    super.onerror(err);
  }

  /**
     * Create a http server.
     *
     * @param callback - request handler
     */
  createServer(callback, options?) {
    // more detail...

    return http.createServer(callback) as any;
  }
}