redis 分布式锁 说明 如何实现可重入？ 首先锁信息(指redis中lockKey关联的value值) 必须得设计的能负载更多信息，之前non-reentrant时value直接就是一个超时时间，但是要实现可重入单超时时
如何实现可重入？
首先锁信息(指redis中lockKey关联的value值) 必须得设计的能负载更多信息，之前non-reentrant时value直接就是一个超时时间，但是要实现可重入单超时时间是不够的，必须要标识锁是被谁持有的，也就是说要标识分布式环境中的线程，还要记录锁被入了多少次。
如何在分布式线程中标识唯一线程？
MAC地址 + jvm进程ID + 线程ID（或者线程地址都行）,三者结合即可唯一分布式环境中的线程。
实现
锁的信息采用json存储，格式如下：
{
"count":1,
"expires":1475848548547,
"MAC":"2d-8f-ee-4f",
"jvmpid"222245,
"threadid":14
}
代码框架还是和之前实现的非重入的差不多，重点是lock方法，代码已有非常详细的注释
redis 代码
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), 以免网路问题而导致线程一直阻塞. * SETNC操作参考资料. * *可重入实现关键: *
-
*
- 在分布式环境中如何确定一个线程? mac地址 + jvm pid + threadId (mac地址唯一, jvm * pid在单机内唯一, threadId在单jvm内唯一) *
- 任何一个线程从redis拿到value值后都需要能确定 该锁是否被自己持有, 因此value值要有以下特性: 保存持有锁的主机(mac), jvm * pid, 持有锁的线程ID, 重复持有锁的次数 *
* redis中value设计如下(in json): *
* {
* 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}表示.
*
*
* Usage Example:
*
* {@link Lock} lock = new {@link ReentrantLock}(jedis, "lockKey", lockExpires, timeServerAddr);
* if (lock.tryLock(3, TimeUnit.SECONDS)) {
* try {
* // do something
* } catch (Exception e) {
* lock.unlock();
* }
* }
*
*
*
* @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();
}
}
redis 锁代码
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
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
createThreads(int count) throws IOException { List
threads = new ArrayList
(); 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(); } } } } } }
DistributedLock-master.zip
DistributedLock-master.zip