【图灵干货】Java高级教程第十四节：redis实现分布式阻塞队列
2021-11-24 13:47:54
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
*/
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架构师视频学习地址：免费视频
