单机Redis实现分布式互斥锁

代码地址如下：
http://www.demodashi.com/demo/12520.html

0、准备工作

0-1 运行环境

jdk1.8
gradle
一个能支持以上两者的代码编辑器，作者使用的是IDEA。

0-2 知识储备

对Java并发包，互斥锁有一定的理解。
对Redis数据库有一定了解。
本案例偏难，案例内有注释讲解，有不明白的地方，或者不正确的地方，欢迎联系作者本人或留言。

1、设计思路

1.1 项目结构

图2——项目结构

/lock/DestributeLock.java：锁的具体实现类，有lock()和unlock()两个方法。

/lock/DestributeLockRepository.java：锁的工厂类，用来配置Redis连接信息，获取锁的实例。

/lock/Expired**.java：Redis的 pub/sub 功能相关类，用来实现超时自动解锁。

/test.java：3个测试类，用来模拟不同的加锁情况。

1.2 实现难点

咱们可以在网上轻松的找到，用Redis实现简单的互斥锁的案例。

那为什么说是简单的？因为不安全：

1.Redis的stNX()与expire()方法是两个独立的操作，即非原子性。咱们可以假设这么一个情况，当你执行stNX()之后，服务器挂了，没有执行expire()方法。那如果没有去解锁的话，是不是就死锁了？所以，咱们需要保证这两个操作的原子性。

2.expire()方法只是告知Redis在一定时间后，自动删除某个键。但是，服务器并不知道expire()在超时之后，是否成功地解锁（删除了key）。所以，咱们需要Redis通知服务器expire()方法已经彻底执行完毕，即Redis已经删除了key,才能确定为解锁状态。

2、具体实现

2.1 DestributeLockRepository.java

public class DistributeLockRepository {

    private String host;

    private int port;

    private int maxTotal;

    private JedisPool jedisPool;

    /**

     * @param host redis地址

     * @param port 端口

     * @param maxTotal 锁的最大个数，也就是说最多有maxTotal个线程能同时操作锁

     *

     **/

    public DistributeLockRepository(String host,int port,int maxTotal){

        this.host = host;

        this.port = port;

        this.maxTotal = maxTotal;

        JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();

        jedisPoolConfig.setMaxTotal(maxTotal);

        jedisPool = new JedisPool(jedisPoolConfig, host, port);

    }

    public DistributeLock instance(String lockname) {

        Jedis jedis = jedisPool.getResource();

        // 若超过最大连接数，会在这里阻塞

        return new DistributeLock(jedis, lockname);

    }

}

2.2 DestributeLock.java

public class DistributeLock implements ExpiredListener {

    private Jedis redisClient = null;

    private String key = ""; //锁的key

    private int expire = 0; //Redis自动删除时间

    private long startTime = 0L; //尝试获取锁的开始时间

    private long lockTime = 0L; //获取到锁的时间

    private boolean lock = false; //锁状态

    private void setLock(boolean lock) {

        this.lock = lock;

    }

    private void closeClient() {

        redisClient.close();

    }

    private static String script =

            "if redis.call('setnx',KEYS[1],KEYS[2]) == 1 then\n"

                    + "redis.call('expire',KEYS[1],KEYS[3]);\n"

                    + "return 1;\n"

                    + "else\n"

                    + "return 0;\n"

                    + "end\n";

    DistributeLock(Jedis jedis, String key) {

        this.redisClient = jedis;

        this.key = key;

    }

    @Override

    public void onExpired() {

        ExpiredManager.remove(key,this);

        this.setLock(false);

        redisClient.close();//关闭连接

        redisClient = null;

        System.out.println(key +lockTime+ "Redis超时自动解锁" + Thread.currentThread().getName());

    }

    //redisClient.psubscribe(new ExpiredSub(this),"__key*__:expired");

    /**

     * @param timeout 锁阻塞超时的时间  单位：毫秒

     * @param expire  redis锁超时自动删除的时间   单位：秒

     * @return true-加锁成功  false-加锁失败

     */

    public synchronized boolean lock(long timeout, int expire) {

        this.expire = expire;

        this.startTime = System.currentTimeMillis();

        if (!lock) {

            //System.out.println(Thread.currentThread().getName() + lock);

            try {

                //在timeout的时间范围内不断轮询锁

                while (System.currentTimeMillis() - startTime < timeout) {

                    //System.out.println(Thread.currentThread().getName() + "inWhile");

                    //使用Lua脚本保证setnx与expire的原子性

                    Object object = redisClient.eval(script, 3, key, "a", String.valueOf(expire));

                    //System.out.println(Thread.currentThread().getName() + "afterScript");

                    if ((long) object == 1) {

                        this.lockTime = System.currentTimeMillis();

                        //锁的情况下锁过期后消失，不会造成永久阻塞

                        this.lock = true;

                        System.out.println(key+lockTime + "加锁成功" + Thread.currentThread().getName());

                        //交给超时管理器

                        ExpiredManager.add(key, this);

                        return this.lock;

                    }

                    System.out.println(key+lockTime +"出现锁等待" + Thread.currentThread().getName());

                    //短暂休眠，避免可能的活锁

                    Thread.sleep(500);

                }

                System.out.println(key+lockTime +"锁超时" + Thread.currentThread().getName());

            } catch (Exception e) {

                if(e instanceof NullPointerException){

                    throw new RuntimeException("无法对已经解锁后的锁重新加锁，请重新获取", e);

                }

                throw new RuntimeException("locking error", e);

            }

        } else {

            //System.out.println(key + "不可重入/用");

            throw new RuntimeException(key +lockTime+ "不可重入/用");

        }

        this.lock = false;

        return this.lock;

    }

    public synchronized void unlock() {

        if (this.lock) {

            //解决在 Redis自动删除锁后，尝试解锁的问题

            if (System.currentTimeMillis() - lockTime <= expire) {

                redisClient.del(key);//直接删除  如果没有key，也没关系，不会有异常

            }

            this.lock = false;

            redisClient.close();//关闭连接

            redisClient = null;

            System.out.println(key+ lockTime+ "解锁成功" + Thread.currentThread().getName());

        }else {

            System.out.println(key +lockTime+ "已经解锁" + Thread.currentThread().getName());

        }

    }

}

2.3 ExpiredManager.java

public class ExpiredManager {

    private static final String HOST = "localhost";

    private static final Integer PORT = 16379;

    private static boolean isStart = false;

    private static Jedis jedis;

    private static ConcurrentHashMap<String,CopyOnWriteArrayList<ExpiredListener>> locks = new ConcurrentHashMap<>();

    public static void add(String key,ExpiredListener listener){

        CopyOnWriteArrayList<ExpiredListener> copyOnWriteArrayList = locks.get(key);

        if(copyOnWriteArrayList==null){

            copyOnWriteArrayList = new CopyOnWriteArrayList<ExpiredListener>();

            copyOnWriteArrayList.add(listener);

            locks.put(key,copyOnWriteArrayList);

        }else {

            copyOnWriteArrayList.add(listener);

        }

    }

    public static void remove(String key,ExpiredListener listener){

        CopyOnWriteArrayList<ExpiredListener> copyOnWriteArrayList = locks.get(key);

        if(copyOnWriteArrayList!=null){

            copyOnWriteArrayList.remove(listener);

        }

    }

    public synchronized static void start(){

        if(!isStart) {

            isStart = true;

            jedis = new Jedis(HOST, PORT);

            new Thread(new Runnable() {

                @Override

                public void run() {

                    try {

                        jedis.psubscribe(new ExpiredSub(locks), "__key*__:expired");

                    }catch (Exception e){

                        System.out.println(e.getMessage());

                    }

                }

            }).start();

        }

    }

    public synchronized static void close(){

        if(isStart) {

            isStart = false;

            jedis.close();

        }

    }

}

2.4 测试

public class Test3 {

    public static void main(String[] args) {

        //第三个参数表示  同一时间 最多有多少锁能  处于加锁或者阻塞状态  其实就是连接池大小

        DistributeLockRepository distributeLockRepository = new DistributeLockRepository("localhost", 16379, 6);

        //获取锁实例

        DistributeLock lock1 = distributeLockRepository.instance("lock[A]");

        DistributeLock lock2 = distributeLockRepository.instance("lock[A]");

        //开启超时解锁管理器

        ExpiredManager.start();

        //lock1和lock2其实模拟的是两个消费者，对同一资源（lock[A]）的竞争使用

        lock1.lock(1000 * 20L, 5);

        try {

            Thread.sleep(6000);

        } catch (InterruptedException e) {

            e.printStackTrace();

        }

        lock2.lock(1000 * 20L, 5);

        lock1.unlock();

        System.out.println("----");

        //关闭超时解锁管理器

        ExpiredManager.close();

    }

}

Test3的运行结果：

3、总结

上面是贴出的主要代码，完整的请下载demo包，有不明白的地方请在下方评论，或者联系邮箱yaoyunxiaoli@163.com。

我是妖云小离,这是我第一次在Demo大师上发文章，感谢阅读。单机Redis实现分布式互斥锁