從零搭建 Node.js 企業級 Web 服務器（十二）：遠程調用

關於遠程調用

遠程調用（簡稱 RPC）主要是指服務器或集羣之間對處理過程的調用。經過遠程調用能夠打通不一樣系統之間的數據與功能，或是抽離與創建公共的邏輯統一提供服務。遠程調用的兩端也稱爲遠程調用的客戶端與服務端，通常是多對多的關係，須要引入註冊與發現機制進行治理，下圖爲最多見實踐：

治理機制一般是因地制宜的，能夠基於 ZooKeeper 建設，本章再也不展開。

gRPC

gPRC 是谷歌開源的一款跨語言高性能的 RPC 框架，底層使用 protobuf 進行數據交換，已在谷歌、奈非、思科等企業大規模應用。如下爲 gRPC 的調用過程，客戶端先進行 DNS 解析，再根據解析結果直連服務端或向負載均衡/治理服務獲取服務端信息再鏈接：

本章將基於上一章已完成的工程 host1-tech/nodejs-server-examples - 11-schedule 加入遠程調用功能實現一個簡單消息的輸入與輸出，如今到工程根目錄安裝 grpc 相關模塊：

$ yarn add @grpc/grpc-js @grpc/proto-loader # 本地安裝 @grpc/grpc-js、@grpc/proto-loader
# ...
info Direct dependencies
├─ @grpc/grpc-js@1.1.3
└─ @grpc/proto-loader@0.5.5
# ...

加上遠程調用

經過 .proto 文件定義遠程接口並寫入基於該定義的 gRPC 客戶端與服務端：

$ mkdir src/rpc       # 新建 src/rpc 存放遠程調用邏輯

$ mkdir src/rpc/echo  # 新建 src/rpc/echo

$ tree src -L 1       # 展現 src 目錄內容結構
src
├── config
├── controllers
├── middlewares
├── models
├── moulds
├── rpc
├── schedules
├── server.js
├── services
└── utils

// src/rpc/echo/def.proto
syntax = "proto3";

service Echo {
  rpc Get(EchoRequest) returns (EchoResponse) {}
}

message EchoRequest { string message = 1; }

message EchoResponse { string message = 1; }

// src/rpc/echo/client.js
const { resolve } = require('path');
const { promisify } = require('util');
const protoLoader = require('@grpc/proto-loader');
const grpc = require('@grpc/grpc-js');
const { rpc } = require('../../config');

class EchoClient {
  grpcClient;

  async init() {
    const grpcObject = grpc.loadPackageDefinition(
      await protoLoader.load(resolve(__dirname, 'def.proto'))
    );

    this.grpcClient = new grpcObject.Echo(
      `${rpc.domain}:${rpc.port}`,
      grpc.credentials.createInsecure()
    );
  }

  get = async ({ s, logger }) => {
    const { grpcClient } = this;

    const { message } = await promisify(
      grpcClient.get.bind(grpcClient, { message: s })
    )();

    logger.info('Echo/Get Invoked');

    return { message };
  };
}

let client;
module.exports = async () => {
  if (!client) {
    client = new EchoClient();
    await client.init();
  }
  return client;
};

// src/rpc/echo/server.js
const { resolve } = require('path');
const { callbackify } = require('util');
const protoLoader = require('@grpc/proto-loader');
const grpc = require('@grpc/grpc-js');

class EchoServer {
  grpcServer;

  async init() {
    const grpcObject = grpc.loadPackageDefinition(
      await protoLoader.load(resolve(__dirname, 'def.proto'))
    );

    this.grpcServer.addService(grpcObject.Echo.service, this);
  }

  get = callbackify(async (call) => {
    const { message } = call.request;
    return { message };
  });
}

let server;
module.exports = async (grpcServer) => {
  if (!server) {
    server = new EchoServer();
    Object.assign(server, { grpcServer });
    await server.init();
  }
  return server;
};

// src/config/index.js
// ...
const config = {
  // 默認配置
  default: {
    // ...
+    
+    rpc: {
+      domain: 'localhost',
+      port: process.env.PORT_RPC || 9001,
+    },
  },
  // ...
};
// ...

開啓 gRPC 日誌輸出並初始化：

# .env
LOG_LEVEL='debug'
+
+GRPC_TRACE='all'
+GRPC_VERBOSITY='DEBUG'

// src/utils/logger.js
// ...
+const GRPC_LOGGER_REGEXP = /^.+Z\s+\|\s+/;
+
+function grpcLogger(logger, level = 'debug') {
+  const verbosities = ['debug', 'info', 'error'];
+
+  return {
+    error(severity, message) {
+      if (typeof severity != 'number') {
+        message = severity;
+        severity = 0;
+      }
+
+      if (typeof message != 'string') {
+        message = String(message || '');
+      }
+
+      logger[verbosities[severity] || level](
+        message.replace(GRPC_LOGGER_REGEXP, '')
+      );
+    },
+  };
+}
+
module.exports = logger;

-Object.assign(module.exports, { logging });
+Object.assign(module.exports, { logging, grpcLogger });

// src/rpc/index.js
const { promisify } = require('util');
const grpc = require('@grpc/grpc-js');
const { rpc } = require('../config');
const logger = require('../utils/logger');
const echoClient = require('./echo/client');
const echoServer = require('./echo/server');
const { grpcLogger } = logger;

module.exports = async function initRpc() {
  grpc.setLogger(grpcLogger(logger.child({ type: 'rpc' }), 'debug'));

  // init rpc servers
  const grpcServer = new grpc.Server();
  await echoServer(grpcServer);

  await promisify(grpcServer.bindAsync.bind(grpcServer))(
    `0.0.0.0:${rpc.port}`,
    grpc.ServerCredentials.createInsecure()
  );
  grpcServer.start();

  // init rpc clients
  await echoClient();
};

// src/server.js
const express = require('express');
const { resolve } = require('path');
const { promisify } = require('util');
const initMiddlewares = require('./middlewares');
const initControllers = require('./controllers');
const initSchedules = require('./schedules');
+const initRpc = require('./rpc');
const logger = require('./utils/logger');

const server = express();
const port = parseInt(process.env.PORT || '9000');
const publicDir = resolve('public');
const mouldsDir = resolve('src/moulds');

async function bootstrap() {
+  await initRpc();
  server.use(await initMiddlewares());
  server.use(express.static(publicDir));
  server.use('/moulds', express.static(mouldsDir));
  server.use(await initControllers());
  server.use(errorHandler);
  await initSchedules();
  await promisify(server.listen.bind(server, port))();
  logger.info(`> Started on port ${port}`);
}
// ...

添加 gRPC 客戶端 logger 與控制層入口：

// src/middlewares/trace.js
const { v4: uuid } = require('uuid');
const morgan = require('morgan');
const onFinished = require('on-finished');
const logger = require('../utils/logger');
const { logging } = logger;

module.exports = function traceMiddleware() {
  return [
    morgan('common', { skip: () => true }),
    (req, res, next) => {
      req.uuid = uuid();
      req.logger = logger.child({ uuid: req.uuid });
      req.loggerSql = req.logger.child({ type: 'sql' });
      req.logging = logging(req.loggerSql, 'info');
+      req.loggerRpc = req.logger.child({ type: 'rpc' });

      onFinished(res, () => {
        // ...
      });

      next();
    },
  ];
};

// src/controllers/echo.js
const { Router } = require('express');
const cc = require('../utils/cc');
const rpcEchoClient = require('../rpc/echo/client');

class EchoController {
  rpcEchoClient;

  async init() {
    this.rpcEchoClient = await rpcEchoClient();

    const router = Router();
    router.get('/', this.get);
    return router;
  }

  get = cc(async (req, res) => {
    const { s = '' } = req.query;
    const message = await this.rpcEchoClient.get({ s, logger: req.loggerRpc });
    res.send({ success: true, message });
  });
}

module.exports = async () => {
  const c = new EchoController();
  return await c.init();
};

// src/controllers/index.js
const { Router } = require('express');
const shopController = require('./shop');
const chaosController = require('./chaos');
const healthController = require('./health');
const loginController = require('./login');
const csrfController = require('./csrf');
+const echoController = require('./echo');

module.exports = async function initControllers() {
  const router = Router();
  router.use('/api/shop', await shopController());
  router.use('/api/chaos', await chaosController());
  router.use('/api/health', await healthController());
  router.use('/api/login', await loginController());
  router.use('/api/csrf', await csrfController());
+  router.use('/api/echo', await echoController());
  return router;
};

訪問 http://localhost:9000/api/echo?s=Hello%20RPC 便可看到效果：

同時在命令行可以看到充分的 gRPC 日誌：

# ...
08:20:52.320Z DEBUG 12-rpc: dns_resolver | Resolver constructed for target dns:0.0.0.0:9001 (type=rpc)
08:20:52.321Z DEBUG 12-rpc: dns_resolver | Resolution update requested for target dns:0.0.0.0:9001 (type=rpc)
08:20:52.321Z DEBUG 12-rpc: dns_resolver | Returning IP address for target dns:0.0.0.0:9001 (type=rpc)
08:20:52.322Z DEBUG 12-rpc: server | Attempting to bind 0.0.0.0:9001 (type=rpc)
08:20:52.324Z DEBUG 12-rpc: server | Successfully bound 0.0.0.0:9001 (type=rpc)
08:20:52.327Z DEBUG 12-rpc: resolving_load_balancer | dns:localhost:9001 IDLE -> IDLE (type=rpc)
08:20:52.327Z DEBUG 12-rpc: connectivity_state | dns:localhost:9001 IDLE -> IDLE (type=rpc)
08:20:52.327Z DEBUG 12-rpc: dns_resolver | Resolver constructed for target dns:localhost:9001 (type=rpc)
# ...

本章源碼

host1-tech/nodejs-server-examples - 12-rpc

更多閱讀

從零搭建 Node.js 企業級 Web 服務器（零）：靜態服務
從零搭建 Node.js 企業級 Web 服務器（一）：接口與分層
從零搭建 Node.js 企業級 Web 服務器（二）：校驗
從零搭建 Node.js 企業級 Web 服務器（三）：中間件
從零搭建 Node.js 企業級 Web 服務器（四）：異常處理
從零搭建 Node.js 企業級 Web 服務器（五）：數據庫訪問
從零搭建 Node.js 企業級 Web 服務器（六）：會話
從零搭建 Node.js 企業級 Web 服務器（七）：認證登陸
從零搭建 Node.js 企業級 Web 服務器（八）：網絡安全
從零搭建 Node.js 企業級 Web 服務器（九）：配置項
從零搭建 Node.js 企業級 Web 服務器（十）：日誌
從零搭建 Node.js 企業級 Web 服務器（十一）：定時任務從零搭建 Node.js 企業級 Web 服務器（十二）：遠程調用