分布式锁本质上是解决分布式事务中的隔离性（某个行为可能需要访问多个目标，为了避免并发情况的发生，需要保证执行操作的顺序性）。

二、解决的问题

可以理解为把多线程中解决的问题的场景迁移到分布式场景中。某个资源或行为同时只允许一个实体执行，其他实体需要等待。

分布式锁由两部分构成：
（1）资源。锁存储。锁本质上是资源；即这是一个所有人都能访问到的资源。在分布式场景中，锁资源是通过网络交互来访问的，实体访问锁资源，锁资源要告诉实体是否访问成功。
（2）行为。加锁或解锁；网络交互的方式实现。隐含重点：加锁对象和解锁对象必须为同一个；借鉴mutex的理念。

三、分布式锁特性

（1）互斥性。分布式锁是互斥类型的锁；同时只允许一个持锁对象进入临界资源，其他待持锁对象要么等待，要么轮询检测是否能获取锁。需要记录持有锁对象（加锁对象和解锁对象必须为同一个）方便判定锁被谁占有了。
（2）锁超时。允许持锁对象持锁最长时间；如果持锁对象宕机，需要外力解除锁定，方便其他持锁对象获取锁。
在单进程的多线程场景下，资源和行为是同生共死的关系，程序宕机会自动释放所有资源和行为。而在分布式场景中有比较大的差别，锁资源和行为是分离的，通过网络交互操作锁，要考虑到锁资源宕机和行为实体宕机的情况如何释放资源和解除行为。
比如行为实体宕机了，如何释放锁？如果不能释放锁则其他的实体将一直等待；所以，需要锁超时机制，设置操作时长的最大值，超时释放锁。再比如锁资源宕机的情况。
（3）高可用性。合理实时间内得到合理的回复；琐存储位置若宕机，可能引发整个系统不可用；应有备份存储位置和切换备份存储的机制，从而确保服务可用。
实现上有两种方式：计算型（不存储数据只完成行为，比如网关只计算不存储）和存储型（存储数据，有多个备份点，存储超过半数以上的备份节点；具有切换功能，以解决宕机问题）。
（4）容错性。若锁存储位置宕机，恰好锁丢失的话，是否能正确处理。
一个实体行为申请了锁，此时锁资源宕机，切换到了备份资源，但是备份资源没有该实体行为的记录，这是种错误，那么就要一致性来解决（raft一致性算法和redlock来实现）。
redlock的实现：开奇数个进程，写锁的时候，写入进程半数以上成功的返回获取锁成功，否则失败。

四、分布式锁类型

（1）重入锁和非重入锁。是否允许持锁对象再次获取锁。
对应一个进程多线环境的递归锁和非递归锁；即允不允许持锁对象多次进入临界资源。
（2）公平锁和非公平锁。对应互斥锁和自旋锁。公平锁通常通过排队来实现；非公平锁通常不间断尝试获取锁来实现。
互斥锁中如果一个线程发现锁以及被占用了，那么会进入阻塞等待，会引起线程切换，线程进入阻塞队列；当锁释放了，阻塞队列会取出一个阻塞线程加入就绪队列，等待CPU调度。

自旋锁中如果一个线程发现锁以及被占用了，那么会空转CPU，如果一定时间锁还不可用，那么会调用sched_yield()，引起线程切换，线程直接进入就绪队列，等待CPU调度。自旋锁不需要考虑内存序和编译器优化，循环中不断使用原子操作检查锁是否可用。

DROP TABLE IF EXISTS `dislock`;
CREATE TABLE `dislock` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT
COMMENT '主键',
  `lock_type` varchar(64) NOT NULL COMMENT '锁类型',
  `owner_id` varchar(255) NOT NULL COMMENT '持锁对象',
  `update_time` timestamp NOT NULL DEFAULT
CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `idx_lock_type` (`lock_type`) ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT
CHARSET=utf8 COMMENT='分布式锁表';

INSERT INTO dislock (`lock_type`, `owner_id`) VALUES ('act_lock', 'ad2daf3');

// 连续分布式锁的使用案例
while (1) {
    CLock my_lock;
    bool flag = dlm->ContinueLock("foo", 14000, my_lock);
    if (flag) {
        printf("获取成功, Acquired by client name:%s, res:%s, vttl:%d\n",
               my_lock.m_val, my_lock.m_resource, my_lock.m_validityTime);
        // do resource job
        sleep(10);
    } else {
        printf("获取失败, lock not acquired, name:%s\n", my_lock.m_val);
        sleep(rand() % 3);
    }
}

DELETE FROM dislock WHERE `lock_type` = 'act_lock' AND `owner_id` = 'ad2daf3';

# set key value NX
set act_lock 123123123 NX

--[[
    KEYS[1]         lock_name
    KEYS[2]         lock_channel_name
    ARGV[1]         lock_time (ms)
    ARGV[2]         uuid
]]
if redis.call('exists', KEYS[1]) == 0 then
    redis.call('hset', KEYS[1], ARGV[2], 1)
    redis.call('pexpire', KEYS[1], ARGV[1])
    return
end
-- 若支持锁重入，将注释去掉
-- if redis.call('hexists', KEYS[1], ARGV[2]) ==
1 then
--     redis.call('hincrby', KEYS[1], ARGV[2], 1)
--     redis.call('pexpire', KEYS[1], ARGV[1])
--     return
-- end
redis.call("subscribe", KEYS[2])
return redis.call('pttl', KEYS[1])

del act_lock

get act_lock

--[[
    KEYS[1]         lock_name
    KEYS[2]         lock_channel_name
    ARGV[1]         0 sign of unlock
    ARGV[2]         lock_expire_time
    ARGV[3]         uuid
]]
if redis.call('exists', KEYS[1]) == 0 then
    redis.call('publish', KEYS[2], ARGV[1])
    return 1
end
if redis.call('hexists', KEYS[1], ARGV[3]) == 0
then
    return
end
-- 若支持锁重入，将注释去掉
-- local cnt = redis.call('hincrby', KEYS[1],
ARGV[3], -1)
-- if cnt > 0 then
--     redis.call('pexpire', KEYS[1], ARGV[2])
--     return 0
-- else
    redis.call('del', KEYS[1])
    redis.call('publish', KEYS[2], ARGV[1])
    return 1
-- end

set act_lock 123123123 NX EX 10
# 或者
set act_lock 123123123 NX PX 10000

ttl act_lock

class CLock {
public:
                            CLock();
                            ~CLock();
public:
    int                     m_validityTime; // 当前锁可以存活的时间, 毫秒
    sds                     m_resource;     // 要锁住的资源名称
    sds                     m_val;          // 锁住资源的进程随机名字
};

class CRedLock {
public:
                            CRedLock();
    virtual                 ~CRedLock();
public:
    bool                    Initialize();
    bool                    AddServerUrl(const char *ip, const int port);
    void                    SetRetry(const int count, const int delay);
    bool                    Lock(const char *resource, const int ttl, CLock &lock);
    bool                    ContinueLock(const char *resource, const int ttl,
                                         CLock &lock);
    bool                    Unlock(const CLock &lock);
private:
    bool                    LockInstance(redisContext *c, const char *resource,
                                         const char *val, const int ttl);
    bool                    ContinueLockInstance(redisContext *c, const char *resource,
                                                 const char *val, const int ttl);
    void                    UnlockInstance(redisContext *c, const char *resource,
                                           const char *val);
    sds                     GetUniqueLockId();
    redisReply *            RedisCommandArgv(redisContext *c, int argc, char **inargv);
private:
    static int              m_defaultRetryCount;    // 默认尝试次数3
    static int              m_defaultRetryDelay;    // 默认尝试延时200毫秒
    static float            m_clockDriftFactor;     // 电脑时钟误差0.01
private:
    sds                     m_unlockScript;         // 解锁脚本
    int                     m_retryCount;           // try count
    int                     m_retryDelay;           // try delay
    int                     m_quoRum;               // majority nums
    int                     m_fd;                   // rand file fd
    vector<redisContext *>  m_redisServer;          // redis master servers
    CLock                   m_continueLock;         // 续锁
    sds                     m_continueLockScript;   // 续锁脚本
};

八、分布式锁的使用场景

（1）有些业务场景，需要严格的执行顺序，如果顺序打破可能会引起业务数据错乱。