Redis如何实现分布式阻塞队列？

Posted 2021-10-11 妙蛙种子

1. Redis分布式锁实现原理

分布式锁本质上要实现的目标就是在 Redis 里面占一个“茅坑”，当别的进程也要来占时，发现已经有人蹲在那里了，就只好放弃或者稍后再试。占坑一般是使用 setnx(set if not exists) 指令，只允许被一个客户端占坑。先来先占， 用完了，再调用 del 指令释放茅坑。
死锁问题：如果逻辑执行到中间出现异常了，可能会导致 del 指令没有被调用，这样就会陷入死锁，锁永远得不到释放， 解决这个问题我们在拿到锁之后，再给锁加上一个过期时间，比如 5s，这样即使中间出现异常也可以保证 5 秒之后锁会自动释放

2. 普通非阻塞锁实现

public class RedisLock {
    private Jedis jedis;
    public RedisLock(Jedis jedis) {
        this.jedis = jedis;
    }
    public boolean lock(String key) {
    return jedis.set(key, "", "nx", "ex", 5L) != null;
    }
    public void unlock(String key) {
        jedis.del(key);
    }
}

2.1 存在问题

• 如果某一个进程没有拿到锁得到了false的结果那么次进程是否执行当前任务？显然对于一般情况来说我们的任务都是必须执行的那么此时我们就要考虑该何时执行了，在传统的锁中我们如果没有拿到锁线程就进入了阻塞状态那么此处我们是否可以改进同样实现阻塞唤醒机制

3. 分布式阻塞锁具体实现

3.1 解决思路

1. 首先我们改造lock锁，当不能创建key时就利用当前key阻塞当前线程
2. 当某一个线程释放锁时通过redis的pub/sub发送一个消息消息内容为key
3. 所有使用锁的应用监听lock通道的消息，在收到消息时通过key唤醒对应线程

3.2具体实现

package com.hgy.common.redis;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPubSub;

import java.util.HashMap;

public class RedisLock extends JedisPubSub {

    //是否已经初始化监听
    private static volatile boolean isListen = false;
    //每一个redis的key对应一个阻塞对象
    private  HashMap<String, Object> blockers = new HashMap<>();

    private Jedis jedis;

    //当前获得锁的线程
    private Thread curThread;



    public RedisLock(Jedis jedis) {
        this.jedis = jedis;

        //保证没一个应用只初始化一次监听
        if (!isListen) {
            synchronized (RedisLock.class) {
                if (!isListen) {
                    // 启动一个线程做消息监听
                    new Thread(()->{
                        new Jedis("192.168.200.128", 6379).subscribe(this, "lock");
                    }).start();
                    isListen = true;
                }
            }
        }
    }


    public void lock(String key) throws InterruptedException {
        //循环判断是否能够创建key， 不能则直接wait释放CPU执行权
        while (jedis.set(key, "", "nx", "ex", 20L) == null) {
            synchronized (key) {
                System.out.println(Thread.currentThread().getName() + "=======" + key);
                blockers.put(key, key);
                key.wait();
            }
        }
        blockers.put(key, key);
        //能够成功创建，获取锁成功记录当前获取锁线程
        curThread = Thread.currentThread();
    }

    public void unlock(String key) {
        //判断是否为加锁的线程执行解锁， 不是则直接忽略
        if( curThread == Thread.currentThread()) {
            jedis.del(key);
            //删除key之后需要notifyAll所有的应用， 所以这里采用发订阅消息给所有的应用
            jedis.publish("lock", key);
        }
    }


    /**
     * 所有应用接收到消息后在当前应用中执行对应key的notifyAll方法
     * @param channel
     * @param message
     */
    @Override
    public void onMessage(String channel, String message) {
        Object lock = blockers.get(message);
        if(lock != null) {
            synchronized (lock) {
                lock.notifyAll();
            }
        }
    }
}

4. 测试

目标： 开启两个mian线程， 在第一个中首先暂停3秒然后打印1-100然后线程休眠5秒释放锁并打印最后的毫秒数； main1在执行的同时执行main2，在2中打印开始时间；最后比对1和2的开始时间即可验证

注意： 先启动1然后启动2

• main1

package com.hgy;

import com.hgy.common.redis.RedisLock;
import redis.clients.jedis.Jedis;

public class RedisLockApp1 {

    private static RedisLock redisLock;

    public static void main(String[] args) throws InterruptedException {
        Jedis client = new Jedis("192.168.200.128", 6379);
        redisLock = new RedisLock(client);

        redisLock.lock("demo");
        Thread.sleep(3000);
        for (int i = 0; i < 100; i++) {
            System.out.println("app1" + i);
        }
        Thread.sleep(5000);
        redisLock.unlock("demo");
        System.out.println("App1==> end:" + System.currentTimeMillis());
    }
}

• main2

package com.hgy;

import com.hgy.common.redis.RedisLock;
import redis.clients.jedis.Jedis;

public class RedisLockApp2 {
    private static RedisLock redisLock;

    public static void main(String[] args) throws InterruptedException {
        Jedis client = new Jedis("192.168.200.128", 6379);
        redisLock = new RedisLock(client);

        redisLock.lock("demo");
        System.out.println("App2==> start:" + System.currentTimeMillis());
        for (int i = 0; i < 100; i++) {
            System.out.println("app2" + i);
        }
        redisLock.unlock("demo");
    }
}

注意

如果细心的小伙伴儿可能已经发现了unlock其实不是一个原子操作，可能在未发布消息但删除key之后的这段时间如果有人此时执行lock那么可以直接拿到锁；但是影响不大因为拿到锁之后其他被阻塞的线程被唤醒之后将会继续阻塞

此处unlock中有两个操作，删除key和发送消息如果在这两个操作之间机器异常并没有新的线程抢占锁那么此时被阻塞的线程将永远阻塞