一、使用分布式鎖要滿足的幾個條件：
1、系統(tǒng)是一個分布式系統(tǒng)（關鍵是分布式例隆，單機的可以使用ReentrantLock或者synchronized代碼塊來實現(xiàn)）
2论巍、共享資源（各個系統(tǒng)訪問同一個資源，資源的載體可能是傳統(tǒng)關系型數(shù)據(jù)庫或者NoSQL）
3地沮、同步訪問（即有很多個進程同事訪問同一個共享資源嗜浮。沒有同步訪問，誰管你資源競爭不競爭）
二摩疑、應用的場景例子
管理后臺的部署架構（多臺tomcat服務器+redis【多臺tomcat服務器訪問一臺redis】+mysql【多臺tomcat服務器訪問一臺服務器上的mysql】）就滿足使用分布式鎖的條件危融。多臺服務器要訪問redis全局緩存的資源，如果不使用分布式鎖就會出現(xiàn)問題雷袋。 看如下偽代碼：
long N=0L;
//N從redis獲取值
if(N<5){
N++吉殃；
//N寫回redis
}
上面的代碼主要實現(xiàn)的功能：
從redis獲取值N，對數(shù)值N進行邊界檢查楷怒，自加1蛋勺，然后N寫回redis中。 這種應用場景很常見率寡，像秒殺迫卢，全局遞增ID、IP訪問限制等冶共。
以IP訪問限制來說乾蛤，惡意攻擊者可能發(fā)起無限次訪問每界，并發(fā)量比較大，分布式環(huán)境下對N的邊界檢查就不可靠家卖，因為從redis讀的N可能已經是臟數(shù)據(jù)眨层。
傳統(tǒng)的加鎖的做法（如java的synchronized和Lock）也沒用，因為這是分布式環(huán)境上荡，這個同步問題的救火隊員也束手無策趴樱。在這危急存亡之秋，分布式鎖終于有用武之地了酪捡。
分布式鎖可以基于很多種方式實現(xiàn)叁征，比如zookeeper、redis...逛薇。不管哪種方式捺疼，他的基本原理是不變的：用一個狀態(tài)值表示鎖，對鎖的占用和釋放通過狀態(tài)值來標識永罚。
這里主要講如何用redis實現(xiàn)分布式鎖啤呼。
三、使用redis的setNX命令實現(xiàn)分布式鎖
1呢袱、實現(xiàn)的原理
Redis為單進程單線程模式官扣，采用隊列模式將并發(fā)訪問變成串行訪問，且多客戶端對Redis的連接并不存在競爭關系羞福。redis的SETNX命令可以方便的實現(xiàn)分布式鎖惕蹄。
2、基本命令解析
1）setNX（SET if Not eXists）
語法：
SETNX key value
將 key 的值設為 value 坯临，當且僅當 key 不存在焊唬。
若給定的 key 已經存在，則 SETNX 不做任何動作看靠。
SETNX 是『SET if Not eXists』(如果不存在赶促，則 SET)的簡寫
返回值：
設置成功，返回 1 挟炬。
設置失敗鸥滨，返回 0 。
例子：
redis> EXISTS job# job 不存在
(integer) 0
redis> SETNX job "programmer"# job 設置成功
(integer) 1
redis> SETNX job "code-farmer"# 嘗試覆蓋 job 谤祖，失敗
(integer) 0
redis> GET job# 沒有被覆蓋
"programmer"
所以我們使用執(zhí)行下面的命令
SETNX lock.foo <current Unix time + lock timeout + 1>
如返回1婿滓，則該客戶端獲得鎖，把lock.foo的鍵值設置為時間值表示該鍵已被鎖定粥喜，該客戶端最后可以通過DEL lock.foo來釋放該鎖凸主。
如返回0，表明該鎖已被其他客戶端取得额湘，這時我們可以先返回或進行重試等對方完成或等待鎖超時卿吐。
2）getSET
語法：
GETSET key value
將給定 key 的值設為 value 旁舰，并返回 key 的舊值(old value)。
當 key 存在但不是字符串類型時嗡官，返回一個錯誤箭窜。
返回值：
返回給定 key 的舊值。
當 key 沒有舊值時衍腥，也即是磺樱， key 不存在時，返回 nil 婆咸。
3）get
語法：
GET key
返回值：
當 key 不存在時竹捉，返回 nil ，否則擅耽，返回 key 的值活孩。
如果 key 不是字符串類型物遇，那么返回一個錯誤乖仇。
四、解決死鎖
上面的鎖定邏輯有一個問題：如果一個持有鎖的客戶端失敗或崩潰了不能釋放鎖询兴，該怎么解決乃沙？
我們可以通過鎖的鍵對應的時間戳來判斷這種情況是否發(fā)生了，如果當前的時間已經大于lock.foo的值诗舰，說明該鎖已失效警儒，可以被重新使用。
發(fā)生這種情況時眶根，可不能簡單的通過DEL來刪除鎖蜀铲，然后再SETNX一次（講道理，刪除鎖的操作應該是鎖擁有這執(zhí)行的属百，這里只需要等它超時即可）记劝，當多個客戶端檢測到鎖超時后都會嘗試去釋放它，這里就可能出現(xiàn)一個競態(tài)條件,讓我們模擬一下這個場景：
C0操作超時了族扰，但它還持有著鎖厌丑，C1和C2讀取lock.foo檢查時間戳，先后發(fā)現(xiàn)超時了渔呵。
C1 發(fā)送DEL lock.foo
C1 發(fā)送SETNX lock.foo 并且成功了怒竿。
C2 發(fā)送DEL lock.foo
C2 發(fā)送SETNX lock.foo 并且成功了。
這樣一來扩氢，C1耕驰，C2都拿到了鎖！問題大了录豺！
幸好這種問題是可以避免的朦肘，讓我們來看看C3這個客戶端是怎樣做的：
C3發(fā)送SETNX lock.foo 想要獲得鎖托嚣，由于C0還持有鎖，所以Redis返回給C3一個0
C3發(fā)送GET lock.foo 以檢查鎖是否超時了厚骗，如果沒超時示启，則等待或重試。
反之领舰，如果已超時夫嗓，C3通過下面的操作來嘗試獲得鎖：
GETSET lock.foo <current Unix time + lock timeout + 1>
通過GETSET，C3拿到的時間戳如果仍然是超時的冲秽，那就說明舍咖，C3如愿以償拿到鎖了。
如果在C3之前锉桑，有個叫C4的客戶端比C3快一步執(zhí)行了上面的操作排霉，那么C3拿到的時間戳是個未超時的值，這時民轴，C3沒有如期獲得鎖攻柠，需要再次等待或重試。留意一下后裸，盡管C3沒拿到鎖瑰钮，但它改寫了C4設置的鎖的超時值，不過這一點非常微小的誤差帶來的影響可以忽略不計微驶。
注意：為了讓分布式鎖的算法更穩(wěn)鍵些浪谴，持有鎖的客戶端在解鎖之前應該再檢查一次自己的鎖是否已經超時，再去做DEL操作因苹，因為可能客戶端因為某個耗時的操作而掛起苟耻，操作完的時候鎖因為超時已經被別人獲得，這時就不必解鎖了扶檐。
五凶杖、代碼實現(xiàn)
expireMsecs 鎖持有超時，防止線程在入鎖以后蘸秘，無限的執(zhí)行下去官卡，讓鎖無法釋放
timeoutMsecs 鎖等待超時，防止線程饑餓醋虏，永遠沒有入鎖執(zhí)行代碼的機會
注意：項目里面需要先搭建好redis的相關配置
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.serializer.StringRedisSerializer;
/**
?* Redis distributed lock implementation.
?*
?* @author zhengcanrui
?*/
public class RedisLock {
????private static Logger logger = LoggerFactory.getLogger(RedisLock.class);
????private RedisTemplate redisTemplate;
????private static final int DEFAULT_ACQUIRY_RESOLUTION_MILLIS = 100;
????/**
?????* Lock key path.
?????*/
????private String lockKey;
????/**
?????* 鎖超時時間寻咒，防止線程在入鎖以后，無限的執(zhí)行等待
?????*/
????private int expireMsecs = 60 * 1000;
????/**
?????* 鎖等待時間颈嚼，防止線程饑餓
?????*/
????private int timeoutMsecs = 10 * 1000;
????private volatile boolean locked = false;
????/**
?????* Detailed constructor with default acquire timeout 10000 msecs and lock expiration of 60000 msecs.
?????*
?????* @param lockKey lock key (ex. account:1, ...)
?????*/
????public RedisLock(RedisTemplate redisTemplate, String lockKey) {
????????this.redisTemplate = redisTemplate;
????????this.lockKey = lockKey + "_lock";
????}
????/**
?????* Detailed constructor with default lock expiration of 60000 msecs.
?????*
?????*/
????public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs) {
????????this(redisTemplate, lockKey);
????????this.timeoutMsecs = timeoutMsecs;
????}
????/**
?????* Detailed constructor.
?????*
?????*/
????public RedisLock(RedisTemplate redisTemplate, String lockKey, int timeoutMsecs, int expireMsecs) {
????????this(redisTemplate, lockKey, timeoutMsecs);
????????this.expireMsecs = expireMsecs;
????}
????/**
?????* @return lock key
?????*/
????public String getLockKey() {
????????returnlockKey;
????}
????private String get(final String key) {
????????Object obj = null;
????????try{
????????????obj = redisTemplate.execute(newRedisCallback<Object>() {
????????????????@Override
????????????????public Object doInRedis(RedisConnection connection) throws DataAccessException {
????????????????????StringRedisSerializer serializer = newStringRedisSerializer();
????????????????????byte[] data = connection.get(serializer.serialize(key));
????????????????????connection.close();
????????????????????if(data == null) {
????????????????????????returnnull;
????????????????????}
????????????????????returnserializer.deserialize(data);
????????????????}
????????????});
????????} catch(Exception e) {
????????????logger.error("get redis error, key : {}", key);
????????}
????????returnobj != null? obj.toString() : null;
????}
????private boolean setNX(final String key, final String value) {
????????Object obj = null;
????????try{
????????????obj = redisTemplate.execute(newRedisCallback<Object>() {
????????????????@Override
????????????????public Object doInRedis(RedisConnection connection) throws DataAccessException {
????????????????????StringRedisSerializer serializer = newStringRedisSerializer();
????????????????????Boolean success = connection.setNX(serializer.serialize(key), serializer.serialize(value));
????????????????????connection.close();
????????????????????returnsuccess;
????????????????}
????????????});
????????} catch(Exception e) {
????????????logger.error("setNX redis error, key : {}", key);
????????}
????????returnobj != null? (Boolean) obj : false;
????}
????private String getSet(final String key, final String value) {
????????Object obj = null;
????????try{
????????????obj = redisTemplate.execute(newRedisCallback<Object>() {
????????????????@Override
????????????????public Object doInRedis(RedisConnection connection) throws DataAccessException {
????????????????????StringRedisSerializer serializer = newStringRedisSerializer();
????????????????????byte[] ret = connection.getSet(serializer.serialize(key), serializer.serialize(value));
????????????????????connection.close();
????????????????????returnserializer.deserialize(ret);
????????????????}
????????????});
????????} catch(Exception e) {
????????????logger.error("setNX redis error, key : {}", key);
????????}
????????returnobj != null? (String) obj : null;
????}
????/**
?????* 獲得 lock.
?????* 實現(xiàn)思路: 主要是使用了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 synchronized boolean lock() throws InterruptedException {
????????int timeout = timeoutMsecs;
????????while(timeout >= 0) {
????????????long expires = System.currentTimeMillis() + expireMsecs + 1;
????????????String expiresStr = String.valueOf(expires); //鎖到期時間
????????????if(this.setNX(lockKey, expiresStr)) {
????????????????// lock acquired
????????????????locked = true;
????????????????returntrue;
????????????}
????????????String currentValueStr = this.get(lockKey); //redis里的時間
????????????if(currentValueStr != null&& Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
????????????????//判斷是否為空毛秘，不為空的情況下，如果被其他線程設置了值，則第二個條件判斷是過不去的
????????????????// lock is expired
????????????????String oldValueStr = this.getSet(lockKey, expiresStr);
????????????????//獲取上一個鎖到期時間叫挟，并設置現(xiàn)在的鎖到期時間艰匙，
????????????????//只有一個線程才能獲取上一個線上的設置時間，因為jedis.getSet是同步的
????????????????if(oldValueStr != null&& oldValueStr.equals(currentValueStr)) {
????????????????????//防止誤刪（覆蓋抹恳，因為key是相同的）了他人的鎖——這里達不到效果员凝，這里值會被覆蓋，但是因為什么相差了很少的時間奋献，所以可以接受
????????????????????//[分布式的情況下]:如過這個時候健霹，多個線程恰好都到了這里，但是只有一個線程的設置值和當前值相同瓶蚂，他才有權利獲取鎖
????????????????????// lock acquired
????????????????????locked = true;
????????????????????returntrue;
????????????????}
????????????}
????????????timeout -= DEFAULT_ACQUIRY_RESOLUTION_MILLIS;
????????????/*
????????????????延遲100 毫秒,? 這里使用隨機時間可能會好一點,可以防止饑餓進程的出現(xiàn),即,當同時到達多個進程,
????????????????只會有一個進程獲得鎖,其他的都用同樣的頻率進行嘗試,后面有來了一些進行,也以同樣的頻率申請鎖,這將可能導致前面來的鎖得不到滿足.
????????????????使用隨機的等待時間可以一定程度上保證公平性
?????????????*/
????????????Thread.sleep(DEFAULT_ACQUIRY_RESOLUTION_MILLIS);
????????}
????????returnfalse;
????}
????/**
?????* Acqurired lock release.
?????*/
????public synchronized void unlock() {
????????if(locked) {
????????????redisTemplate.delete(lockKey);
????????????locked = false;
????????}
????}
}
調用：
RedisLock lock = newRedisLock(redisTemplate, key, 10000, 20000);
?try{
????????????if(lock.lock()) {
???????????????????//需要加鎖的代碼
????????????????}
????????????}
????????} catch(InterruptedException e) {
????????????e.printStackTrace();
????????}finally {
????????????//為了讓分布式鎖的算法更穩(wěn)鍵些糖埋，持有鎖的客戶端在解鎖之前應該再檢查一次自己的鎖是否已經超時，再去做DEL操作窃这，因為可能客戶端因為某個耗時的操作而掛起瞳别，
????????????//操作完的時候鎖因為超時已經被別人獲得，這時就不必解鎖了杭攻。 ————這里沒有做
????????????lock.unlock();
????????}
六祟敛、一些問題
1、為什么不直接使用expire設置超時時間朴上，而將時間的毫秒數(shù)其作為value放在redis中垒棋？
如下面的方式，把超時的交給redis處理：
lock(key, expireSec){
isSuccess = setnx key
if(isSuccess)
expire key expireSec
}
這種方式貌似沒什么問題痪宰，但是假如在setnx后，redis崩潰了畔裕，expire就沒有執(zhí)行衣撬，結果就是死鎖了。鎖永遠不會超時扮饶。
2具练、為什么前面的鎖已經超時了，還要用getSet去設置新的時間戳的時間獲取舊的值甜无，然后和外面的判斷超時時間的時間戳比較呢扛点？
因為是分布式的環(huán)境下，可以在前一個鎖失效的時候岂丘，有兩個進程進入到鎖超時的判斷陵究。如：
C0超時了，還持有鎖,C1/C2同時請求進入了方法里面
C1/C2獲取到了C0的超時時間
C1使用getSet方法
C2也執(zhí)行了getSet方法
假如我們不加 oldValueStr.equals(currentValueStr) 的判斷奥帘，將會C1/C2都將獲得鎖铜邮，加了之后，能保證C1和C2只能一個能獲得鎖，一個只能繼續(xù)等待松蒜。
注意：這里可能導致超時時間不是其原本的超時時間扔茅，C1的超時時間可能被C2覆蓋了，但是他們相差的毫秒及其小秸苗，這里忽略了召娜。
