【图灵干货】Java高级教程第十四节：redis实现分布式阻塞队列

1. Redis分布式锁实现原理

本质上，分布式锁的目标是在Redis中占据一个茅坑。当其他过程被占用时，如果你发现有人蹲在那里，你必须放弃或稍后再试。setnx(setifnotexists)通常用于占坑。
允许被客户端占用。先先占，用完，再调用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.当某个线程释放锁时，通过redispub/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

main1

main2

*/

@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

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那么可以直接拿到锁；但是影响不大因为拿到锁之后其他

被阻塞的线程被唤醒之后将会继续阻塞。

图灵学院成立于2017年7月15日，现阶段提供 计算机基础原理、JavaSE核心、Java后端、 面试必备算法、python核心编程、数据分析、web 开发题、人工智能等专题课程，为想学习Python的学员提供优质的培训服务，帮助学员掌握更加全面的技能，是计算机人员职场中提职加薪的首选。
免费java架构师视频学习地址：免费视频