分布式可重入锁 · 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} ~~~