怎样实现redis缓存与数据库同步？Redis如何实现分布式阻塞队列？

谁借莪１个温暖的怀抱￠ 2023-10-09 20:09 23阅读 0赞

**怎样实现redis缓存与数据库同步？**

（1）问题分析

考官主要考察面试者对于项目中缓存使用的能力

（2）核心答案讲解

答1:  
我们会先去redis中判断数据是否存在，如果存在，则直接返回缓存好的数据。而如果不存在的话，就会去数据库中，读取数据，并把数据缓存到Redis中。适用场合：如果数据量比较大，但不是经常更新的情况(比如用户排行)  
答2:  
只要使用了缓存就涉及到缓存同步的问题。缓存同步其实就是当缓存的信息发生变化，也就是对后台对缓存的数据进行增、删、改操作后，数据库中的数据发生了变化同时要把缓存中的数据对应删除即可。当页面再次请求数据时，缓存中不能命中就会从数据库中查询并且添加到缓存中，即实现了缓存同步。

（3）问题扩展

回顾redis雪崩和redis穿透

（4）结合项目中使用

情景一：广告数据

情景二：做搜索的分类对应的品牌数据、规格数据

**Redis如何实现分布式阻塞队列？**

**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 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那么可以直接拿到锁；但是影响不大因为拿到锁之后其他被阻塞的线程被唤醒之后将会继续阻塞。

**Java最新课程：**

[**Java零基础视频教程(2022最新Java入门，含斯坦福大学练习题+力扣算法题**][Java_2022_Java]

**Java基础入门：**

[**java零基础自学首Java入门教程（含Java项目和Java真题）**][java_Java_Java_Java]

**Javaweb核心基础**

[**JavaWeb基础教程，Java web从入门到企业实战完整版**][JavaWeb_Java web]

**Spring Cloud最全微服务架构**：

[**史上最全面的springcloud微服务技术栈**][springcloud]

**SSM框架教程：**

[**SSM框架教程\_Spring+SpringMVC+Maven高级+Spring**][SSM_Spring_SpringMVC_Maven_Spring]

[Java_2022_Java]: https://www.bilibili.com/video/BV17F411T7Ao?spm_id_from=333.999.0.0
[java_Java_Java_Java]: https://www.bilibili.com/video/BV1Cv411372m?spm_id_from=333.999.0.0
[JavaWeb_Java web]: https://www.bilibili.com/video/BV1Qf4y1T7Hx?spm_id_from=333.999.0.0
[springcloud]: https://www.bilibili.com/video/BV1LQ4y127n4?spm_id_from=333.999.0.0
[SSM_Spring_SpringMVC_Maven_Spring]: https://www.bilibili.com/video/BV1Fi4y1S7ix?spm_id_from=333.999.0.0