首页 > 数据库 >Redis实现分布式阻塞队列

Redis实现分布式阻塞队列

时间:2023-01-28 21:33:32浏览次数:46  
标签:队列 lock Redis jedis RedisLock key Jedis public 分布式


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);
}
}
123456789101112

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();
}
}
}
}


1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980

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());
}
}
1234567891011121314151617181920212223
  • 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");
}
}

123456789101112131415161718192021

注意

如果细心的小伙伴儿可能已经发现了unlock其实不是一个原子操作,可能在未发布消息但删除key之后的这段时间如果有人此时执行lock那么可以直接拿到锁;但是影响不大因为拿到锁之后其他被阻塞的线程被唤醒之后将会继续阻塞
此处unlock中有两个操作,删除key和发送消息如果在这两个操作之间机器异常并没有新的线程抢占锁那么此时被阻塞的线程将永远阻塞

标签:队列,lock,Redis,jedis,RedisLock,key,Jedis,public,分布式
From: https://blog.51cto.com/u_8238263/6025161

相关文章

  • 队列
    队列,具有先进先出特点,只允许在一端进行插入操作,在另一端进行删除。 代码实现#-*-coding=utf-8-*-#@Author:Wchime#@time:2023/1/2313:50#......
  • 双端队列
    双端队列是队列的扩展,可以在队列两端进行插入和删除。 代码实现#-*-coding=utf-8-*-#@Author:Wchime#@time:2023/1/2313:59#@file:双端......
  • Redis的同步机制
    Redis可以使用主从同步,从从同步。第一次同步时,主节点做一次bgsave,并同时将后续修改操作记录到内存buffer,待完成后将RDB文件全量同步到复制节点,复制节点接受完成后将RDB镜像......
  • 【分布式】浅谈CAP、BASE理论(1)
    【分布式】浅谈CAP、BASE理论(1)大家好,我们的gzh是朝阳三只大明白,满满全是干货,分享近期的学习知识以及个人总结(包括读研和IT),跪求一波关注,希望和大家一起努力、进步!!CAP理论......
  • Redis 缓存穿透,缓存击穿,缓存雪崩
    缓存击穿缓存击穿指的是缓存中没有数据但数据库中有数据(一般是热点数据缓存时间到期),同一时间大量的并发请求由于读缓存没读到数据,就去数据库去取数据,导致某个时间内数据库......
  • c#的Redis帮助类
    ///<summary>///Redis帮助类///</summary>publicclassRedisHelper:IDisposable{privatestaticreadonlyobjectLocker=newobj......
  • ReentrantLock中的阻塞队列与唤醒机制
    阻塞的状态不是被创建后就会进入阻塞形态的    即进入无限期等待,即使其他线程调用了interrupt方法也无法将其唤醒,除非有其他线程释放了锁,并且该线程拿到了锁,才会......
  • minio分布式存储的go语言开发衔接
    minio是分布式存储,可集群部署,阵列磁盘,纠错码等大数据存储必备的技术。由于它是go语言开发的,我们用go来与它衔接:上传文件,比如图片,然后预览。这里涉及几个重要的知识点。一......
  • redis集群之间数据迁移工具redis-migrate-tool测试
    1、学习背景,四川广电redis分布式缓存存在数据迁移,因此需要迁移对应的数据到新建集群;2、redis-migrate-tool特性1)快速,多线程2)支持异构迁移,支持twemproxy集群,rediscluste......
  • 代码随想录算法训练营第十天 | 理论基础,232.用栈实现队列,225. 用队列实现栈
    一、参考资料理论基础文章讲解:https://programmercarl.com/%E6%A0%88%E4%B8%8E%E9%98%9F%E5%88%97%E7%90%86%E8%AE%BA%E5%9F%BA%E7%A1%80.html用栈实现队列题目链......