高并发如何保证微信 access_token 的有效

[swfbarhr]

本文源地址：高并发如何保证微信 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 的一些自己的看法。

[LevineChen]

搞个异步脚本过期前更新一个 应该是最简单可靠的方案吧

[swfbarhr]

@LevineChen 万一脚本挂了那就没得玩了

[odirus]

把 access_token 存放在 InnoDB 中，利用行锁的功能就好了。

[odirus]

以前我也喜欢玩 redis ，不过最近在维护三年前的项目，苦不堪言。所以现在做东西越简单越好。

[mooncakejs]

同一楼，另起一个进程定时更新比较好，挂了重启。

[orangemi]

呵呵，推荐一下： https://github.com/orangemi/process-locker
利用 redis 解决跨进城单一资源的异步锁，在请求一个唯一资源的时候，其它请求会被 pending 住，只有一个请求真正请求，其它的请求能够得到结果，结果会保存在 redis 中。

[orangemi]

这个还能解决用户使用第三方 OAuth 登录时，用户获取授权会得到一个 code ，往往这个 code 只有 1 次有效性， code 被试用后就会失效，用户因为各种原因（例如网络很慢）实际 code 已经发给服务方，但是用户会再次请求（这里也有可能是前端存在 bug 导致请求 2 次），服务端在第一时间拿到 code 去授权获取 access_token ，第二次 code 去获取授权时就会失败，前端就会出现登录失败。用跨进程异步资源锁也可以很好的解决这个问题。

[swfbarhr]

@mooncakejs 挂了重启的话，当前 pending 的请求都会出错，或者超时，我的想法是需要保证尽可能多的用户得到需要的结果

[swfbarhr]

@odirus 这样也是可以的

[ety001]

@swfbarhr 如果单独一个脚本定时请求更新的话，其实就不是一个高并发的问题了吧？

[dsphper]

@swfbarhr 万一系统挂了，那就没得玩了，万一世界毁灭了，那就没得玩了。我在想 1 分钟 cron 的一次的程序为啥要挂？

[swfbarhr]

@dsphper 人写的东西不能保证没有错，只是一个保险而已

[swfbarhr]

@all 我想说的是，我尽可能在能想到的情况来堵住每一个可能的情况，不是说其他方法就不行，要做就要考虑周全，这是我对软件的态度，不会强加到其他任何人身上，做好自己就行

[herozzm]

长文阅读体验真的不好

[reus]

我的做法是起一个进程定期刷新，然后存到数据库。其他进程要用，直接读数据库，不依赖更新 token 的进程。
这是一个服务进程，服务在设计的时候就应当考虑到随时可能挂掉，随时会重启。这不是一个脚本，这是一个服务。
如果这个进程无法服务，直接报警，开修。需要 token 的进程也不受影响，因为暂时还未过期，在过期前修好就是了。
并不会出现“挂了重启的话，当前 pending 的请求都会出错，或者超时”这种情况。每次都从数据库里拿的，不用上锁。

我不赞同微服务架构里分得太细的做法，例如用户一个服务，各类内容各一个服务，进程间通讯和同步带来过多复杂度，得不偿失。一个服务应当尽可能少和其他服务通讯，用中间服务来解耦。例如 token 更新服务不能工作，不影响使用 token 的服务，数据库也可以看作一个服务，这两个服务只依赖数据库，而不是直接依赖。数据库是比较稳定的，不会设计成可能频繁重启的。

[HunterPan]

搞个队列,多线程订阅队列,去取 token.没有并发问题啦

[pubby]

我们是专门写一个 token 服务，过期前提前更新。应用服务器各自来取，并自己缓存。

[csdreamdong]

处理好失效后的回滚，重新获取一次正确的 access_token

[500miles]

这不类似缓存雪崩问题么 = =.

[faceair]

@pubby +1
应用服务器自己的缓存可以设低一点，两三秒差不多了，内部 RPC 调用也很快

[marvinwilliam]

每 7000 秒刷新一次 access_token,为什么一定要等过期了才去刷新啊,不能提前刷新么?

[pubby]

@faceair 是的，应用服务器得到的超时要提前十秒。而 token 服务器会提前 1 分钟更新 token 。所以有 50 秒的时间窗口去更新 token

[goofansu]

起一个服务去定时刷新到 redis 不就行了

[swfbarhr]

@marvinwilliam 首先我认为定时服务去刷新 access_token 没有问题，我也承认这是最简单的解决方法。但是我要讨论的是边界性问题，也就是考虑到各种意外情况，在程序可控的范围内最大限度的去保证 API 的可用性。如你所说， 7000 秒刷新一次就 OK ，但是想想，谁又能保证刷新程序就一定能长远的运行呢？我这边的前提其实是如果我们刷新服务不可用的情况下，如何还能保证期间的请求可以正确的执行。但是如果 PM 或者用户能接受可能出现的一段时间的服务不可用，其实使用刷新服务就已经满足需求了。

[reus]

@swfbarhr 我们是 3600 秒，出现问题的话，有一个小时的时间可用于处理，处理故障期间，服务还是可用的， token 还未过期。服务都用 systemd 管理，退出了就自动重启，一直出错就报警通知技术人员。服务不可用这种情况，除非是微信方面的问题，否则不可能出现。就一个 GET 请求，实现复杂度基本为零，可能出问题的只是微信服务器那边。

文档里说的做法已经是最佳的了：
1 、为了保密 appsecrect ，第三方需要一个 access_token 获取和刷新的中控服务器。而其他业务逻辑服务器所使用的 access_token 均来自于该中控服务器，不应该各自去刷新，否则会造成 access_token 覆盖而影响业务；

根本就没有什么并发的事情，你这就是“各自去刷新”，所以才需要锁之类。
如果认为 3600 秒还不够，甚至可以减少到几百秒。 access token 每日限额是 2000 ，你可以算算间隔可以到多少。
这事其实很简单，不用搞得太复杂，用分布式锁之类的，根本就是降低可用性。

[dwood]

这么细心地题主写出来的程序一定是没有 bug 的。。。。

[swfbarhr]

@dwood 我也不是说我写的东西没有问题，我只是认为我们可以做的，代价也不是很大，那为什么不做一下呢？

[swfbarhr]

@reus 你说的没有错，可能是我考虑的太多了，我是假设刷新服务不可用的情况下。但是生产环境中可能会出现各种不可预期的问题，做好 2 手防备岂不是更好？

[xiaolongyuan]

@swfbarhr 万一 redis 挂了呢

[swfbarhr]

@xiaolongyuan 哈哈，那就真没得玩了

[magicdawn]

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


setnx lock
expire lock 2

1. redis 事务, 执行出错的话, 还是会继续执行
2. setnx exists-key value, 不会出错, 结果是 0
3. 导致一直 exipre 2s

个人愚见, 不对请指正!

[AlexaZhou]

微信推荐通过一个中央服务器来刷新 Token ， 也就是控制在一个地方刷新，避免了并发的问题

Ps: 我写了个 TokenBoy 专门用来解决这个问题, 直接拿去用就行 , 见 Github

[marvinwilliam]

@swfbarhr 那就让集群的节点刷新喽
缓存中保存一个对象,包含 access_token,过期时间,刷新锁
当有节点请求并判断快到过期阈值时,设置刷新锁为 true,其他节点正常访问,但是不能刷新,这个节点获取新的 access_token 后写回缓存,重置过期时间和刷新锁

[swfbarhr]

@magicdawn 貌似这边是有问题，折衷的方法只能是在 setnx 的回调里面去设置过期时间
有一点就是，如果 redis 事务抛处异常，那么事务不会继续执行下去（ redis 事务是保证原子性的），同时感谢你指出我的错误
function aquireLock(callback) {
redis.setnx('lock', function(err, result){
// 处理 err...

if(result > 0){
// 设置超时
redis.expire('lock', 2, function(err, result){
// 处理 err...

// 回调成功
});
}

// 设置未成功...
});
}

[magicdawn]

在应用层去判断 setnx 结果, 然后去 expire 我觉得没有问题, 不会那么巧执行了 setnx / 然后 expire 没吧

[magicdawn]

有 lua script

--[[
setnxAndExpire
]]

-- get args
local key, value, expire = KEYS[1], ARGV[1], ARGV[2]

-- sennx
local nxresult = redis.call('SETNX', key, value)

-- expire
if nxresult == 1 then
redis.call('EXPIRE', key, expire)
end

-- return
return nxresult


redis.defineCommand('setnx_and_expire', {
lua,
numberOfKeys: 1,
});

[magicdawn]

之前用过

let val = yield redis.incr key
if( val = 1) {
redis.expire key timeout
}

细想下来也没啥问题, 不会那么巧

[swfbarhr]

@magicdawn 放在 lua 里面肯定再好不过了，毕竟刻意直接在 redis 玩

[qiukun]

@reus amazon 倒是不让直接暴露数据库

[tairan2006]

一个分布式锁也能写这么多…而且文档里也有最佳方案啊，其实更简单的方法是在 redis 机器跑 crontab ，一个小时更新一次什么的，简单暴力。

[paicha]

想复杂了。

[z5864703]

把一个简单的问题复杂化，本来只要一个程序可用性高即可，现在又做这么多降低可用性的事。
楼主应该这样想，本来一个计算器就可以完成的事，你要用台超算来完成同样时间产生同样的结果，哪个可维护性与可靠性高？

[headin]

@tairan2006 觉得你的方案最好，能展开说一下吗？

[ryd994]

向中控获取，中控直接进队列，然后检查过期，过期就刷新，中控单机队列上锁很简单
或者异步定时更新
40 楼说的就是定时更新而已，只不过让系统服务来做定时触发。真要 crontab 挂了，这台服务器一般也没了

你担心脚本挂，应该考虑怎样做监控做报错做自动重启，不如考虑一下如果其他服务器发现过期时怎样触发中控强制更新，或者如何向运维报警，而不是重新发明分布式锁。

[ryd994]

而且还不是真正的分布式锁，因为 Redis 单点
crontab 单点和 Redis 单点，我宁可相信 crontab
最近在上分布式系统课，最近刚讲过分布式锁，还是在假设所有进程存活且没有通讯故障的前提下

[tairan2006]

@headin 还是建议用微信官方的方法… crontab 的思路的缺陷是官方可能修改 token 的过期时间，这样你无法确认更新周期； cron 可能跑挂； redis 集群也不能用这种方法更新。。

[yutian2211]

我不太懂 nodejs,不过楼主的线程模式下解决方案：
if (!isValid(access_token) && !flag) {
// 标记已经向微信服务器发送获取 access_token 的请求
flag = true;

这里没有并发问题么？

----------------------------------------------------------
2.之前很多的 V 友提了的:单脚本更新 token 的方法 更加的简单可靠，完全没有并发的问题，楼主强行上并发，然后思考各种解决方案，岂不是增加代码的复杂度与降低可维护性？

[zuotech]

定期刷新的有没有想过高并发的问题?
当刷新的那刻有很多高并发访问, 那此时访问的已经是过期的 token, 那这些访问将直接导致不可用...

[zuotech]

楼上的一些人说的, 一小时更新一次, 那一天就是更新 24 次, 那么在这 24 次更新中的高并发的脏读问题是没有解决的?
还有人说一个时更新一次, token 过期是 2 小时, 有一小时的处理时间, 但并未考虑到有些情况是, 更是请求更新成功了, 但是没有返回,或者是没有写入数据库中, 那程序将立即变的不可用啊...

[qianbaooffer]

@zuotech 可以在微信 token 过期之前,比如 token 过期时间 7200 秒,提前一段时间做定时更新

[iamcc]

定期刷新的有没有想过高并发的问题?
当刷新的那刻有很多高并发访问, 那此时访问的已经是过期的 token, 那这些访问将直接导致不可用...

-----


@zuotech 微信官方文档明确说明，再刷新过程中，两个新老 token 会共存。

[swfbarhr]

@iamcc 微信官方没有说明旧的 access_token 会在什么时候失效，根据我的测试，成功获取新的 access_token 后，前一个 access_token 会在 10 分钟左右的时间过期（如果按次数算，多次获取 access_token ，第一次获取的有效 access_token 差不多会在我们获取第 8-10 次的时候失效），所以如果我们提前 1 个小时刷新 access_token ，其实就是我们需要在 10 分钟左右处理完所有的事（如果出叉子的话，留给我们解决问题的时间其实不是 1 个小时）

[iamcc]

@swfbarhr 你所指的出岔子是指挂掉了重启刷新进程吗？

如果只是防止异常退出，那用 supervisor 之类的应该就没啥问题了。

再假如你真的对自己写的刷新进程的稳定性那么不信任，那就开多几个进程，分开不同的物理机，然后用一个分布式锁去实现直有单一进程成功执行的效果。

[c0ming]

这是典型的为了解决问题而引入了其他问题然后为了解决新问题而引入更多其他东西（逃