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 解决思路
- 首先咱们革新lock锁，当不能创立key时就利用以后key阻塞以后线程
- 当某一个线程开释锁时通过redis的pub/sub发送一个音讯音讯内容为key
- 所有应用锁的利用监听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和发送音讯如果在这两个操作之间机器异样并没有新的线程抢占锁那么此时被阻塞的线程将永远阻塞
发表回复