Node.js 中实践基于 Redis 的分布式锁实现

时间  2019-11-29
标签 node.js node 实践 基于 redis 分布式 实现 栏目 Node.js 繁體版
原文   原文链接

在一些分布式环境下、多线程并发编程中，若是对同一资源进行读写操做，避免不了的一个就是资源竞争问题，经过引入分布式锁这一律念，能够解决数据一致性问题。html

做者简介：五月君，Nodejs Developer，慕课网认证做者，热爱技术、喜欢分享的 90 后青年，欢迎关注 Nodejs技术栈 和 Github 开源项目 www.nodejs.rednode

认识线程、进程、分布式锁

线程锁：单线程编程模式下请求是顺序的，一个好处是不须要考虑线程安全、资源竞争问题，所以当你进行 Node.js 编程时，也不会去考虑线程安全问题。那么多线程编程模式下，例如 Java 你可能很熟悉一个词 synchronized，一般也是 Java 中解决并发编程最简单的一种方式，synchronized 能够保证在同一时刻仅有一个线程去执行某个方法或某块代码。git

进程锁：一个服务部署于一台服务器，同时开启多个进程，Node.js 编程中为了利用操做系统资源，根据 CPU 的核心数能够开启多进程模式，这个时候若是对一个共享资源操做仍是会遇到资源竞争问题，另外每个进程都是相互独立的，拥有本身独立的内存空间。关于进程锁经过 Java 中的 synchronized 也很难去解决，synchronized 仅局限于在同一个 JVM 中有效。github

分布式锁：一个服务不管是单线程仍是多进程模式，当多机部署、处于分布式环境下对同一共享资源进行操做仍是会面临一样的问题。此时就要去引入一个概念分布式锁。以下图所示，因为先读数据在经过业务逻辑修改以后进行 SET 操做，这并非一个原子操做，当多个客户端对同一资源进行先读后写操做就会引起并发问题，这时就要引入分布式锁去解决，一般也是一个很普遍的解决方案。redis

图片描述

基于 Redis 的分布式锁实现思路

实现分布式锁的方式有不少：数据库、Redis、Zookeeper。这里主要介绍的是经过 Redis 来实现一个分布式锁，至少要保证三个特性：安全性、死锁、容错。算法

安全性：所谓一个萝卜一个坑，第一点要作的是上锁，在任意时刻要保证仅有一个客户端持有该锁。数据库

死锁：形成死锁多是因为某种缘由，本该释放的锁没有被释放，所以在上锁的时候能够同步的设置过时时间，若是因为客户端本身的缘由没有被释放，也要保证锁可以自动释放。npm

容错：容错是在多节点的模式下须要考虑的，只要能保证 N/2+1 节点可用，客户端就能够成功获取、释放锁。编程

Redis 单实例分布式锁实现

在 Redis 的单节点实例下实现一个简单的分布式锁，这里会借助一些简单的 Lua 脚原本实现原子性，不了解能够参考以前的文章 Node.js 中实践 Redis Lua 脚本安全

上锁

上锁的第一步就是先经过 setnx 命令占坑，为了防止死锁，一般在占坑以后还会设置一个过时时间 expire，以下所示：

setnx key value
expire key seconds
复制代码

以上命令不是一个原子性操做，所谓原子性操做是指命令在执行过程当中并不会被其它的线程或者请求打断，以上若是 setnx 执行成功以后，出现网络闪断 expire 命令便不会获得执行，会致使死锁出现。

也许你会想到使用事物来解决，可是事物有个特色，要么成功要么失败，都是一口气执行完成的，在咱们上面的例子中，expire 是须要先根据 setnx 的结果来判断是否须要进行设置，显然事物在这里是行不通的，社区也有不少库来解决这个问题，如今 Redis 官方 2.8 版本以后支持 set 命令传入 setnx、expire 扩展参数，这样就能够一条命令一口气执行，避免了上面的问题，以下所示：

  • value：建议设置为一个随机值，在释放锁的时候会进一步讲解
  • EX seconds：设置的过时时间
  • PX milliseconds：也是设置过时时间，单位不同
  • NX|XX：NX 同 setnx 效果是同样的
set key value [EX seconds] [PX milliseconds] [NX|XX]
复制代码

释放锁

释放锁的过程就是将本来占有的坑给删除掉，可是也并不能仅仅使用 del key 删除掉就万事大吉了，这样很容易删除掉别人的锁，为何呢？举一个例子客户端 A 获取到一把 key = name1 的锁（2 秒中），紧接着处理本身的业务逻辑，可是在业务逻辑处理这块阻塞了耗时超过了锁的时间，锁是会自动被释放的，这期间该资源又被客户端 B 获取了 key = name1 的锁，那么客户端 A 在本身的业务处理结束以后直接使用 del key 命令删除会把客户端 B 的锁给释放掉了，因此释放锁的时候要作到仅释放本身占有的锁。

加锁的过程当中建议把 value 设置为一个随机值，主要是为了更安全的释放锁，在 del key 以前先判断这个 key 存在且 value 等于本身指定的值才执行删除操做。判断和删除不是一个原子性的操做，此处仍需借助 Lua 脚本实现。

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end
复制代码

Redis 单实例分布式锁 Node.js 实践

使用 Node.js 的 Redis 客户端为 ioredis，npm install ioredis -S 先安装该包。

初始化自定义 RedisLock

class RedisLock {
    /** * 初始化 RedisLock * @param {*} client * @param {*} options */
    constructor (client, options={}) {
        if (!client) {
            throw new Error('client 不存在');
        }

        if (client.status !== 'connecting') {
            throw new Error('client 未正常连接');
        }

        this.lockLeaseTime = options.lockLeaseTime || 2; // 默认锁过时时间 2 秒
        this.lockTimeout = options.lockTimeout || 5; // 默认锁超时时间 5 秒
        this.expiryMode = options.expiryMode || 'EX';
        this.setMode = options.setMode || 'NX';
        this.client = client;
    }
}
复制代码

上锁

经过 set 命令传入 setnx、expire 扩展参数开始上锁占坑，上锁成功返回，上锁失败进行重试，在 lockTimeout 指定时间内仍未获取到锁，则获取锁失败。

class RedisLock {
    
    /** * 上锁 * @param {*} key * @param {*} val * @param {*} expire */
    async lock(key, val, expire) {
        const start = Date.now();
        const self = this;

        return (async function intranetLock() {
            try {
                const result = await self.client.set(key, val, self.expiryMode, expire || self.lockLeaseTime, self.setMode);
        
                // 上锁成功
                if (result === 'OK') {
                    console.log(`${key} ${val} 上锁成功`);
                    return true;
                }

                // 锁超时
                if (Math.floor((Date.now() - start) / 1000) > self.lockTimeout) {
                    console.log(`${key} ${val} 上锁重试超时结束`);
                    return false;
                }

                // 循环等待重试
                console.log(`${key} ${val} 等待重试`);
                await sleep(3000);
                console.log(`${key} ${val} 开始重试`);

                return intranetLock();
            } catch(err) {
                throw new Error(err);
            }
        })();
    }
}
复制代码

释放锁

释放锁经过 redis.eval(script) 执行咱们定义的 redis lua 脚本。

class RedisLock {
    /** * 释放锁 * @param {*} key * @param {*} val */
    async unLock(key, val) {
        const self = this;
        const script = "if redis.call('get',KEYS[1]) == ARGV[1] then" +
        " return redis.call('del',KEYS[1]) " +
        "else" +
        " return 0 " +
        "end";

        try {
            const result = await self.client.eval(script, 1, key, val);

            if (result === 1) {
                return true;
            }
            
            return false;
        } catch(err) {
            throw new Error(err);
        }
    }
}
复制代码

测试

这里使用了 uuid 来生成惟一 ID，这个随机数 id 只要保证惟一无论用哪一种方式均可。

const Redis = require("ioredis");
const redis = new Redis(6379, "127.0.0.1");
const uuidv1 = require('uuid/v1');
const redisLock = new RedisLock(redis);

function sleep(time) {
    return new Promise((resolve) => {
        setTimeout(function() {
            resolve();
        }, time || 1000);
    });
}

async function test(key) {
    try {
        const id = uuidv1();
        await redisLock.lock(key, id, 20);
        await sleep(3000);
        
        const unLock = await redisLock.unLock(key, id);
        console.log('unLock: ', key, id, unLock);
    } catch (err) {
        console.log('上锁失败', err);
    }  
}

test('name1');
test('name1');
复制代码

同时调用了两次 test 方法进行上锁，只有第一个是成功的，第二个 name1 26e02970-0532-11ea-b978-2160dffafa30 上锁的时候发现 key = name1 已被占坑，开始重试，因为以上测试中设置了 3 秒钟以后自动释放锁，name1 26e02970-0532-11ea-b978-2160dffafa30 在通过两次重试以后上锁成功。

name1 26e00260-0532-11ea-b978-2160dffafa30 上锁成功
name1 26e02970-0532-11ea-b978-2160dffafa30 等待重试
name1 26e02970-0532-11ea-b978-2160dffafa30 开始重试
name1 26e02970-0532-11ea-b978-2160dffafa30 等待重试
unLock:  name1 26e00260-0532-11ea-b978-2160dffafa30 true
name1 26e02970-0532-11ea-b978-2160dffafa30 开始重试
name1 26e02970-0532-11ea-b978-2160dffafa30 上锁成功
unLock:  name1 26e02970-0532-11ea-b978-2160dffafa30 true
复制代码

源码地址

https://github.com/Q-Angelo/project-training/tree/master/redis/lock/redislock.js
复制代码

Redlock 算法

以上是使用 Node.js 对 Redis 分布式锁的一个简单实现，在单实例中是可用的，当咱们对 Redis 节点作一个扩展，在 Sentinel、Redis Cluster 下会怎么样呢？

如下是一个 Redis Sentinel 的故障自动转移示例图，假设咱们客户端 A 在主节点 192.168.6.128 获取到锁以后，主节点还将来得及同步信息到从节点就挂掉了，这时候 Sentinel 会选举另一个从节点作为主节点，那么客户端 B 此时也来申请相同的锁，就会出现一样一把锁被多个客户端持有，对数据的最终一致性有很高的要求仍是不行的。

图片描述

Redlock 介绍

鉴于这些问题，Redis 官网 redis.io/topics/dist… 提供了一个使用 Redis 实现分布式锁的规范算法 Redlock，中文翻译版参考 redis.cn/topics/dist…

Redlock 在上述文档也有描述，这里简单作个总结：Redlock 在 Redis 单实例或多实例中提供了强有力的保障，自己具有容错能力，它会从 N 个实例使用相同的 key、随机值尝试 set key value [EX seconds] [PX milliseconds] [NX|XX] 命令去获取锁，在有效时间内至少 N/2+1 个 Redis 实例取到锁，此时就认为取锁成功，不然取锁失败，失败状况下客户端应该在全部的 Redis 实例上进行解锁。

Node.js 中应用 Redlock

github.com/mike-marcac… 是 Node.js 版的 Redlock 实现，使用起来也很简单，开始以前先安装 ioredis、redlock 包。

npm i ioredis -S
npm i redlock -S
复制代码

编码

const Redis = require("ioredis");
const client1 = new Redis(6379, "127.0.0.1");
const Redlock = require('redlock');
const redlock = new Redlock([client1], {
    retryDelay: 200, // time in ms
    retryCount: 5,
});

// 多个 Redis 实例
// const redlock = new Redlock(
// [new Redis(6379, "127.0.0.1"), new Redis(6379, "127.0.0.2"), new Redis(6379, "127.0.0.3")],
// )

async function test(key, ttl, client) {
    try {
        const lock = await redlock.lock(key, ttl);

        console.log(client, lock.value);
        // do something ...

        // return lock.unlock();
    } catch(err) {
        console.error(client, err);
    }
}

test('name1', 10000, 'client1');
test('name1', 10000, 'client2');
复制代码

测试

对同一个 key name1 两次上锁，因为 client1 先取到了锁，client2 没法获取锁，重试 5 次以后报错 LockError: Exceeded 5 attempts to lock the resource "name1".

图片描述
相关文章
相关标签/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公众号
   欢迎关注本站公众号,获取更多信息
联系我们 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程