引言
大概兩個月前小夥伴問我有沒有基於redis實現過度布式鎖，以前看redis的時候知道有一個RedLock算法能夠實現分佈式鎖，我接觸的分佈式項目要麼是github上開源學習的，要麼是小夥伴們公司項目咱們一塊兒討論問題涉及的，我本身公司的項目中沒有實踐分佈式鎖的地方也就沒有仔細研究，向小夥伴推薦使用的是redisson實現的就是RedLock算法；固然有能力的還能夠本身根據redis做者的RedLock算法描述去實現node
插曲
關於RedLock算法的安全性有位大牛 Martin Kleppmann 產生了分歧 How to do distributed locking ；固然Redis做者 antirez 也作出了迴應 Is Redlock safe?；固然這是神仙"打架"，咱們從中學習大牛分析的問題，從而規避便可。git
淺析
加鎖
redisson經過lua腳原本實現加鎖和釋放鎖，使用lua腳本能夠保證原子性github
KEYS[1] 就是咱們本身定義的 鎖名
ARGV[2] 就是生成的鎖id UUID+線程id
ARGV[1] 就是生存時間
if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('hset', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
鎖對應的value+1 熟悉AQS鎖就會知道 這是鎖重入
"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
"redis.call('pexpire', KEYS[1], ARGV[1]); " +
"return nil; " +
"end; " +
"return redis.call('pttl', KEYS[1]);
假設 如今線程a，b來請求鎖，a先請求到，自定義鎖名叫作MY_TEST_LOCK；
b來請求鎖時，發現MY_TEST_LOCK 這個鎖能夠已經存在了，走第二個if；
若是不存在 將key 鎖id 超時時間 設置到redis中，返回null表示獲取到了鎖
第二if判斷這個鎖名+鎖id有沒有存在， 若是存在 說明是重入了 就把value加1
返回null 表示獲取到了鎖
若是不存在返回MY_TEST_LOCK 這個鎖的剩餘時間，代碼中b線程會while循環，
不停的嘗試加鎖
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
釋放鎖
釋放鎖
KEYS[1] 鎖名 例如：MY_TEST_LOCK
KEYS[2] 通道名 當釋放鎖時發現鎖不在redis中時使用
ARGV[1] 鎖id
ARGV[2] 鎖剩餘時間
ARGV[3] 鎖重入的值
若是鎖不存在 說明已經釋放過了 發佈redis消息
"if (redis.call('exists', KEYS[1]) == 0) then " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; " +
"end;" +
若是鎖對應得value 和redis中value不對應，說明該線程沒有持有鎖，不能釋放
"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
"return nil;" +
"end; " +
鎖對應的value -1 也就是釋放鎖
"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
若是鎖value仍是大於0 說明有重入狀況 不刪除
"if (counter > 0) then " +
"redis.call('pexpire', KEYS[1], ARGV[2]); " +
"return 0; " +
不然刪除 發佈redis消息
"else " +
"redis.call('del', KEYS[1]); " +
"redis.call('publish', KEYS[2], ARGV[1]); " +
"return 1; "+
"end; " +
"return nil;";
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
例子
這裏只是很淺的說一下怎麼用，而後解釋一下源碼裏是怎麼while循環獲取鎖的redis
哨兵模式
Config config = new Config();
config.useSentinelServers().addSentinelAddress(
"redis://172.29.3.245:26378","redis://172.29.3.245:26379", "redis://172.29.3.245:26380")
.setMasterName("mymaster")
.setPassword("a123456").setDatabase(0);算法
集羣模式
Config config = new Config();
config.useClusterServers().addNodeAddress(
"redis://172.29.3.245:6375","redis://172.29.3.245:6376", "redis://172.29.3.245:6377",
"redis://172.29.3.245:6378","redis://172.29.3.245:6379", "redis://172.29.3.245:6380")
.setPassword("a123456").setScanInterval(5000);安全
單redis模式
Config config = new Config();
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress("redis://127.0.0.1:6380")
.setTimeout(4000 * 10)
.setIdleConnectionTimeout(1000 * 60 * 10);分佈式
獲取鎖
public static final String MY_TEST_LOCK_NAME = "MY_TEST_LOCK";
RedissonClient redissonClient = Redisson.create(config);
RLock lock = redissonClient.getLock(USER_LOCK_NAME);
boolean getLock = false;
try {
getLock = lock.tryLock(10, 5, TimeUnit.SECONDS);
if (getLock){
獲取到鎖後執行代碼
System.out.println(Thread.currentThread().getName()+"線程 鎖住");
}
} catch (InterruptedException e) {
//todo 處理異常
e.printStackTrace();
} finally {
lock.unlock();
}ide
單redis版 獲取代碼
/**
* waitTime 獲取能夠等待的時間
* leaseTime 過了這個時間以後 redis這個鎖自動消失
*/
@Override
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
long time = unit.toMillis(waitTime);
long current = System.currentTimeMillis();
final long threadId = Thread.currentThread().getId();
先獲取一下鎖
Long ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
若是獲取到了鎖 返回值就是null
if (ttl == null) {
return true;
}
time -= (System.currentTimeMillis() - current);
if (time <= 0) {
time <= 0表示超時了
acquireFailed(threadId);
return false;
}
current = System.currentTimeMillis();
final RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {
if (!subscribeFuture.cancel(false)) {
subscribeFuture.addListener(new FutureListener<RedissonLockEntry>() {
@Override
public void operationComplete(Future<RedissonLockEntry> future) throws Exception {
if (subscribeFuture.isSuccess()) {
unsubscribe(subscribeFuture, threadId);
}
}
});
}
acquireFailed(threadId);
return false;
}性能
try {
time -= (System.currentTimeMillis() - current);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
循環獲取鎖
while (true) {
long currentTime = System.currentTimeMillis();
ttl = tryAcquire(leaseTime, unit, threadId);
// lock acquired
if (ttl == null) {
return true;
}學習
time -= (System.currentTimeMillis() - currentTime);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
// waiting for message
currentTime = System.currentTimeMillis();
if (ttl >= 0 && ttl < time) {
getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
} else {
getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
}
time -= (System.currentTimeMillis() - currentTime);
if (time <= 0) {
acquireFailed(threadId);
return false;
}
}
} finally {
unsubscribe(subscribeFuture, threadId);
}
// return get(tryLockAsync(waitTime, leaseTime, unit));
}
多節點版獲取鎖 RedissonMultiLock
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
// try {
// return tryLockAsync(waitTime, leaseTime, unit).get();
// } catch (ExecutionException e) {
// throw new IllegalStateException(e);
// }
long newLeaseTime = -1;
if (leaseTime != -1) {
newLeaseTime = unit.toMillis(waitTime)*2;
}
long time = System.currentTimeMillis();
long remainTime = -1;
if (waitTime != -1) {
remainTime = unit.toMillis(waitTime);
}
long lockWaitTime = calcLockWaitTime(remainTime);
須要多少個redis 加鎖成功 限制(N/2 + 1)
int failedLocksLimit = failedLocksLimit();
加鎖成功集合
List<RLock> acquiredLocks = new ArrayList<RLock>(locks.size());
for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {
RLock lock = iterator.next();
boolean lockAcquired;
try {
if (waitTime == -1 && leaseTime == -1) {
lockAcquired = lock.tryLock();
} else {
long awaitTime = Math.min(lockWaitTime, remainTime);
lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);
}
} catch (RedisResponseTimeoutException e) {
unlockInner(Arrays.asList(lock));
lockAcquired = false;
} catch (Exception e) {
lockAcquired = false;
}
加鎖成功 加入到成功集合
if (lockAcquired) {
acquiredLocks.add(lock);
} else {
失敗判斷成功節點是否達到了要求
if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {
break;
}
if (failedLocksLimit == 0) {
unlockInner(acquiredLocks);
if (waitTime == -1 && leaseTime == -1) {
return false;
}
failedLocksLimit = failedLocksLimit();
acquiredLocks.clear();
// reset iterator
while (iterator.hasPrevious()) {
iterator.previous();
}
} else {
failedLocksLimit--;
}
}
if (remainTime != -1) {
remainTime -= (System.currentTimeMillis() - time);
time = System.currentTimeMillis();
if (remainTime <= 0) {
unlockInner(acquiredLocks);
return false;
}
}
}
if (leaseTime != -1) { List<RFuture<Boolean>> futures = new ArrayList<RFuture<Boolean>>(acquiredLocks.size()); for (RLock rLock : acquiredLocks) { RFuture<Boolean> future = rLock.expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS); futures.add(future); } for (RFuture<Boolean> rFuture : futures) { rFuture.syncUninterruptibly(http://www.my516.com); } } return true; }123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211zookeeper實現分佈式鎖在於小夥伴討論Redis實現分佈式鎖的同時，咱們在萬能的github上發現了另外一種zookeeper實現分佈式鎖的方式zookeeper只是聽過，沒有用過，這裏簡單說下區別：redis 分佈式鎖，須要本身不斷去嘗試獲取鎖，比較消耗性能，可是效率高zk 分佈式鎖，獲取不到鎖，註冊個監聽器便可，不須要不斷主動嘗試獲取鎖，性能開銷較小，可是健壯性強另一點就是，若是是 redis 獲取鎖的那個客戶端 出現 bug 掛了，那麼只能等待超時時間以後才能釋放鎖；而 zk 的話，由於建立的是臨時 znode，只要客戶端掛了，znode 就沒了，此時就自動釋放鎖---------------------