分布式鎖的解決方式
- 基于數據庫表做樂觀鎖甘磨，用于分布式鎖橡羞。(適用于小并發(fā))
- 使用memcached的add()方法，用于分布式鎖济舆。
- 使用memcached的cas()方法卿泽，用于分布式鎖。(不常用)
- 使用redis的setnx()滋觉、expire()方法签夭，用于分布式鎖。
- 使用redis的setnx()椎侠、get()第租、getset()方法，用于分布式鎖肺蔚。
- 使用redis的watch煌妈、multi儡羔、exec命令宣羊，用于分布式鎖。(不常用)
- 使用zookeeper汰蜘，用于分布式鎖仇冯。(不常用)
這里主要介紹第四種和第五種：
使用redis的setnx()、expire()方法族操，用于分布式鎖
原理
對于使用redis的setnx()苛坚、expire()來實現分布式鎖，這個方案相對于memcached()的add()方案色难，redis占優(yōu)勢的是泼舱，其支持的數據類型更多，而memcached只支持String一種數據類型枷莉。除此之外娇昙，無論是從性能上來說，還是操作方便性來說笤妙，其實都沒有太多的差異冒掌，完全看你的選擇噪裕，比如公司中用哪個比較多，你就可以用哪個股毫。
首先說明一下setnx()命令膳音，setnx的含義就是SET if Not Exists，其主要有兩個參數 setnx(key, value)铃诬。該方法是原子的祭陷，如果key不存在，則設置當前key成功趣席，返回1颗胡；如果當前key已經存在，則設置當前key失敗吩坝，返回0毒姨。但是要注意的是setnx命令不能設置key的超時時間，只能通過expire()來對key設置钉寝。
具體的使用步驟如下:
- setnx(lockkey, 1) 如果返回0弧呐，則說明占位失敗嵌纲；如果返回1俘枫，則說明占位成功
- expire()命令對lockkey設置超時時間，為的是避免死鎖問題逮走。
- 執(zhí)行完業(yè)務代碼后鸠蚪，可以通過delete命令刪除key。
為了保證在某個Redis節(jié)點不可用的時候算法能夠繼續(xù)運行师溅，這個獲取鎖的操作還有一個超時時間(timeOut)茅信，它要遠小于鎖的有效時間（幾十毫秒量級）。
可能存在的問題
這個方案其實是可以解決日常工作中的需求的墓臭，但從技術方案的探討上來說蘸鲸，可能還有一些可以完善的地方。比如窿锉，如果在第一步setnx執(zhí)行成功后酌摇，在expire()命令執(zhí)行成功前，發(fā)生了宕機的現象嗡载，那么就依然會出現死鎖的問題窑多，所以如果要對其進行完善的話，可以使用redis的setnx()洼滚、get()和getset()方法來實現分布式鎖埂息。
具體實現
鎖具體實現RedisLock：
package com.xiaolyuh.lock;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
public class RedisLock {
private static Logger logger = LoggerFactory.getLogger(RedisLock.class);
//////////////////// 靜態(tài)常量定義開始///////////////////////
/**
* 存儲到redis中的鎖標志
*/
private static final String LOCKED = "LOCKED";
/**
* 默認請求鎖的超時時間(ms 毫秒)
*/
private static final long TIME_OUT = 100;
/**
* 默認鎖的有效時間(s)
*/
public static final int EXPIRE = 60;
//////////////////// 靜態(tài)常量定義結束///////////////////////
/**
* 鎖標志對應的key
*/
private String key;
/**
* 鎖的有效時間(s)
*/
private int expireTime = EXPIRE;
/**
* 請求鎖的超時時間(ms)
*/
private long timeOut = TIME_OUT;
/**
* 鎖flag
*/
private volatile boolean isLocked = false;
/**
* Redis管理模板
*/
private StringRedisTemplate redisTemplate;
/**
* 構造方法
*
* @param redisTemplate Redis管理模板
* @param key 鎖定key
* @param expireTime 鎖過期時間 （秒）
* @param timeOut 請求鎖超時時間 (毫秒)
*/
public RedisLock(StringRedisTemplate redisTemplate, String key, int expireTime, long timeOut) {
this.key = key;
this.expireTime = expireTime;
this.timeOut = timeOut;
this.redisTemplate = redisTemplate;
}
/**
* 構造方法
*
* @param redisTemplate Redis管理模板
* @param key 鎖定key
* @param expireTime 鎖過期時間
*/
public RedisLock(StringRedisTemplate redisTemplate, String key, int expireTime) {
this.key = key;
this.expireTime = expireTime;
this.redisTemplate = redisTemplate;
}
/**
* 構造方法(默認請求鎖超時時間30秒，鎖過期時間60秒)
*
* @param redisTemplate Redis管理模板
* @param key 鎖定key
*/
public RedisLock(StringRedisTemplate redisTemplate, String key) {
this.key = key;
this.redisTemplate = redisTemplate;
}
public boolean lock() {
// 系統當前時間，納秒
long nowTime = System.nanoTime();
// 請求鎖超時時間耿芹，納秒
long timeout = timeOut * 1000000;
final Random random = new Random();
// 不斷循環(huán)向Master節(jié)點請求鎖崭篡，當請求時間(System.nanoTime() - nano)超過設定的超時時間則放棄請求鎖
// 這個可以防止一個客戶端在某個宕掉的master節(jié)點上阻塞過長時間
// 如果一個master節(jié)點不可用了，應該盡快嘗試下一個master節(jié)點
while ((System.nanoTime() - nowTime) < timeout) {
// 將鎖作為key存儲到redis緩存中吧秕，存儲成功則獲得鎖
if (redisTemplate.opsForValue().setIfAbsent(key, LOCKED)) {
isLocked = true;
// 設置鎖的有效期琉闪，也是鎖的自動釋放時間，也是一個客戶端在其他客戶端能搶占鎖之前可以執(zhí)行任務的時間
// 可以防止因異常情況無法釋放鎖而造成死鎖情況的發(fā)生
redisTemplate.expire(key, expireTime, TimeUnit.SECONDS);
// 上鎖成功結束請求
break;
}
// 獲取鎖失敗時砸彬，應該在隨機延時后進行重試颠毙，避免不同客戶端同時重試導致誰都無法拿到鎖的情況出現
// 睡眠10毫秒后繼續(xù)請求鎖
try {
Thread.sleep(10, random.nextInt(50000));
} catch (InterruptedException e) {
logger.error("獲取分布式鎖休眠被中斷：", e);
}
}
return isLocked;
}
public boolean isLock() {
redisTemplate.getConnectionFactory().getConnection().time();
return redisTemplate.hasKey(key);
}
public void unlock() {
// 釋放鎖
// 不管請求鎖是否成功，只要已經上鎖砂碉，客戶端都會進行釋放鎖的操作
if (isLocked) {
redisTemplate.delete(key);
}
}
}
調用鎖：
public void redisLock(int i) {
RedisLock redisLock = new RedisLock(redisTemplate, "redisLockKey:"+i % 10, 5*60 , 500);
try {
long now = System.currentTimeMillis();
if (redisLock.lock()) {
logger.info("=" + (System.currentTimeMillis() - now));
// TODO 獲取到鎖要執(zhí)行的代碼塊
logger.info("j:" + j ++);
} else {
logger.info("k:" + k ++);
}
} catch (Exception e) {
logger.info(e.getMessage(), e);
} finally {
// 一定要釋放鎖
redisLock.unlock();
}
}
使用redis的setnx()蛀蜜、get()、getset()方法增蹭，用于分布式鎖
原理
這個方案的背景主要是在setnx()和expire()的方案上針對可能存在的死鎖問題滴某，做了一版優(yōu)化。
那么先說明一下這三個命令滋迈，對于setnx()和get()這兩個命令霎奢，相信不用再多說什么。那么getset()命令饼灿？這個命令主要有兩個參數 getset(key幕侠，newValue)。該方法是原子的碍彭，對key設置newValue這個值晤硕，并且返回key原來的舊值。假設key原來是不存在的庇忌，那么多次執(zhí)行這個命令舞箍，會出現下邊的效果：
- getset(key, "value1") 返回nil 此時key的值會被設置為value1
- getset(key, "value2") 返回value1 此時key的值會被設置為value2
- 依次類推！
介紹完要使用的命令后漆枚，具體的使用步驟如下：
setnx(lockkey, 當前時間+過期超時時間) 创译，如果返回1抵知，則獲取鎖成功墙基；如果返回0則沒有獲取到鎖，轉向2刷喜。
get(lockkey)獲取值oldExpireTime 残制，并將這個value值與當前的系統時間進行比較，如果小于當前系統時間掖疮，則認為這個鎖已經超時初茶，可以允許別的請求重新獲取，轉向3浊闪。
計算newExpireTime=當前時間+過期超時時間恼布，然后getset(lockkey, newExpireTime) 會返回當前l(fā)ockkey的值currentExpireTime螺戳。
判斷currentExpireTime與oldExpireTime 是否相等，如果相等折汞，說明當前getset設置成功倔幼，獲取到了鎖。如果不相等爽待，說明這個鎖又被別的請求獲取走了损同，那么當前請求可以直接返回失敗，或者繼續(xù)重試鸟款。
在獲取到鎖之后膏燃，當前線程可以開始自己的業(yè)務處理，當處理完畢后何什，比較自己的處理時間和對于鎖設置的超時時間组哩，如果小于鎖設置的超時時間，則直接執(zhí)行delete釋放鎖处渣；如果大于鎖設置的超時時間禁炒，則不需要再鎖進行處理。
可能存在的問題
問題1: 在“get(lockkey)獲取值oldExpireTime ”這個操作與“getset(lockkey, newExpireTime) ”這個操作之間霍比，如果有N個線程在get操作獲取到相同的oldExpireTime后幕袱，然后都去getset，會不會返回的newExpireTime都是一樣的悠瞬，都會是成功们豌，進而都獲取到鎖？浅妆？望迎？
我認為這套方案是不存在這個問題的。依據有兩條: 第一凌外，redis是單進程單線程模式辩尊，串行執(zhí)行命令。 第二康辑，在串行執(zhí)行的前提條件下摄欲，getset之后會比較返回的currentExpireTime與oldExpireTime 是否相等。
問題2: 在“get(lockkey)獲取值oldExpireTime ”這個操作與“getset(lockkey, newExpireTime) ”這個操作之間疮薇，如果有N個線程在get操作獲取到相同的oldExpireTime后胸墙，然后都去getset，假設第1個線程獲取鎖成功按咒，其他鎖獲取失敗迟隅，但是獲取鎖失敗的線程它發(fā)起的getset命令確實執(zhí)行了，這樣會不會造成第一個獲取鎖的線程設置的鎖超時時間一直在延長？智袭？奔缠？
我認為這套方案確實存在這個問題的可能。但我個人認為這個微笑的誤差是可以忽略的吼野，不過技術方案上存在缺陷添坊，大家可以自行抉擇哈。
問題3: 這個方案必須要保證分布式服務器的時間一定要同步箫锤，否則這個鎖就會出問題贬蛙。
具體實現
鎖具體實現RedisLock：
package com.xiaolyuh.lock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.util.Random;
import java.util.concurrent.TimeUnit;
/**
* Redis分布式鎖（這種方式服務器時間一定要同步，否則會出問題）
*
* @author yuhao.wangwang
* @version 1.0
* @date 2017年11月3日 上午10:21:27
*/
public class RedisLock2 {
/**
* 默認請求鎖的超時時間(ms 毫秒)
*/
private static final long TIME_OUT = 100;
/**
* 默認鎖的有效時間(s)
*/
public static final int EXPIRE = 60;
private static Logger logger = LoggerFactory.getLogger(RedisLock2.class);
private StringRedisTemplate redisTemplate;
/**
* 鎖標志對應的key
*/
private String lockKey;
/**
* 鎖的有效時間(s)
*/
private int expireTime = EXPIRE;
/**
* 請求鎖的超時時間(ms)
*/
private long timeOut = TIME_OUT;
/**
* 鎖的有效時間
*/
private long expires = 0;
/**
* 鎖標記
*/
private volatile boolean locked = false;
final Random random = new Random();
/**
* 使用默認的鎖過期時間和請求鎖的超時時間
*
* @param redisTemplate
* @param lockKey 鎖的key（Redis的Key）
*/
public RedisLock2(StringRedisTemplate redisTemplate, String lockKey) {
this.redisTemplate = redisTemplate;
this.lockKey = lockKey + "_lock";
}
/**
* 使用默認的請求鎖的超時時間谚攒，指定鎖的過期時間
*
* @param redisTemplate
* @param lockKey 鎖的key（Redis的Key）
* @param expireTime 鎖的過期時間(單位：秒)
*/
public RedisLock2(StringRedisTemplate redisTemplate, String lockKey, int expireTime) {
this(redisTemplate, lockKey);
this.expireTime = expireTime;
}
/**
* 使用默認的鎖的過期時間阳准，指定請求鎖的超時時間
*
* @param redisTemplate
* @param lockKey 鎖的key（Redis的Key）
* @param timeOut 請求鎖的超時時間(單位：毫秒)
*/
public RedisLock2(StringRedisTemplate redisTemplate, String lockKey, long timeOut) {
this(redisTemplate, lockKey);
this.timeOut = timeOut;
}
/**
* 鎖的過期時間和請求鎖的超時時間都是用指定的值
*
* @param redisTemplate
* @param lockKey 鎖的key（Redis的Key）
* @param expireTime 鎖的過期時間(單位：秒)
* @param timeOut 請求鎖的超時時間(單位：毫秒)
*/
public RedisLock2(StringRedisTemplate redisTemplate, String lockKey, int expireTime, long timeOut) {
this(redisTemplate, lockKey, expireTime);
this.timeOut = timeOut;
}
/**
* @return 獲取鎖的key
*/
public String getLockKey() {
return lockKey;
}
/**
* 獲得 lock.
* 實現思路: 主要是使用了redis 的setnx命令,緩存了鎖.
* reids緩存的key是鎖的key,所有的共享, value是鎖的到期時間(注意:這里把過期時間放在value了,沒有時間上設置其超時時間)
* 執(zhí)行過程:
* 1.通過setnx嘗試設置某個key的值,成功(當前沒有這個鎖)則返回,成功獲得鎖
* 2.鎖已經存在則獲取鎖的到期時間,和當前時間比較,超時的話,則設置新的值
*
* @return true if lock is acquired, false acquire timeouted
* @throws InterruptedException in case of thread interruption
*/
public boolean lock() {
// 請求鎖超時時間，納秒
long timeout = timeOut * 1000000;
// 系統當前時間馏臭，納秒
long nowTime = System.nanoTime();
while ((System.nanoTime() - nowTime) < timeout) {
// 分布式服務器有時差野蝇，這里給1秒的誤差值
expires = System.currentTimeMillis() + expireTime + 1;
String expiresStr = String.valueOf(expires); //鎖到期時間
if (redisTemplate.opsForValue().setIfAbsent(lockKey, expiresStr)) {
locked = true;
// 設置鎖的有效期，也是鎖的自動釋放時間括儒，也是一個客戶端在其他客戶端能搶占鎖之前可以執(zhí)行任務的時間
// 可以防止因異常情況無法釋放鎖而造成死鎖情況的發(fā)生
redisTemplate.expire(lockKey, expireTime, TimeUnit.SECONDS);
// 上鎖成功結束請求
return true;
}
String currentValueStr = redisTemplate.opsForValue().get(lockKey); //redis里的時間
if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
//判斷是否為空绕沈，不為空的情況下，如果被其他線程設置了值帮寻，則第二個條件判斷是過不去的
// lock is expired
String oldValueStr = redisTemplate.opsForValue().getAndSet(lockKey, expiresStr);
//獲取上一個鎖到期時間乍狐，并設置現在的鎖到期時間，
//只有一個線程才能獲取上一個線上的設置時間固逗，因為jedis.getSet是同步的
if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
//防止誤刪（覆蓋浅蚪，因為key是相同的）了他人的鎖——這里達不到效果，這里值會被覆蓋烫罩，但是因為什么相差了很少的時間惜傲，所以可以接受
//[分布式的情況下]:如過這個時候，多個線程恰好都到了這里贝攒，但是只有一個線程的設置值和當前值相同盗誊，他才有權利獲取鎖
// lock acquired
locked = true;
return true;
}
}
/*
延遲10 毫秒, 這里使用隨機時間可能會好一點,可以防止饑餓進程的出現,即,當同時到達多個進程,
只會有一個進程獲得鎖,其他的都用同樣的頻率進行嘗試,后面有來了一些進行,也以同樣的頻率申請鎖,這將可能導致前面來的鎖得不到滿足.
使用隨機的等待時間可以一定程度上保證公平性
*/
try {
Thread.sleep(10, random.nextInt(50000));
} catch (InterruptedException e) {
logger.error("獲取分布式鎖休眠被中斷：", e);
}
}
return locked;
}
/**
* 解鎖
*/
public synchronized void unlock() {
// 只有加鎖成功并且鎖還有效才去釋放鎖
if (locked && expires > System.currentTimeMillis()) {
redisTemplate.delete(lockKey);
locked = false;
}
}
}
調用方式：
public void redisLock2(int i) {
RedisLock2 redisLock2 = new RedisLock2(redisTemplate, "redisLock:" + i % 10, 5 * 60, 500);
try {
long now = System.currentTimeMillis();
if (redisLock2.lock()) {
logger.info("=" + (System.currentTimeMillis() - now));
// TODO 獲取到鎖要執(zhí)行的代碼塊
logger.info("j:" + j++);
} else {
logger.info("k:" + k++);
}
} catch (Exception e) {
logger.info(e.getMessage(), e);
} finally {
redisLock2.unlock();
}
}
對于上面兩種redis實現分布式鎖的方案都有一個問題：
- 就是你獲取鎖后執(zhí)行業(yè)務邏輯的代碼只能在redis鎖的有效時間之內，因為隘弊，redis的key到期后會自動清除哈踱，這個鎖就算釋放了。所以這個鎖的有效時間一定要結合業(yè)務做好評估长捧。
- 這兩種方式解鎖的時候是直接刪除key嚣鄙，假如C1獲取到了鎖，這個時候redis掛了串结，并且數據沒有持久化，等redis服務啟動起來，C2請求過來獲取到了鎖肌割。但是C1請求現在執(zhí)行完了刪除了key卧蜓，這個時候就把C2的鎖刪掉了。（在下一篇文章中有解決方案）
使用redis的SET resource-name anystring NX EX max-lock-time方式來實現分布式鎖
下一篇文章介紹
Spring-data-redis + redis 分布式鎖（二）
源碼： https://github.com/wyh-spring-ecosystem-student/spring-boot-student/tree/releases
spring-boot-student-data-redis-distributed-lock 工程
參考：
