乐趣区

关于redis:Redis如何实现分布式阻塞队列

jiezi

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 和发送音讯如果在这两个操作之间机器异样并没有新的线程抢占锁那么此时被阻塞的线程将永远阻塞

退出移动版