分布式锁 redis实现

beston123 · beston123 · commit da5d934a654f · 2019-09-26T19:16:31.000+08:00
diff --git a/src/main/java/code/cache/CacheSnowCrash.java b/src/main/java/code/cache/CacheSnowCrash.java
@@ -65,7 +65,7 @@ public Object getWithLock(String key) throws InterruptedException {
         Object value = cache.get(key);
         if (value == null) { //代表缓存值过期
             String lockKey = key;
-            String lockSeq = lock.getLock(lockKey);
+            lock.lock(lockKey);
             try {
                 //双重判断
                 if((value = cache.get(key)) == null){
@@ -74,7 +74,7 @@ public Object getWithLock(String key) throws InterruptedException {
                     return value;
                 }
             } finally {
-                lock.unlock(lockKey, lockSeq);
+                lock.unlock(lockKey);
             }
         } else{
             return value;
diff --git a/src/main/java/code/distribution/lock/DistributedLock.java b/src/main/java/code/distribution/lock/DistributedLock.java
@@ -12,15 +12,14 @@ public interface DistributedLock {
     /**
      * 获取锁
      * @param key   业务Key
-     * @return      锁序号
+     * @return
      */
-    String getLock(String key);
+    void lock(String key);
 
     /**
      * 解锁
      * @param key     业务Key
-     * @param lockSeq 锁序号
      * @return
      */
-    boolean unlock(String key, String lockSeq);
+    void unlock(String key);
 }
diff --git a/src/main/java/code/distribution/lock/RedisLock.java b/src/main/java/code/distribution/lock/RedisLock.java
@@ -0,0 +1,151 @@
+package code.distribution.lock;
+
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * 〈Redis锁〉<p>
+ * 〈功能详细描述〉
+ *
+ * @author zixiao
+ * @date 2019/8/26
+ */
+public class RedisLock implements DistributedLock {
+
+    private RedisClient redisClient = new RedisClient();
+
+    @Override
+    public void lock(String key) {
+        long expired = 3000;
+        long threadId = Thread.currentThread().getId();
+        Long ttl = tryAcquire(key, threadId, expired);
+        if(ttl == null){
+            return;
+        }
+
+        while (true){
+            ttl = tryAcquire(key, threadId, expired);
+            if(ttl == null){
+                return;
+            }
+        }
+    }
+
+    private Long tryAcquire(String key, long threadId, long expired){
+        return redisClient.tryLockInner(key, String.valueOf(threadId), expired);
+    }
+
+    @Override
+    public void unlock(String key) {
+        tryRelease(key, Thread.currentThread().getId());
+    }
+
+    private void tryRelease(String key, long threadId){
+        redisClient.tryUnlockInner(key, String.valueOf(threadId));
+    }
+
+    class RedisClient{
+
+        private Map<String/*lockKey*/, Map<String/*lockOwnerId*/, AtomicInteger>> lockHolder = new HashMap<>();
+
+        private Map<String, ExpiredObject> expiredMap = new HashMap<>();
+
+        public Long tryLockInner(String lockKey, String threadId, long expiredMills){
+            String threadUk = UUID.randomUUID().toString() + threadId;
+            //lua保证张一个事务里面
+            synchronized (lockHolder){
+                if(!exist(lockKey)){
+                    hset(lockKey, threadUk, 1);
+                    pexpire(lockKey, expiredMills);
+                    System.out.println("Lock success, I'm the first guy.");
+                    return null;
+                }
+                else if(hexist(lockKey, threadUk)){
+                    int locks = hincrby(lockKey, threadUk, 1);
+                    pexpire(lockKey, expiredMills);
+                    System.out.println("Locked by me(reentrant), locks="+locks);
+                    return null;
+                }
+                System.out.println("Locked by another guy.");
+                return pttl(lockKey);
+            }
+        }
+
+        public void tryUnlockInner(String lockKey, String threadId){
+            String threadUk = UUID.randomUUID().toString() + threadId;
+            if(!exist(lockKey)){
+                System.out.println("no lock existed.");
+                return;
+            }
+            else if(hexist(lockKey, threadUk)){
+                int locks = hdecrby(lockKey, threadUk, 1);
+                System.out.println("Unlock by me(reentrant), remain locks="+locks);
+                return;
+            }
+        }
+
+        public boolean exist(String key){
+            return lockHolder.containsKey(key) ;
+        }
+
+        public void hset(String key, String subKey, int value){
+            Map<String, AtomicInteger> map = new HashMap<>();
+            map.put(subKey, new AtomicInteger(Integer.valueOf(value)));
+            lockHolder.put(key, map);
+        }
+
+        /**
+         * ms为单位
+         */
+        public void pexpire(String key, long expireMills){
+            //expire
+            expiredMap.put(key, new ExpiredObject(expireMills));
+        }
+
+        public boolean hexist(String key, String subKey){
+            Map<String, AtomicInteger> map = lockHolder.get(key);
+            if( map != null && map.containsKey(subKey)){
+                return true;
+            }
+            return false;
+        }
+
+        public int hincrby(String key, String subKey, int delta){
+            Map<String, AtomicInteger> map = lockHolder.get(key);
+            return map.get(subKey).addAndGet(delta);
+        }
+
+        public int hdecrby(String key, String subKey, int delta){
+            Map<String, AtomicInteger> map = lockHolder.get(key);
+            return map.get(subKey).addAndGet(-1 * delta);
+        }
+
+        public Long pttl(String key){
+            ExpiredObject expiredObject = expiredMap.get(key);
+            long esapled = System.currentTimeMillis() - expiredObject.start;
+            if(esapled >= expiredObject.getExpiredMills()){
+                return null;
+            }else{
+                return expiredObject.getExpiredMills() - esapled;
+            }
+        }
+    }
+
+
+    @Data
+    class ExpiredObject{
+
+        private long start;
+
+        private long expiredMills;
+
+        public ExpiredObject(long expiredMills) {
+            this.start = System.currentTimeMillis();
+            this.expiredMills = expiredMills;
+        }
+    }
+}
diff --git a/src/main/java/code/distribution/lock/ZooKeeperLock.java b/src/main/java/code/distribution/lock/ZooKeeperLock.java
@@ -41,8 +41,10 @@ public ZooKeeperLock(){
         zkClient = new ZkClient("zk.tbj.com:2181", 6000, 6000);
     }
 
+    private ThreadLocal<String> lockSeqHolder = new ThreadLocal<>();
+
     @Override
-    public String getLock(String key) {
+    public void lock(String key) {
         String lockPath = ROOT + SEP +  key;
         //1、新建已key作为节点名的锁lockPath
         if(!zkClient.exists(lockPath)){
@@ -60,7 +62,7 @@ public String getLock(String key) {
         //3、尝试获取锁
         tryGetLock(lockPath, myLockSeq);
 
-        return myLockSeq;
+        lockSeqHolder.set(myLockSeq);
     }
 
     /**
@@ -79,6 +81,7 @@ public int compare(String o1, String o2) {
         });
         // a、如果是序号最小的子节点, 则认为获得锁
         if(nodeSeqList.size() == 0 || nodeSeqList.size() == 1 || myLockSeq.equals(nodeSeqList.get(0))){
+            System.out.println(">>>Get lock on node " + myLockSeq);
             return;
         }
 
@@ -121,9 +124,13 @@ public void handleDataDeleted(String dataPath) throws Exception {
     }
 
     @Override
-    public boolean unlock(String key, String lockSeq) {
+    public void unlock(String key) {
+        String lockSeq = lockSeqHolder.get();
         String lockFullPath = ROOT + SEP +  key + SEP +lockSeq;
-        return zkClient.delete(lockFullPath);
+        if(zkClient.delete(lockFullPath)){
+            lockSeqHolder.remove();
+        }
+        System.out.println(">>>Release lock on node " + lockSeq);
     }
 
     public static void main(String[] args) {
@@ -134,17 +141,15 @@ public static void main(String[] args) {
             executorService.execute(new Runnable() {
                 @Override
                 public void run() {
-                    DistributedLock dsLock = new ZooKeeperLock();
+                    DistributedLock lock = new ZooKeeperLock();
                     String bizKey = "loanId-5";
-                    String lockSeq = dsLock.getLock(bizKey);
+                    lock.lock(bizKey);
                     try {
-                        System.out.println(Thread.currentThread().getName() + " get lock on " + lockSeq);
                         doBiz(bizKey);
                     } catch (Exception e) {
                         e.printStackTrace();
                     } finally {
-                        System.out.println(Thread.currentThread().getName() + " release on " + lockSeq);
-                        dsLock.unlock(bizKey, lockSeq);
+                        lock.unlock(bizKey);
                     }
                 }
             });
diff --git a/src/main/java/code/http/HttpRequestConfig.java b/src/main/java/code/http/HttpRequestConfig.java
@@ -23,21 +23,21 @@ public class HttpRequestConfig {
 
     /**
      * 从连接池中获取可用连接超时时间
-     * 尝试从连接池中获取，若是在等待了一定的时间后还没有获取到可用连接，则会抛出获取连接超时异常。
+     * 尝试从连接池中获取，若是在等待了一定的时间后还没有获取到可用连接，则会抛出获取 ConnectionPoolTimeoutException。
      */
     private int connectionRequestTimeout = TIMEOUT;
 
     /**
      * 连接目标超时时间
      * 指的是连接目标url的连接超时时间，即客服端发送请求到与目标url建立起连接的最大时间。
-     * 如果在该时间范围内还没有建立起连接，则就抛出ConnectionTimeOut异常。
+     * 如果在该时间范围内还没有建立起连接，则就抛出 ConnectTimeoutException。
      */
     private int connectTimeout = TIMEOUT;
 
     /**
      * 等待响应超时（读取数据超时）
      * 连接上一个url后，获取response的返回等待时间 ，即在与目标url建立连接后，等待放回response的最大时间，
-     * 在规定时间内没有返回响应的话就抛出SocketTimeout异常。
+     * 在规定时间内没有返回响应的话就抛出 SocketTimeoutException。
      */
     private int socketTimeout = TIMEOUT<<1;
 
