Node.js 开发实践总结-定时脚本的设计与实现

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

前言

作为Node语言的初学者去实践后端开发时，不仅仅有见猎心喜，也有一些忐忑，好在大家都很open，给予了很多建议和分享，到目前为止，也成功建立了三个基于Node.js + TypeScript + IMServer 1 的工程，也是时候将自己最近的学习过程进行总结，下面就以一个小小的开发任务为载体分享下我的成长过程。

需求

在完成Node工程的搭建之后，我接受到第一个Node后台开发任务：定时将企业微信的组织架构信息拉取到业务数据库系统中，并且提供手机号查询用户查询接口。一开始对这个任务还是比较乐观的，信心满满的去开发了。

初步方案

经过方案设计之后形成了上述的方案：

在服务器部署初始化时（init.ts初始启动文件中）启动node-schedule的定时任务，读取数据库中的企业微信的企业配置，然后并行启动若干企业的组织架构更新进程。
企业微信提供了获取部门成员的详情，因此只需并行更新每个部门的信息，并且写入mysql数据库中。

3. 当查询接口到达服务器后，首先从数据库中查询该手机号对应的成员，若不存在则从企业微信侧调用手机号获取userid API，然后通过获取用户信息API获取最新的用户信息，避免定时更新带来的更新时间gap；若存在则直接返回数据库中的信息。

开发过程中的踩雷

整体业务逻辑并不复杂，调试和部署的过程中遇到许多问题，这里给大家一一列举下：

访问频率受限企业微信官方规定同一时间对同一份资源的请求数不可超过一定数值（60），由于部门详情的请求接口采用的并行模式，因此超过了阈值，测试过程中被官方封禁了IP。
过多进程导致SQL慢查询没有考虑多地部署（3地 * 5服务器 * 8 worker）导致同时存在了120个更新进程，进而导致数据库mysql的读写混乱，也消耗了大量性能，导致数据库读写压力比较大时，出现了部分慢查询的情况
无效手机号不可调用企业微信api 企业微信对手机号获取userid的接口，具有以下限制：当查询中出现一定数量的无效手机号时，会触发企业微信官方IP封禁。但是业务系统中存在大量离职后的无效手机号，因此当检查到数据库中不存在时，频繁调用上述接口则会触发封禁。
数据库读写冲突由于存在多台服务器同时读写数据库，导致数据库出现了部分重复、缺少的情况。
网络环境导致读写锁的平衡性失效，产生衍生问题为了优化上述部分，引入的任务读写锁，保证单一进程更新。但是未考虑各地服务网络情况，导致内网服务器一直持有读写锁，失去了均衡负载的设计有效性。也导致在配置预上线环境时，预上线环境由于网络环境良好一直持有读写锁，进而影响线上的实时数据。
未考虑失败情况进行报警和恢复

深度优化设计

下面介绍下如何解决这些问题和思路和方案。

1、访问频率受限

这里针对“部门成员信息API“的并行请求，改造成基于有效频率值的串行发送机制，设计成10个/每秒的调用速度。

2、过多进程导致SQL慢查询

这个解决方案比较明确，就是减少启动定时任务的进程数。

由于后端服务一般分为测试环境、预上线环境、正式环境，不同的环境中是否需要启动各个定时器脚本可以通过部署时（以SKTE为例）,设置环境变量“SCHEDULE_ENV”来管理。

每台服务器会启动8个worker进程，每个worker使用“process.env.IMSERVER_WORKER_ID”变量进行标识，因此可以设计只有“worker1”进程来进行定时任务的启动；

3、无效手机号不可调用企业微信api

这个是在技术调研中没能发现的情况，发现前期技术调研的工作疏忽。

首先是业务调用方是无法得知手机号是否有效，且也不应该去关心这个限制，因此原先为了解决部分新纪录更新不及时的问题，而引入的实时查询机制是不合理的。

实时查询机制：“对于数据库中不存在的手机号，通过企业微信官方api进行实时查询来返回结果”

因此移除了这个机制，并且提供了一个基于企业微信官方API的实时查询接口，每次业务方调用时，也将结果同步更新到组织架构中。

4、数据库读写冲突

引入redis任务锁机制，保证同一时间内只有一个进程能够进行数据库更新操作。

其次是企业之间的更新采用并行机制，由于相互之间是互不冲突的，因此不会引起同一条记录的读写冲突，也可以提升其更新速度。

5、网络环境导致读写锁的平衡性失效，产生衍生问题

在最初的设计中，我希望服务器之间能够根据自身的负载情况来进行公平竞争任务锁，但是实际情况是由于多地部署，其中稳定的内网环境可以一直优先获取到任务锁，就是没有所谓的公平性了。

特别是当压测需要部署预上线环境时，如果没有设置只读db账号并且没有设置启动定时任务环境变量，这两个失误会导致某一次的组织架构更新逻辑调整的代码更新到线上时，线上一直是旧的逻辑在执行，经过一系列排查我们发现预上线环境一直获取了读写锁，使用旧的逻辑更新数据库。

因此增加环境变量来控制定时任务启动、对于压测的环境的中的数据库权限进行了区分，增加了只读模式。

6、报警和错误恢复

这里有一点前端思维定势的影响了，这一部分是同样重要的。

报警方面

则是接入IMLog的Node SDK，通过 Kibana 和 Grafana 的系统配置，可以有效监控组织架构的更新情况。

错误恢复方面

这里的错误主要是发生在企业微信API的access_token过期的情况，常发生于以下两种情况：

企业微信官方主动使access_token过期
在组织架构更新过程中，access_token刚好失效情况，也就是http传输到企业微信刚好失效的情况

以上的情况是无法避免的。这里使用中间件对node.fetch进行封装，增加对response的返回值的校验，如果企业微信api的返回值是 “WX_CODE.INVALIDE_TOKEN” 则进行预警和重置accessToken。

export default (app) => {
  const { utils: { imlogHelper } } = app;
  const wrapperLogFetch = (originFetch, {
    traceId,
    header,
    client_ip,
  }) => async (...args) => {
    const res = await originFetch(...args);
    if (res.errcode === WX_CODE.INVALIDE_TOKEN) {
      // 进行更新逻辑
      wxService.clearAllRedisKey();
      imlogHelper({
        cmd: url,
        message: 'accessToken_update_warning',
        body: JSON.stringify(res),
        trace_id: traceId,
        retcode,
        headers: header,
      });
    }
    return res;
  };
    // 覆盖context.fetch方法
  return async (ctx, next) => {
    if (!ctx.logFetch) {
      const originFetch = ctx.fetch;
      const { traceId, ip: client_ip } = ctx.request;
      const header = JSON.stringify(ctx.request.header);
      const logFetch = wrapperLogFetch(originFetch, {
        traceId,
        header,
        client_ip,
      });
      ctx.logFetch = logFetch;
    }
    if (ctx.fetch !== ctx.logFetch) {
      ctx.fetch = ctx.logFetch;
    }
    await next();
  };
}