本文源地址：高并发如何保证微信 access_token 的有效，求 star

前言

消失了快 2 个月，俺又回来了。最近换比较忙，好久没写博客，但是学习的脚步一直没停下。前段时间在cnode上看到一个关于微信 access_token 的问题：高并发如何保证微信 token 的有效。其中本人也在上面回复了一下，但是觉得解决方案还是不够好，于是就有了本篇：本文主要以渐进的方式，来一步一步解决高并发情况下 access_token 的获取与保存。

前提条件

由于本文讨论是基于微信公众平台开发展开的，所以如果对微信公众平台开发不熟悉的同学可以先去看下微信公众平台的开发文档

需要解决的问题

本文讨论的其实是 access_token 获取与保存在高并发情况下的边界问题：

• 1.在 node 单进程情况下 第一个请求（请求 A ）过来，程序发现 access_token 过期，这时就会去向微信服务器获取新的 access_token ，然后更新到全局变量中。这个过程本身没有问题，但是如果请求 A 在向微信服务器请求新的 access_token 期间，来了第二个请求（请求 B ），因为这时新的 access_token 还未更新到全局变量中，请求 B 就会认为 access_token 已过期，也会同请求 A 一样，去微信服务器请求新的 access_token ，这样就会导致请求 A 得到的 access_token 失效。想象一下，如果在请求 A 后面又来了 N 个请求，并且此时请求 A 还未成功更新 access_token ，那么后面的所有请求都会去微信服务器获取新的 access_token 。以上的情况，不仅会导致浪费带宽，而且会导致最后一个请求之前获取的 access_token 都失效。那我们应该如何做控制，在高并发情况下仅请求一次微信服务器呢？

• 2.在 node 多进程模式下 如果我们的项目开启了 node 多进程，情况将更加复杂：我们不仅会遇到在单进程情况下的问题，而且由于 node 进程之间是不共享内存的，也就是说上面说到的全局变量是无法使用的，如果第一个请求由进程 1 处理，而此时 access_token 已过期，然后第二个请求由进程 2 处理。即使在进程 1 成功更新了 access_token ，但是由于进程 1 与进程 2 内存不共享，所以如果不借助外部存储的话，后面一个请求也无法得知 access_token 已更新。那么，这种情况有该如何解决呢？

（单进程模式下）思路

首先我们来讨论，如何解决单进程模式下高并发遇到的问题。

• 第一步我们需要引入一个新的全局变量，用来标记已经有请求在获取新的 access_token
• 第二步就是将后续的请求缓存到 node 的事件队列中去
• 第三步统一触发事件

具体实现代码：

var Emitter = require('events').Emitter;
var util = require('util');

var access_token, flag;

function TokenEmitter() {
  Emitter.call(this);
}
util.inherits(TokenEmitter, Emitter);

myEmitter = new TokenEmitter();
// 消除警告
myEmitter.setMaxListeners(0);

function getAccessToken(appID, appSecret, callback) {
  // 将 callback 缓存到事件队列中，等待触发
  myEmitter.once('token', callback);

  // 判断 access_token 是否过期
  if (!isValid(access_token) && !flag) {
    // 标记已经向微信服务器发送获取 access_token 的请求
    flag = true;

    // 向微信服务器请求新的 access_token
    requestForAccessToken(appID, appSecret, function(err, newToken) {
      if (err) {
        // 通知出错
        return myEmitter.emit('token', err);
      }

      // 更新 access_token
      access_token = newToken;
      // 触发所有监听回调函数
      myEmitter.emit('token', null, newToken.access_token);
      // 还原标记
      flag = false;
    });
  } else {
    process.nextTick(function(){
      callback(null, access_token.access_token);
    });
  }
}

以上代码主要的思路就是利用， node 自带的事件监听器，也就是代码中的'myEmitter.once()'方法，在 access_token 失效的情况下把所有调用的回调方法添加为'token'事件监听函数。并且只有第一个调用者可以去更新 access_token 的值（主要用 flag 来控制）。当获得新的 access_token 后，以新 access_token 为参数，去触发'token'事件。此时，所有监听了'token'事件的函数都会被调用，也就是说，所有调用者的回调函数都会被调用。这样，我们就实现了高并发情况下，防止 access_token 被多次更新的问题，也就是解决了问题 1 。

（多进程模式下）思路

解决了单进程模式下的问题，可以说我们多进程问题也解决了一部分。在多进程模式下，我们的主题思路还是与单进程一直，将调用缓存到事件队列中。但是，多进程的各个进程是不共享内存的，所以我们的 access_token 和 flag 标记不可以存储在变量中，因此需要引入外部存储： redis 。使用 redis 作为外部存储有以下几个原因：

• 1.在 redis 中统一存储 access_token ，各个进程都可以自由访问
• 2.利用 redis 作为锁媒介（相当于单进程中的 flag ）
• 3.利用 redis 发布订阅功能来触发事件（相当于单进程中的 emit ）

统一存储 access_token

这一点大家都应该没什么疑问， access_token 统一存储的好处就是不需要面对复杂的进程见通信。

锁媒介

当我们标记“正在请求微信服务器”的 flag 标志不可以放在代码的变量中时，那就要寻求代码之外的解决方法，其实我们可以存在 mongodb 、 mysql 等等可以存储的媒介中，甚至可以存放在文本文件中。但是为了保证速度，我还是考虑将其存放在速度更快的 redis 中。

redis 发布订阅功能

当然，如果我们的程序使用的是 node 的 cluster 模块开启的多进程模式，进程间通信还是相对容易一些：每个 worker 都可以向 master 发送 message ，利用这一点把 master 当做中心，来交换数据。但是如果我们是使用 pm2 开启了多实例， pm2 虽然提供了实例间通信的 API ，但是使用起来各种不顺畅，最终选择 redis 来作为各个实例接受通知的发起方。

以上思路的实现代码大致如下：

1.第一步需要做的就是判断 access_token 是否过期（为了方便起见，直接用 appID + appSecret 作为存储 access_token 的键）：从 redis 获取键为 appID + appSecret 的内容，因为我们在设置 access_token 时，是将其设为了过期键（设置过程涉及到锁，将在之后给出），所以只要能取到值，就说明 access_token 没有过期。代码如下：

function isValid(appID, appSecret, callback) {
  redis.get(appID + appSecret, function(err, token) {
    if (err) {
      return callback(err);
    }

    // 可以取到值
    if (tokenInfo) {
      return callback(null, token);
    }

    // 未取到值
    callback(null);
  });
}

2.如果在第一步的判断中，我们得出结论： access_token 已经过期，那么我们需要做的下一步就是设置一个代码级别的锁，防止之后的程序访问之后的代码：

function aquireLock(callback) {
  redis.setnx('lock', callback);
}

function releaseLock(callback) {
  redis.del('lock', callback);
}

这 2 个函数，一个用于设置锁，一个用于释放锁。我们设置锁是利用了 redis 的 setnx 命令原理： setnx 只可以设置不存在的 key ，即使同一时间有多个 setnx 命令来设置同一个 key ，最终只有一个客户端可以成功设置'lock'键，也就是说只有一个请求获得了锁的权限。这样就控制了并发产生的问题。

3.最后我们将所有程序写入主函数中：

• 1).主函数中一进来，首先添加监听器
• 2).第二步我们需要订阅 redis 的 new_token 和 new_token_err 频道
• 3).第三步判断 access_token 是否过期
• 4).如果过期，就尝试去获取锁权限
• 5).如果获取锁失败，就什么都不做，等待事件触发；如果获取锁权限成功后，就可以请求微信服务器来获取新的 access_token ，当获得新的 access_token 之后将其更新到 redis 中，并且设置合理的过期时间
• 6).释放锁并且发出通知

function getAccessToken(appID, appSecret, callback) {
  // 将 callback 缓存到事件队列中，等待触发
  myEmitter.once('token', callback);

  // 处理订阅消息
  subscribe.on('message', (channel, message) => {
    switch (channel) {
      case 'new_token':
        myEmitter.emit('token', null, message);
        break;
      case 'new_token_err':
        myEmitter.emit('token', new Error(message));
        break;
      default:
        break;
    }
  });

  // 判断 access_token 是否过期
  isValid(appID, appSecret, function(err, token) {
    // 出错
    if (err) {
      return myEmitter.emit('token', err);
    }

    // token 正常
    if (token) {
      return myEmitter.emit('token', null, token.access_token);
    }

    // token 已过期，获取锁
    aquireLock(function(err, result) {
      // 如果获取锁成功，则开始更新 access_token ，如果未得到锁，等待'token'触发
      if (result) {
        // 向微信服务器请求新的 access_token
        requestForAccessToken(appID, appSecret, function(err, newToken) {
          if (err) {
            // 释放锁标记
            releaseLock();
            // 通知出错
            return myEmitter.emit('token', err);
          }

          // 更新 access_token ，将新的 access_token 保存到 redis ，并且提前 5 分钟过期
          redis.setex(appID + appSecret, (newToken.expires_in - 300), newToken.access_token);
          // 发布更新
          publish.publish('new_token', newToken.access_token);
          // 释放锁标记
          releaseLock();
        });
      }
    });

    // 订阅
    subscribe.subscribe('new_token');
  });
}

进一步思考

到此，一个简单多进程控制 access_token 并发的解决方法已经呈现在眼前，但是我们还需要考虑一下边界情况：

• 1.当我们获得到锁权限的进程在未知因素下崩溃了，并且此时用于标记锁状态的'lock'已被成功设置
• 2.微信服务器崩溃了（无法正常提供服务） 以上 2 个情况，都会导致锁状态永远无法释放，针对这一类问题，我们可以引入超时的概念，在设置'lock'的时候给'lock'键设置一个过期时间，如果过了指定的时间还没有释放锁，那么锁权限就会被收回。代码如下：

function aquireLock(callback) {
  redis.watch('lock');
  redis.multi().setnx('lock').expire('lock', 2).exec(callback);
}

由于设置锁和设置锁的过期时间需要同一时间完成，所以这里我使用了 redis 的事务来保证了原子性。

更进一步的思考

虽然我们解决了锁问题，但是此时所有未获得锁的请求还处于 pending 状态，等待着 access_token 的到来，但是由于获得锁的请求已经走在天堂的路上，已经无法再来给其他这些个请求触发事件了。所以为了解决此类问题，我们需要引入另一个超时，那就是函数调用超时，在一定时间内未完成的话，我们就回调超时错误给调用者：

function getAccessToken(appID, appSecret, callback) {
  // 将 callback 缓存到事件队列中，等待触发
  myEmitter.once('token', callback);

  // 设置函数调用超时
  setTimeout(function () {
    callback(null, new Error('time out'));
  }, 2000);
  
  // ...
}

总结

其实在使用 redis 的订阅功能之前，我还考虑过tj的axon作为进程通信的手段，但是由于 axon 初始化过程有一定的延迟，不符合我的预期，所以放弃了。但是不得不说 axon 是一个非常好的项目，有条件的话可以用在项目当中。好了，以上就是我对高并发下处理 access_token 的一些自己的看法。