一、基於key/value實現
我們在構建分布式系統的時候，經常需要控制對共享資源的互斥訪問。這個時候我們就涉及到分布式鎖（也稱為全局鎖）的實現，基於目前的各種工具，我們已經有了大量的實現方式，比如：基於Redis的實現、基於Zookeeper的實現。本文將介紹一種基於Consul 的Key/Value存儲來實現分布式鎖以及信號量的方法。
分布式鎖實現
基於Consul的分布式鎖主要利用Key/Value存儲API中的acquire和release操作來實現。acquire和release操作是類似Check-And-Set的操作：
- acquire操作只有當鎖不存在持有者時才會返回true，並且set設置的Value值，同時執行操作的session會持有對該Key的鎖，否則就返回false
- release操作則是使用指定的session來釋放某個Key的鎖，如果指定的session無效，那么會返回false，否則就會set設置Value值，並返回true
具體實現中主要使用了這幾個Key/Value的API：
- create session：https://www.consul.io/api/session.html#session_create
- delete session：https://www.consul.io/api/session.html#delete-session
- KV acquire/release：https://www.consul.io/api/kv.html#create-update-key
基本流程
具體實現
public class Lock {
private static final String prefix = "lock/"; // 同步鎖參數前綴
private ConsulClient consulClient;
private String sessionName;
private String sessionId = null;
private String lockKey;
/**
*
* @param consulClient
* @param sessionName 同步鎖的session名稱
* @param lockKey 同步鎖在consul的KV存儲中的Key路徑，會自動增加prefix前綴，方便歸類查詢
*/
public Lock(ConsulClient consulClient, String sessionName, String lockKey) {
this.consulClient = consulClient;
this.sessionName = sessionName;
this.lockKey = prefix + lockKey;
}
/**
* 獲取同步鎖
*
* @param block 是否阻塞，直到獲取到鎖為止
* @return
*/
public Boolean lock(boolean block) {
if (sessionId != null) {
throw new RuntimeException(sessionId + " - Already locked!");
}
sessionId = createSession(sessionName);
while(true) {
PutParams putParams = new PutParams();
putParams.setAcquireSession(sessionId);
if(consulClient.setKVValue(lockKey, "lock:" + LocalDateTime.now(), putParams).getValue()) {
return true;
} else if(block) {
continue;
} else {
return false;
}
}
}
/**
* 釋放同步鎖
*
* @return
*/
public Boolean unlock() {
PutParams putParams = new PutParams();
putParams.setReleaseSession(sessionId);
boolean result = consulClient.setKVValue(lockKey, "unlock:" + LocalDateTime.now(), putParams).getValue();
consulClient.sessionDestroy(sessionId, null);
return result;
}
/**
* 創建session
* @param sessionName
* @return
*/
private String createSession(String sessionName) {
NewSession newSession = new NewSession();
newSession.setName(sessionName);
return consulClient.sessionCreate(newSession, null).getValue();
}
}
單元測試
下面單元測試的邏輯：通過線程的方式來模擬不同的分布式服務來競爭鎖。多個處理線程同時以阻塞方式來申請分布式鎖，當處理線程獲得鎖之后，Sleep一段隨機事件，以模擬處理業務邏輯，處理完畢之后釋放鎖。
public class TestLock {
private Logger logger = Logger.getLogger(getClass());
@Test
public void testLock() throws Exception {
new Thread(new LockRunner(1)).start();
new Thread(new LockRunner(2)).start();
new Thread(new LockRunner(3)).start();
new Thread(new LockRunner(4)).start();
new Thread(new LockRunner(5)).start();
Thread.sleep(200000L);
}
class LockRunner implements Runnable {
private Logger logger = Logger.getLogger(getClass());
private int flag;
public LockRunner(int flag) {
this.flag = flag;
}
@Override
public void run() {
Lock lock = new Lock(new ConsulClient(), "lock-session", "lock-key");
try {
if (lock.lock(true)) {
logger.info("Thread " + flag + " start!");
Thread.sleep(new Random().nextInt(3000L));
logger.info("Thread " + flag + " end!");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
lock.unlock();
}
}
}
}
單元測試執行結果如下：
2017-04-12 21:28:09,698 INFO [Thread-0] LockRunner - Thread 1 start!
2017-04-12 21:28:12,717 INFO [Thread-0] LockRunner - Thread 1 end!
2017-04-12 21:28:13,219 INFO [Thread-2] LockRunner - Thread 3 start!
2017-04-12 21:28:15,672 INFO [Thread-2] LockRunner - Thread 3 end!
2017-04-12 21:28:15,735 INFO [Thread-1] LockRunner - Thread 2 start!
2017-04-12 21:28:17,788 INFO [Thread-1] LockRunner - Thread 2 end!
2017-04-12 21:28:18,249 INFO [Thread-4] LockRunner - Thread 5 start!
2017-04-12 21:28:19,573 INFO [Thread-4] LockRunner - Thread 5 end!
2017-04-12 21:28:19,757 INFO [Thread-3] LockRunner - Thread 4 start!
2017-04-12 21:28:21,353 INFO [Thread-3] LockRunner - Thread 4 end!
從測試結果我們可以看到，通過分布式鎖的形式來控制並發時，多個同步操作只會有一個操作能夠被執行，其他操作只有在等鎖釋放之后才有機會去執行，所以通過這樣的分布式鎖，我們可以控制共享資源同時只能被一個操作進行執行，以保障數據處理時的分布式並發問題。
優化建議
本文我們實現了基於Consul的簡單分布式鎖，但是在實際運行時，可能會因為各種各樣的意外情況導致unlock操作沒有得到正確地執行，從而使得分布式鎖無法釋放。所以為了更完善的使用分布式鎖，我們還必須實現對鎖的超時清理等控制，保證即使出現了未正常解鎖的情況下也能自動修復，以提升系統的健壯性。那么如何實現呢？請持續關注我的后續分解！
參考文檔
Key/Value的API：https://www.consul.io/api/kv.html
二、基於consul分布式信號量實現
在上面《基於Consul的分布式鎖實現》中我們介紹如何基於Consul的KV存儲來實現分布式互斥鎖。本文將繼續討論基於Consul的分布式鎖實現。信號量是我們在實現並發控制時會經常使用的手段，主要用來限制同時並發線程或進程的數量，比如：Zuul默認情況下就使用信號量來限制每個路由的並發數，以實現不同路由間的資源隔離。
信號量(Semaphore)，有時被稱為信號燈，是在多線程環境下使用的一種設施，是可以用來保證兩個或多個關鍵代碼段不被並發調用。在進入一個關鍵代碼段之前，線程必須獲取一個信號量；一旦該關鍵代碼段完成了，那么該線程必須釋放信號量。其它想進入該關鍵代碼段的線程必須等待直到第一個線程釋放信號量。為了完成這個過程，需要創建一個信號量VI，然后將Acquire Semaphore VI以及Release Semaphore VI分別放置在每個關鍵代碼段的首末端，確認這些信號量VI引用的是初始創建的信號量。如在這個停車場系統中，車位是公共資源，每輛車好比一個線程，看門人起的就是信號量的作用。
實現思路
- 信號量存儲：semaphore/key
- acquired操作：
- 創建session
- 鎖定key競爭者：semaphore/key/session
- 查詢信號量：semaphore/key/.lock，可以獲得如下內容（如果是第一次創建信號量，將獲取不到，這個時候就直接創建）
- 如果持有者已達上限，返回false，如果阻塞模式，就繼續嘗試acquired操作
- 如果持有者未達上限，更新semaphore/key/.lock的內容，將當前線程的sessionId加入到holders中。注意：更新的時候需要設置cas，它的值是“查詢信號量”步驟獲得的“ModifyIndex”值，該值用於保證更新操作的基礎沒有被其他競爭者更新。如果更新成功，就開始執行具體邏輯。如果沒有更新成功，說明有其他競爭者搶占了資源，返回false，阻塞模式下繼續嘗試acquired操作
- release操作：
- 從semaphore/key/.lock的holders中移除當前sessionId
- 刪除semaphore/key/session
- 刪除當前的session
流程圖
代碼實現
public class Semaphore {
private Logger logger = Logger.getLogger(getClass());
private static final String prefix = "semaphore/"; // 信號量參數前綴
private ConsulClient consulClient;
private int limit;
private String keyPath;
private String sessionId = null;
private boolean acquired = false;
/**
*
* @param consulClient consul客戶端實例
* @param limit 信號量上限值
* @param keyPath 信號量在consul中存儲的參數路徑
*/
public Semaphore(ConsulClient consulClient, int limit, String keyPath) {
this.consulClient = consulClient;
this.limit = limit;
this.keyPath = prefix + keyPath;
}
/**
* acquired信號量
*
* @param block 是否阻塞。如果為true，那么一直嘗試，直到獲取到該資源為止。
* @return
* @throws IOException
*/
public Boolean acquired(boolean block) throws IOException {
if(acquired) {
logger.error(sessionId + " - Already acquired");
throw new RuntimeException(sessionId + " - Already acquired");
}
// create session
clearSession();
this.sessionId = createSessionId("semaphore");
logger.debug("Create session : " + sessionId);
// add contender entry
String contenderKey = keyPath + "/" + sessionId;
logger.debug("contenderKey : " + contenderKey);
PutParams putParams = new PutParams();
putParams.setAcquireSession(sessionId);
Boolean b = consulClient.setKVValue(contenderKey, "", putParams).getValue();
if(!b) {
logger.error("Failed to add contender entry : " + contenderKey + ", " + sessionId);
throw new RuntimeException("Failed to add contender entry : " + contenderKey + ", " + sessionId);
}
while(true) {
// try to take the semaphore
String lockKey = keyPath + "/.lock";
String lockKeyValue;
GetValue lockKeyContent = consulClient.getKVValue(lockKey).getValue();
if (lockKeyContent != null) {
// lock值轉換
lockKeyValue = lockKeyContent.getValue();
BASE64Decoder decoder = new BASE64Decoder();
byte[] v = decoder.decodeBuffer(lockKeyValue);
String lockKeyValueDecode = new String(v);
logger.debug("lockKey=" + lockKey + ", lockKeyValueDecode=" + lockKeyValueDecode);
Gson gson = new Gson();
ContenderValue contenderValue = gson.fromJson(lockKeyValueDecode, ContenderValue.class);
// 當前信號量已滿
if(contenderValue.getLimit() == contenderValue.getHolders().size()) {
logger.debug("Semaphore limited " + contenderValue.getLimit() + ", waiting...");
if(block) {
// 如果是阻塞模式，再嘗試
try {
Thread.sleep(100L);
} catch (InterruptedException e) {
}
continue;
}
// 非阻塞模式，直接返回沒有獲取到信號量
return false;
}
// 信號量增加
contenderValue.getHolders().add(sessionId);
putParams = new PutParams();
putParams.setCas(lockKeyContent.getModifyIndex());
boolean c = consulClient.setKVValue(lockKey, contenderValue.toString(), putParams).getValue();
if(c) {
acquired = true;
return true;
}
else
continue;
} else {
// 當前信號量還沒有，所以創建一個，並馬上搶占一個資源
ContenderValue contenderValue = new ContenderValue();
contenderValue.setLimit(limit);
contenderValue.getHolders().add(sessionId);
putParams = new PutParams();
putParams.setCas(0L);
boolean c = consulClient.setKVValue(lockKey, contenderValue.toString(), putParams).getValue();
if (c) {
acquired = true;
return true;
}
continue;
}
}
}
/**
* 創建sessionId
* @param sessionName
* @return
*/
public String createSessionId(String sessionName) {
NewSession newSession = new NewSession();
newSession.setName(sessionName);
return consulClient.sessionCreate(newSession, null).getValue();
}
/**
* 釋放session、並從lock中移除當前的sessionId
* @throws IOException
*/
public void release() throws IOException {
if(this.acquired) {
// remove session from lock
while(true) {
String contenderKey = keyPath + "/" + sessionId;
String lockKey = keyPath + "/.lock";
String lockKeyValue;
GetValue lockKeyContent = consulClient.getKVValue(lockKey).getValue();
if (lockKeyContent != null) {
// lock值轉換
lockKeyValue = lockKeyContent.getValue();
BASE64Decoder decoder = new BASE64Decoder();
byte[] v = decoder.decodeBuffer(lockKeyValue);
String lockKeyValueDecode = new String(v);
Gson gson = new Gson();
ContenderValue contenderValue = gson.fromJson(lockKeyValueDecode, ContenderValue.class);
contenderValue.getHolders().remove(sessionId);
PutParams putParams = new PutParams();
putParams.setCas(lockKeyContent.getModifyIndex());
consulClient.deleteKVValue(contenderKey);
boolean c = consulClient.setKVValue(lockKey, contenderValue.toString(), putParams).getValue();
if(c) {
break;
}
}
}
// remove session key
}
this.acquired = false;
clearSession();
}
public void clearSession() {
if(sessionId != null) {
consulClient.sessionDestroy(sessionId, null);
sessionId = null;
}
}
class ContenderValue implements Serializable {
private Integer limit;
private List<String> holders = new ArrayList<>();
public Integer getLimit() {
return limit;
}
public void setLimit(Integer limit) {
this.limit = limit;
}
public List<String> getHolders() {
return holders;
}
public void setHolders(List<String> holders) {
this.holders = holders;
}
@Override
public String toString() {
return new Gson().toJson(this);
}
}
}
單元測試
下面單元測試的邏輯：通過線程的方式來模擬不同的分布式服務來獲取信號量執行業務邏輯。由於信號量與簡單的分布式互斥鎖有所不同，它不是只限定一個線程可以操作，而是可以控制多個線程的並發，所以通過下面的單元測試，我們設置信號量為3，然后同時啟動15個線程來競爭的情況，來觀察分布式信號量實現的結果如何。
public class TestLock {
private Logger logger = Logger.getLogger(getClass());
@Test
public void testSemaphore() throws Exception {
new Thread(new SemaphoreRunner(1)).start();
new Thread(new SemaphoreRunner(2)).start();
new Thread(new SemaphoreRunner(3)).start();
new Thread(new SemaphoreRunner(4)).start();
new Thread(new SemaphoreRunner(5)).start();
new Thread(new SemaphoreRunner(6)).start();
new Thread(new SemaphoreRunner(7)).start();
new Thread(new SemaphoreRunner(8)).start();
new Thread(new SemaphoreRunner(9)).start();
new Thread(new SemaphoreRunner(10)).start();
Thread.sleep(1000000L);
}
}
public class SemaphoreRunner implements Runnable {
private Logger logger = Logger.getLogger(getClass());
private int flag;
public SemaphoreRunner(int flag) {
this.flag = flag;
}
@Override
public void run() {
Semaphore semaphore = new Semaphore(new ConsulClient(), 3, "mg-init");
try {
if (semaphore.acquired(true)) {
// 獲取到信號量，執行業務邏輯
logger.info("Thread " + flag + " start!");
Thread.sleep(new Random().nextInt(10000));
logger.info("Thread " + flag + " end!");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
try {
// 信號量釋放、Session鎖釋放、Session刪除
semaphore.release();
} catch (IOException e) {
e.printStackTrace();
}
}
}
}
執行結果：
INFO [Thread-6] SemaphoreRunner - Thread 7 start!
INFO [Thread-2] SemaphoreRunner - Thread 3 start!
INFO [Thread-7] SemaphoreRunner - Thread 8 start!
INFO [Thread-2] SemaphoreRunner - Thread 3 end!
INFO [Thread-5] SemaphoreRunner - Thread 6 start!
INFO [Thread-6] SemaphoreRunner - Thread 7 end!
INFO [Thread-9] SemaphoreRunner - Thread 10 start!
INFO [Thread-5] SemaphoreRunner - Thread 6 end!
INFO [Thread-1] SemaphoreRunner - Thread 2 start!
INFO [Thread-7] SemaphoreRunner - Thread 8 end!
INFO [Thread-10] SemaphoreRunner - Thread 11 start!
INFO [Thread-10] SemaphoreRunner - Thread 11 end!
INFO [Thread-12] SemaphoreRunner - Thread 13 start!
INFO [Thread-1] SemaphoreRunner - Thread 2 end!
INFO [Thread-3] SemaphoreRunner - Thread 4 start!
INFO [Thread-9] SemaphoreRunner - Thread 10 end!
INFO [Thread-0] SemaphoreRunner - Thread 1 start!
INFO [Thread-3] SemaphoreRunner - Thread 4 end!
INFO [Thread-14] SemaphoreRunner - Thread 15 start!
INFO [Thread-12] SemaphoreRunner - Thread 13 end!
INFO [Thread-0] SemaphoreRunner - Thread 1 end!
INFO [Thread-13] SemaphoreRunner - Thread 14 start!
INFO [Thread-11] SemaphoreRunner - Thread 12 start!
INFO [Thread-13] SemaphoreRunner - Thread 14 end!
INFO [Thread-4] SemaphoreRunner - Thread 5 start!
INFO [Thread-4] SemaphoreRunner - Thread 5 end!
INFO [Thread-8] SemaphoreRunner - Thread 9 start!
INFO [Thread-11] SemaphoreRunner - Thread 12 end!
INFO [Thread-14] SemaphoreRunner - Thread 15 end!
INFO [Thread-8] SemaphoreRunner - Thread 9 end!
從測試結果，我們可以發現當信號量持有者數量達到信號量上限3的時候，其他競爭者就開始進行等待了，只有當某個持有者釋放信號量之后，才會有新的線程變成持有者，從而開始執行自己的業務邏輯。所以，分布式信號量可以幫助我們有效的控制同時操作某個共享資源的並發數。
優化建議與參考文檔
同前文一樣，這里只是做了簡單的實現。線上應用還必須加入TTL的session清理以及對.lock資源中的無效holder進行清理的機制。
參考文檔：
https://www.consul.io/docs/guides/semaphore.html
轉自：http://mp.weixin.qq.com/s?__biz=MzAxODcyNjEzNQ==&mid=2247483857&idx=1&sn=495c0faad9bc237132aca49e722022ec&chksm=9bd0ac49aca7255fec67f9364fab63638b30e7a69fc0771f5977a6cc9a38856879b64832bc67&scene=21#wechat_redirect