分布式可重入鎖 · JAVA

[TOC] # 什么是可重入鎖? 就是一個鎖住的資源(比如代碼塊)可以被持有鎖的線程反復進入這個可重入鎖擁有一個計數器當同一個線程再次進入鎖住的代碼塊時,計數器+1 當這個線程離開代碼塊的時候,如果技術器大于1,則計時器-1 當這個線程離開代碼塊的時候,如果計數器等于1,則釋放鎖 JDK提供的可重入鎖是ReentrantLock,當這個可重入鎖只能用于單進程在分布式環境下,redis的可重入鎖需要自己來實現 # 如何實現可重入？首先鎖信息(指redis中lockKey關聯的value值) 必須得設計的能負載更多信息，之前non-reentrant時value直接就是一個超時時間，但是要實現可重入單超時時間是不夠的，必須要標識鎖是被誰持有的，也就是說要標識分布式環境中的線程，還要記錄鎖被入了多少次。 # 如何在分布式線程中標識唯一線程？ MAC地址 + jvm進程ID + 線程ID（或者線程地址都行）,三者結合即可唯一分布式環境中的線程。 # 實現重點是lock方法，代碼已有非常詳細的注釋 ~~~ package cc.lixiaohui.lock.redis; import java.io.IOException; import java.net.SocketAddress; import java.util.concurrent.TimeUnit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import redis.clients.jedis.Jedis; import cc.lixiaohui.lock.AbstractLock; import cc.lixiaohui.lock.Lock; import cc.lixiaohui.lock.time.nio.client.TimeClient; import cc.lixiaohui.lock.util.LockInfo; /** * 基于Redis的SETNX操作實現的分布式鎖, 獲取鎖時最好用tryLock(long time, TimeUnit unit), 以免網路問題而導致線程一直阻塞. * <a href="http://redis.io/commands/setnx">SETNC操作參考資料.</a> * * 可重入實現關鍵: * <ul> * <li>在分布式環境中如何確定一個線程? mac地址 + jvm pid + threadId (mac地址唯一, jvm * pid在單機內唯一, threadId在單jvm內唯一)</li> * <li>任何一個線程從redis拿到value值后都需要能確定該鎖是否被自己持有, 因此value值要有以下特性: 保存持有鎖的主機(mac), jvm * pid, 持有鎖的線程ID, 重復持有鎖的次數</li> * </ul> * * redis中value設計如下(in json): * <pre> * { * expires : expire time in long * mac : mac address of lock holder's machine * pid : jvm process id * threadId : lock holder thread id * count : hold count(for use of reentrancy) * } * 由{@link LockInfo LockInfo}表示. * </pre> * * Usage Example: * <pre> * {@link Lock} lock = new {@link ReentrantLock}(jedis, "lockKey", lockExpires, timeServerAddr); * if (lock.tryLock(3, TimeUnit.SECONDS)) { * try { * // do something * } catch (Exception e) { * lock.unlock(); * } * } * </pre> * * * @author lixiaohui * @date 2016年9月15日下午2:52:38 * */ public class ReentrantLock extends AbstractLock { private Jedis jedis; private TimeClient timeClient; // 鎖的名字 protected String lockKey; // 鎖的有效時長(毫秒) protected long lockExpires; private static final Logger logger = LoggerFactory.getLogger(ReentrantLock.class); public ReentrantLock(Jedis jedis, String lockKey, long lockExpires, SocketAddress timeServerAddr) throws IOException { this.jedis = jedis; this.lockKey = lockKey; this.lockExpires = lockExpires; timeClient = new TimeClient(timeServerAddr); } // 阻塞式獲取鎖的實現 protected boolean lock(boolean useTimeout, long time, TimeUnit unit, boolean interrupt) throws InterruptedException { if (interrupt) { checkInterruption(); } // 超時控制的時間可以從本地獲取, 因為這個和鎖超時沒有關系, 只是一段時間區間的控制 long start = localTimeMillis(); long timeout = unit.toMillis(time); // if !useTimeout, then it's useless // walkthrough // 1. lockKey未關聯value, 直接設置lockKey, 成功獲取到鎖, return true // 2. lock 已過期, 用getset設置lockKey, 判斷返回的舊的LockInfo // 2.1 若仍是超時的, 則成功獲取到鎖, return true // 2.2 若不是超時的, 則進入下一次循環重新開始步驟1 // 3. lock沒過期, 判斷是否是當前線程持有 // 3.1 是, 則計數加 1, return true // 3.2 否, 則進入下一次循環重新開始步驟1 // note: 每次進入循環都檢查 : 1.是否超時, 若是則return false; 2.是否檢查中斷(interrupt)被中斷, // 若需檢查中斷且被中斷, 則拋InterruptedException while (useTimeout ? !isTimeout(start, timeout) : true) { if (interrupt) { checkInterruption(); } long lockExpireTime = serverTimeMillis() + lockExpires + 1;// 鎖超時時間 String newLockInfoJson = LockInfo.newForCurrThread(lockExpireTime).toString(); if (jedis.setnx(lockKey, newLockInfoJson) == 1) { // 條件能成立的唯一情況就是redis中lockKey還未關聯value // TODO 成功獲取到鎖, 設置相關標識 logger.debug("{} get lock(new), lockInfo: {}", Thread.currentThread().getName(), newLockInfoJson); locked = true; return true; } // value已有值, 但不能說明鎖被持有, 因為鎖可能expired了 String currLockInfoJson = jedis.get(lockKey); // 若這瞬間鎖被delete了 if (currLockInfoJson == null) { continue; } LockInfo currLockInfo = LockInfo.fromString(currLockInfoJson); // 競爭條件只可能出現在鎖超時的情況, 因為如果沒有超時, 線程發現鎖并不是被自己持有, 線程就不會去動value if (isTimeExpired(currLockInfo.getExpires())) { // 鎖超時了 LockInfo oldLockInfo = LockInfo.fromString(jedis.getSet(lockKey, newLockInfoJson)); if (oldLockInfo != null && isTimeExpired(oldLockInfo.getExpires())) { // TODO 成功獲取到鎖, 設置相關標識 logger.debug("{} get lock(new), lockInfo: {}", Thread.currentThread().getName(), newLockInfoJson); locked = true; return true; } } else { // 鎖未超時, 不會有競爭情況 if (isHeldByCurrentThread(currLockInfo)) { // 當前線程持有 // TODO 成功獲取到鎖, 設置相關標識 currLockInfo.setExpires(serverTimeMillis() + lockExpires + 1); // 設置新的鎖超時時間 currLockInfo.incCount(); jedis.set(lockKey, currLockInfo.toString()); logger.debug("{} get lock(inc), lockInfo: {}", Thread.currentThread().getName(), currLockInfo); locked = true; return true; } } } locked = false; return false; } public boolean tryLock() { long lockExpireTime = serverTimeMillis() + lockExpires + 1; String newLockInfo = LockInfo.newForCurrThread(lockExpireTime).toString(); if (jedis.setnx(lockKey, newLockInfo) == 1) { locked = true; return true; } String currLockInfoJson = jedis.get(lockKey); if (currLockInfoJson == null) { // 再一次嘗試獲取 if (jedis.setnx(lockKey, newLockInfo) == 1) { locked = true; return true; } else { locked = false; return false; } } LockInfo currLockInfo = LockInfo.fromString(currLockInfoJson); if (isTimeExpired(currLockInfo.getExpires())) { LockInfo oldLockInfo = LockInfo.fromString(jedis.getSet(lockKey, newLockInfo)); if (oldLockInfo != null && isTimeExpired(oldLockInfo.getExpires())) { locked = true; return true; } } else { if (isHeldByCurrentThread(currLockInfo)) { currLockInfo.setExpires(serverTimeMillis() + lockExpires + 1); currLockInfo.incCount(); jedis.set(lockKey, currLockInfo.toString()); locked = true; return true; } } locked = false; return false; } /** * Queries if this lock is held by any thread. * * @return {@code true} if any thread holds this lock and {@code false} * otherwise */ public boolean isLocked() { // walkthrough // 1. lockKey未關聯value, return false // 2. 若 lock 已過期, return false, 否則 return true if (!locked) { // 本地locked為false, 肯定沒加鎖 return false; } String json = jedis.get(lockKey); if (json == null) { return false; } if (isTimeExpired(LockInfo.fromString(json).getExpires())) { return false; } return true; } @Override protected void unlock0() { // walkthrough // 1. 若鎖過期, return // 2. 判斷自己是否是鎖的owner // 2.1 是, 若 count = 1, 則刪除lockKey; 若 count > 1, 則計數減 1, return // 2.2 否, 則拋異常 IllegalMonitorStateException, reutrn // done, return LockInfo currLockInfo = LockInfo.fromString(jedis.get(lockKey)); if (isTimeExpired(currLockInfo.getExpires())) { return; } if (isHeldByCurrentThread(currLockInfo)) { if (currLockInfo.getCount() == 1) { jedis.del(lockKey); logger.debug("{} unlock(del), lockInfo: null", Thread.currentThread().getName()); } else { currLockInfo.decCount(); // 持有鎖計數減1 String json = currLockInfo.toString(); jedis.set(lockKey, json); logger.debug("{} unlock(dec), lockInfo: {}", Thread.currentThread().getName(), json); } } else { throw new IllegalMonitorStateException(String.format("current thread[%s] does not holds the lock", Thread.currentThread().toString())); } } public void release() { jedis.close(); timeClient.close(); } public boolean isHeldByCurrentThread() { return isHeldByCurrentThread(LockInfo.fromString(jedis.get(lockKey))); } // ------------------- utility methods ------------------------ private boolean isHeldByCurrentThread(LockInfo lockInfo) { return lockInfo.isCurrentThread(); } private void checkInterruption() throws InterruptedException { if (Thread.currentThread().isInterrupted()) { throw new InterruptedException(); } } private boolean isTimeExpired(long time) { return time < serverTimeMillis(); } private boolean isTimeout(long start, long timeout) { // 這里拿本地的時間來比較 return start + timeout < System.currentTimeMillis(); } private long serverTimeMillis() { return timeClient.currentTimeMillis(); } private long localTimeMillis() { return System.currentTimeMillis(); } } ~~~ # 測試 5個線程，每個線程都是不同的jedis連接，模擬分布式環境，線程的任務就是不斷的去嘗試重入地獲取鎖，重入的次數為隨機但在0-5之間。代碼 ~~~ package cc.lixiaohui.DistributedLock.DistributedLock; import java.io.IOException; import java.net.InetSocketAddress; import java.net.SocketAddress; import java.util.ArrayList; import java.util.List; import java.util.Random; import java.util.concurrent.TimeUnit; import org.junit.Test; import redis.clients.jedis.Jedis; import cc.lixiaohui.lock.redis.ReentrantLock; /** * @author lixiaohui * @date 2016年9月28日下午8:41:36 * */ public class ReentrantTest { final int EXPIRES = 10 * 1000; final String LOCK_KEY = "lock.lock"; final SocketAddress TIME_SERVER_ADDR = new InetSocketAddress("localhost", 9999); @Test public void test() throws Exception { // 創建5個線程不停地去重入(隨機次數n, 0 <= n <=5)獲取鎖 List<Thread> threads = createThreads(5); //開始任務 for (Thread t : threads) { t.start(); } // 執行60秒 Thread.sleep(60 * 1000); //停止所有線程 Task.alive = false; // 等待所有線程終止 for (Thread t : threads) { t.join(); } } // 創建count個線程，每個線程都是不同的jedis連接以及不同的與時間服務器的連接 private List<Thread> createThreads(int count) throws IOException { List<Thread> threads = new ArrayList<Thread>(); for (int i = 0; i < count; i++) { Jedis jedis = new Jedis("localhost", 6379); ReentrantLock lock = new ReentrantLock(jedis, LOCK_KEY, EXPIRES, TIME_SERVER_ADDR); Task task = new Task(lock); Thread t = new Thread(task); threads.add(t); } return threads; } private static class Task implements Runnable { private ReentrantLock lock; private final int MAX_ENTRANT = 5; private final Random random = new Random(); private static boolean alive = true; Task(ReentrantLock lock) { this.lock = lock; } public void run() { while (alive) { int times = random.nextInt(MAX_ENTRANT); doLock(times); } } private void doLock(int times) { if (lock.tryLock(5, TimeUnit.SECONDS)) { try { if (times > 0) { doLock(--times); } } finally { if (lock != null) { lock.unlock(); } } } } } } ~~~ 測試結果下面是一部分日志輸出： ~~~ 21:09:22.328-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(new), lockInfo: {"count":1,"expires":1475068172329,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.330-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(inc), lockInfo: {"count":2,"expires":1475068172330,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.331-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(inc), lockInfo: {"count":3,"expires":1475068172332,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.332-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(inc), lockInfo: {"count":4,"expires":1475068172333,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.334-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(inc), lockInfo: {"count":5,"expires":1475068172334,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.335-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(dec), lockInfo: {"count":4,"expires":1475068172334,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.336-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(dec), lockInfo: {"count":3,"expires":1475068172334,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.338-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(dec), lockInfo: {"count":2,"expires":1475068172334,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.339-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(dec), lockInfo: {"count":1,"expires":1475068172334,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.340-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(del), lockInfo: null 21:09:22.341-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(new), lockInfo: {"count":1,"expires":1475068172341,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} 21:09:22.341-[DEBUG] Thread-0 ReentrantLock - Thread-0 unlock(del), lockInfo: null 21:09:22.342-[DEBUG] Thread-0 ReentrantLock - Thread-0 get lock(new), lockInfo: {"count":1,"expires":1475068172342,"jvmPid":22224,"mac":"28-D2-44-0E-0D-9A","threadId":11} ~~~