^-^
后端技术_手写分布式锁
最近更新：2024-09-23   |   字数总计：2.7k   |   阅读估时：13分钟   |   阅读量：次
  1. 进程中的锁
  2. 进程间通信（IPC）
    1. 信号量
    2. 条件变量
  3. 分布式锁
    1. mysql实现分布式非公平锁
      1. 表结构
      2. 加锁
      3. 解锁
      4. 超进程解锁
      5. 其他实现细节
    2. redis实现分布式非公平锁
      1. 锁存储
      2. 加锁
      3. 解锁
      4. 可用性
      5. 实现细节
    3. etcd实现分布式公平锁
    4. 总结

进程中的锁

  1. 互斥锁
  2. 自旋锁
  3. 读写锁
  4. 信号量
  5. 条件变量
  6. 原子变量和内存屏障

进程间通信（IPC）

  1. pipe
  2. FIFO
  3. 信号量
  4. 信号
  5. 消息队列
  6. 共享内存
  7. socket

信号量

1
2
3
sem_init(&sem, pshared, 10) // 10为资源个数
P操作：sem_wait(&sem) // sem-1, 信号量大于0，立即返回；为0则阻塞
V操作：sem_post(&sem) // sem+1

条件变量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
// thread1
pthread_mutex_lock(&mtx);
    // ！！使用while循环才能防止虚假唤醒，不能改成if
while(condition) 
    pthread_cond_wait(&cond, &mtx);
pthread_mutex_unlock(&mtx);

// thread2
pthread_mutex_lock(&mtx);
 // ...修改条件
pthread_mutex_unlock(&mtx);
pthread_cond_signal(&cond); 
// or 
pthread_cond_broadcast(&cond);
  1. 条件变量为什么与互斥锁协同？
    • condition是临界资源（竞态条件），所以需要配合mutex使用；
  2. 哪些锁加锁和解锁对象必须是同一个线程？
    • 互斥锁和自旋锁，其他的可以由其他线程解锁；
  3. 产生虚假唤醒的原因：
    1. 信号被打断，如：read == -1，errno == EINTR时，在有些系统中可能会被唤醒
    2. 信号劫持，也会唤醒
    3. 多处理器时的bug，当pthread_cond_signal发送的时候有可能唤醒多个

分布式锁

  • 锁类型：互斥锁、自旋锁（同一时刻只允许一个执行体进入临界资源，执行体意味着可能是多线程也可能是多节点）
  • 分布式：一个机器上的多进程 + 多机器 + 多网段，也就是说节点采用socket通信
  • 锁存储位置：可以共同访问的节点
  • 分布式锁特性：
    1. 互斥性：加锁解锁通过网络交互，每个锁都有打标记，即执行体的唯一标志
    2. 锁超时：加锁解锁需要同一个执行体，但是网络异常的情况下需要实现锁超时（由锁存储的节点实现）
    3. 续锁功能：假设锁即将超时了，但是节点还没有计算完业务，此时锁节点将会自动释放锁，该节点需要实现发一个包进行续锁的功能
    4. 可用性：假设锁存储位置宕机，会导致严重错误，所以需要多个备份点（计算型 + 存储型）
    5. 容错性：存储型的备份点需要实现主从切换和一致性算法（raft一致性算法实现数据同步 + redlock算法）
  • 分布式锁类型：
    1. 重入锁（允许一个对象多次重复持有锁，使用init的时候设置PTHREAD_RECURSIVE_MUTEX_INITIALIZER_NP）和非重入锁（默认）
    2. 公平锁（也就是互斥锁，要排队）和非公平锁(也就是自旋锁，空转轮询) ；
      1. 互斥锁流程：先在用户态自旋一会，让出执行权进入阻塞队列，切换上下文，锁释放后，进入就绪队列，等待cpu调度再进入运行态；
      2. 自旋锁流程：先在用户态自旋一会，也会让出执行权，但是进入就绪队列，完成线程切换，随时cpu调度再进入运行态；
      3. 公平锁在分布式场景下：请求响应式模型，s1s4发出获取锁请求，只有s1收到锁成功回答，s2s4没有回应且进入排队，等s2可以获取时，锁节点主动回包给s2（也可能是通过广播直接通知s2~s4）
      4. 非公平锁在分布式场景下：s1s4发出获取锁请求，s1获得成功回应，s2s4获得失败回应，s2~s4需要设置定时器不断去继续锁请求

mysql实现分布式非公平锁

表结构

1
2
3
4
5
6
7
8
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='分布式锁表';

加锁

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

解锁

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

超进程解锁

1
DELETE FROM dislock WHERE 'lock_type' = 'act_lock'; # 若是超时释放，不需要owner_id

其他实现细节

  • 锁存储：表
  • 使用unique key及primary key实现了互斥特性（也可以用innodb中的s锁和x锁）
  • 锁超时：因为mysql没有定时器删除行的特性，所以启动另一个超进程做定时器，update_time超时删除记录
  • mysql没有主动通知特性，所以其他线程需要实现定时器去主动探寻锁情况
  • mysql可用性有很大缺陷，计算和存储是一体的，不适合做多点，单点挂掉将导致业务系统失效
  • mysql实现分布式锁并不常用，除非系统没有redis和etcd

redis实现分布式非公平锁

锁存储

  • 内存数据库

加锁

1
set lock uuin nx ex 30 # nx 是只有lock不存在的时候才能成功，ex是以秒为单位超时删除，px是以毫秒为单位

解锁

1
2
1. 需要通过lua脚本 先确认是否是持有锁对象
2. del lock

可用性

  • redlock算法
    1. 准备奇数个redis节点
    2. 加锁的时候同时给多个节点发送消息，只要有半数以上成功就算成功
    3. 解锁也是只要有半数以上就算成功

实现细节

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
// redlock.h
#ifndef __redlock__
#define __redlock__

#include <iostream>
#include <vector>
#include "hiredis/hiredis.h"
extern "C" {
#include "sds.h"
}

using namespace std;

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;   // 续锁脚本
};

#endif /* defined(__redlock__) */
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
// redlock.cpp
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/types.h>
#include <unistd.h>
#include <math.h>
#include "redlock.h"

/* Turn the plain C strings into Sds strings */
static char **convertToSds(int count, char** args) {
    int j;
    char **sds = (char**)malloc(sizeof(char*)*count);
    for(j = 0; j < count; j++)
        sds[j] = sdsnew(args[j]);
    return sds;
}

CLock::CLock() : m_validityTime(0), m_resource(NULL), m_val(NULL) {
}

CLock::~CLock() {
    sdsfree(m_resource);
    sdsfree(m_val);
}

int CRedLock::m_defaultRetryCount = 3;
int CRedLock::m_defaultRetryDelay = 200;
float CRedLock::m_clockDriftFactor = 0.01;

// ----------------
// init redlock
// ----------------
CRedLock::CRedLock() {
    Initialize();
}

// ----------------
// release redlock
// ----------------
CRedLock::~CRedLock() {
    sdsfree(m_continueLockScript);
    sdsfree(m_unlockScript);
    close(m_fd);
    /* Disconnects and frees the context */
    for (int i = 0; i < (int)m_redisServer.size(); i++) {
        redisFree(m_redisServer[i]);
    }
}

// ----------------
// Initialize the server: scripts...
// ----------------
bool CRedLock::Initialize() {
    m_continueLockScript = sdsnew("if redis.call('get', KEYS[1]) == ARGV[1] then redis.call('del', KEYS[1]) end return redis.call('set', KEYS[1], ARGV[2], 'px', ARGV[3], 'nx')");
    m_unlockScript       = sdsnew("if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end");
    m_retryCount = m_defaultRetryCount;
    m_retryDelay = m_defaultRetryDelay;
    m_quoRum     = 0;
    // open rand file
    m_fd = open("/dev/urandom", O_RDONLY);
    if (m_fd == -1) {
        //Open error handling
        printf("Can't open file /dev/urandom\n");
        exit(-1);
        return false;
    }
    return true;
}

// ----------------
// add redis server
// ----------------
bool CRedLock::AddServerUrl(const char *ip, const int port) {
    redisContext *c = NULL;
    struct timeval timeout = { 1, 500000 }; // 1.5 seconds
    c = redisConnectWithTimeout(ip, port, timeout);	
    if (c) {
        m_redisServer.push_back(c);
    }
    m_quoRum = (int)m_redisServer.size() / 2 + 1;
    return true;
}

// ----------------
// set retry
// ----------------
void CRedLock::SetRetry(const int count, const int delay) {
    m_retryCount = count;
    m_retryDelay = delay;
}

// ----------------
// lock the resource
// ----------------
bool CRedLock::Lock(const char *resource, const int ttl, CLock &lock) {
    sds val = GetUniqueLockId();
    if (!val) {
        return false;
    }
    lock.m_resource = sdsnew(resource);
    lock.m_val = val;
    printf("Get the unique id is %s\n", val);
    int retryCount = m_retryCount;
    do {
        int n = 0;
        int startTime = (int)time(NULL) * 1000;
        int slen = (int)m_redisServer.size();
        for (int i = 0; i < slen; i++) {
            if (LockInstance(m_redisServer[i], resource, val, ttl)) {
                n++;
            }
        }
        //Add 2 milliseconds to the drift to account for Redis expires
        //precision, which is 1 millisecond, plus 1 millisecond min drift
        //for small TTLs.
        int drift = (ttl * m_clockDriftFactor) + 2;
        int validityTime = ttl - ((int)time(NULL) * 1000 - startTime) - drift;
        printf("The resource validty time is %d, n is %d, quo is %d\n",
               validityTime, n, m_quoRum);
        if (n >= m_quoRum && validityTime > 0) {
            lock.m_validityTime = validityTime;
            return true;
        } else {
            Unlock(lock);
        }
        // Wait a random delay before to retry
        int delay = rand() % m_retryDelay + floor(m_retryDelay / 2);
        usleep(delay * 1000);
        retryCount--;
    } while (retryCount > 0); // 非公平锁，不断检测
    return false;
}

// ----------------
// release resource
// ----------------
bool CRedLock::ContinueLock(const char *resource, const int ttl, CLock &lock) {
    sds val = GetUniqueLockId();
    if (!val) {
        return false;
    }
    lock.m_resource = sdsnew(resource);
    lock.m_val = val;
    if (m_continueLock.m_resource == NULL) {
        m_continueLock.m_resource = sdsnew(resource);
        m_continueLock.m_val = sdsnew(val);
    }
    printf("Get the unique id is %s\n", val);
    int retryCount = m_retryCount;
    do {
        int n = 0;
        int startTime = (int)time(NULL) * 1000;
        int slen = (int)m_redisServer.size();
        for (int i = 0; i < slen; i++) {
            if (ContinueLockInstance(m_redisServer[i], resource, val, ttl)) {
                n++;
            }
        }
        // update old val
        sdsfree(m_continueLock.m_val);
        m_continueLock.m_val = sdsnew(val);
        //Add 2 milliseconds to the drift to account for Redis expires
        //precision, which is 1 millisecond, plus 1 millisecond min drift
        //for small TTLs.
        int drift = (ttl * m_clockDriftFactor) + 2;
        int validityTime = ttl - ((int)time(NULL) * 1000 - startTime) - drift;
        printf("The resource validty time is %d, n is %d, quo is %d\n",
               validityTime, n, m_quoRum);
        if (n >= m_quoRum && validityTime > 0) {
            lock.m_validityTime = validityTime;
            return true;
        } else {
            Unlock(lock);
        }
        // Wait a random delay before to retry
        int delay = rand() % m_retryDelay + floor(m_retryDelay / 2);
        usleep(delay * 1000);
        retryCount--;
    } while (retryCount > 0);
    return false;
}

// ----------------
// lock the resource
// ----------------
bool CRedLock::Unlock(const CLock &lock) {
    int slen = (int)m_redisServer.size();
    for (int i = 0; i < slen; i++) {
        UnlockInstance(m_redisServer[i], lock.m_resource, lock.m_val);
    }
    return true;
}

// ----------------
// lock the resource milliseconds
// ----------------
bool CRedLock::LockInstance(redisContext *c, const char *resource,
                            const char *val, const int ttl) {
    redisReply *reply;
    reply = (redisReply *)redisCommand(c, "set %s %s px %d nx",
                                       resource, val, ttl); // px 实现锁超时，nx实现互斥
    if (reply) {
        printf("Set return: %s [null == fail, OK == success]\n", reply->str);
    }
    if (reply && reply->str && strcmp(reply->str, "OK") == 0) {
        freeReplyObject(reply);
        return true;
    }
    if (reply) {
        freeReplyObject(reply);
    }
    return false;
}

// ----------------
// 对redismaster续锁, 私有函数
// ----------------
bool CRedLock::ContinueLockInstance(redisContext *c, const char *resource,
                                    const char *val, const int ttl) {
    int argc = 7;
    sds sdsTTL = sdsempty();
    sdsTTL = sdscatprintf(sdsTTL, "%d", ttl);
    char *continueLockScriptArgv[] = {(char*)"EVAL",
                                      m_continueLockScript,
                                      (char*)"1",
                                      (char*)resource,
                                      m_continueLock.m_val,
                                      (char*)val,
                                      sdsTTL};
    redisReply *reply = RedisCommandArgv(c, argc, continueLockScriptArgv);
    sdsfree(sdsTTL);
    if (reply) {
        printf("Set return: %s [null == fail, OK == success]\n", reply->str);
    }
    if (reply && reply->str && strcmp(reply->str, "OK") == 0) {
        freeReplyObject(reply);
        return true;
    }
    if (reply) {
        freeReplyObject(reply);
    }
    return false;
}

// ----------------
// 对redismaster解锁
// ----------------
void CRedLock::UnlockInstance(redisContext *c, const char *resource,
                              const char *val) {
    int argc = 5;
    char *unlockScriptArgv[] = {(char*)"EVAL",
                                m_unlockScript,
                                (char*)"1",
                                (char*)resource,
                                (char*)val};
    redisReply *reply = RedisCommandArgv(c, argc, unlockScriptArgv);
    if (reply) {
        freeReplyObject(reply);
    }
}

// ----------------
// 对redismaster执行脚本命令
// ----------------
redisReply * CRedLock::RedisCommandArgv(redisContext *c, int argc, char **inargv) {
    char **argv;
    argv = convertToSds(argc, inargv);
    /* Setup argument length */
    size_t *argvlen;
    argvlen = (size_t *)malloc(argc * sizeof(size_t));
    for (int j = 0; j < argc; j++)
        argvlen[j] = sdslen(argv[j]);
    redisReply *reply = NULL;
    reply = (redisReply *)redisCommandArgv(c, argc, (const char **)argv, argvlen);
    if (reply) {
        printf("RedisCommandArgv return: %lld\n", reply->integer);
    }
    free(argvlen);
    sdsfreesplitres(argv, argc);
    return reply;
}

// ----------------
// 获取唯一id
// ----------------
sds CRedLock::GetUniqueLockId() {
    unsigned char buffer[20];
    if (read(m_fd, buffer, sizeof(buffer)) == sizeof(buffer)) {
        //获取20byte的随机数据
        sds s;
        s = sdsempty();
        for (int i = 0; i < 20; i++) {
            s = sdscatprintf(s, "%02X", buffer[i]);
        }
        return s;
    } else {
        //读取失败
        printf("Error: GetUniqueLockId %d\n", __LINE__);
    }
    return NULL;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
// LockExample.cpp
#include <stdlib.h>
#include <unistd.h>
#include "redlock-cpp/redlock.h"

int main (int argc, char **argv) {
    CRedLock * dlm = new CRedLock();
    dlm->AddServerUrl("127.0.0.1", 6379);
    // dlm->AddServerUrl("127.0.0.1", 5006);
    // dlm->AddServerUrl("127.0.0.1", 5007);
    // 分布式锁的使用案例
    while (1) {
        CLock my_lock;
        bool flag = dlm->Lock("foo", 100000, 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(8);
            dlm->Unlock(my_lock);
            // do other job
            sleep(2);
        } else {
            printf("获取失败, lock not acquired, name:%s\n", my_lock.m_val);
            sleep(rand() % 3);
        }
    }
    return 0;
}

etcd实现分布式公平锁

总结

  1. 完备性：etcd>redis>mysql
  2. 性能：redis>etcd>mysql
  3. 选择：都有选择etcd，如果只有mysql不会为了实现分布式锁加入redis或etcd
2024-03-05 该篇文章被 Cleofwine 归为分类: 服务端

上一篇：后端技术_异步日志方案log4cpp 下一篇：后端技术_手写内存泄漏检测组件（施工）
© 2023-2026 Cleofwine / Email / GitHub / 浙ICP备2024136127号
🌙赠我弯弯一枚月，也赠予我晚星⭐ 本站由Hexo驱动｜使用Hexo-theme-A4主题