Java 实现基于Redis的分布式可重入锁

ABKYH · 发表于 2018-11-4 13:08:32

　　如何实现可重入？
　　首先锁信息(指redis中lockKey关联的value值) 必须得设计的能负载更多信息，之前non-reentrant时value直接就是一个超时时间，但是要实现可重入单超时时间是不够的，必须要标识锁是被谁持有的，也就是说要标识分布式环境中的线程，还要记录锁被入了多少次。
　　如何在分布式线程中标识唯一线程？
　　MAC地址 +jvm进程 + 线程ID（或者线程地址都行）,三者结合即可唯一分布式环境中的线程。下载
　　实现
　　锁的信息采用json存储，格式如下：

　　代码框架还是和之前实现的非重入的差不多，重点是lock方法，代码已有非常详细的注释
　　Java代码下载

　　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();
　　}
　　}

　　测试
　　5个线程，每个线程都是不同的jedis连接，模拟分布式环境，线程的任务就是不断的去尝试重入地获取锁，重入的次数为随机但在0-5之间。
　　代码
　　Java代码下载

　　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

账号		自动登录	找回密码
密码			立即注册

wirelessnetview好用的无线分析工具

Red Hat RHCE 8 (EX294) Cert Guide

Shell从入门到精通（阿良）

亿图图示专家(EDraw Max) V7.9 中文破解版

zabbix3.4.1安装部署+微信推送信息+大屏显

Red Hat OpenShift I: Containers & Kubern

2025 年，C++ 还能“硬核”多久？

[经验分享] Java 实现基于Redis的分布式可重入锁

浏览过的版块

扫码加入运维网微信交流群