Redis 双写一致性问题探究与解决方案，附 Java 代码实现

快来打我* 2023-01-02 06:30 195阅读 0赞

### Redis 双写一致性问题探究与解决方案，附 Java 代码实现 ###

*  一、前言
 *  二、什么是 Redis 双写一致？
 *  三、什么情况下缓存与数据库不一致？
 *   *  3.1 写操作
     *   *  3.1.1 先更新缓存再更新数据库
         *  3.1.2 先更新数据库再更新缓存
     *  3.2 删操作
     *  3.3 查询数据
 *  四、解决方案
 *   *  4.1 原理
     *  4.2 代码实现
     *   *  4.2.1 定义函数式接口
         *  4.2.2 初始化线程池
         *  4.2.2 删除缓存
         *  4.2.3 获取缓存数据

# 一、前言 #

本文中，只使用到 Mysql 与 Redis，依然还存在优化空间，比如结合消息队列等

# 二、什么是 Redis 双写一致？ #

或许大家都遇到过这样的情况

举例：比如用户数据更新了，往往我们会做以下操作

1.  更新数据库
2.  更新 Redis 缓存

通过这种方式来保证从缓存中读取的数据的正确性

这种缓存的数据与数据库保持一致即双写一致性

# 三、什么情况下缓存与数据库不一致？ #

## 3.1 写操作 ##

### 3.1.1 先更新缓存再更新数据库 ###

众所周知，数据库重要性是要在缓存之上的，缓存只是为了提高性能。

假如先更新缓存，但是在操作数据库失败

此时缓存是新值，数据库是旧值，就会出现双写不一致

数据库失败的概率远比更新缓存失败概率大：SQL 错误，连接问题，死锁等

因此这种操作本身就是不合理的

### 3.1.2 先更新数据库再更新缓存 ###

这种最为常规的操作在并发情况下，也会出现双写不一致的问题  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MTQzNjcz_size_16_color_FFFFFF_t_70_pic_center]

## 3.2 删操作 ##

简单举例，假如在删除数据库后，删除缓存时，正好遇到 Redis 节点挂了导致失败，这种事谁说的准，假如碰上了呢

## 3.3 查询数据 ##

这种很常见，特别是在小公司，小系统上。很多人从缓存中获取数据就是简单获取，完全不考虑获取不到的情况，大概是懒吧。

假如本应有的数据拿不到了，怎么办？

为什么会出现这种情况？大概是想，要是每次读缓存时候都要去判断下能不能拿到，拿不到去数据库拿再去更新缓存，那也太麻烦了。。。。。

# 四、解决方案 #

## 4.1 原理 ##

**采用双删策略：**  
**写操作：**

1.  删除缓存  
    PS：先删除，保证后续的查询都会先从数据库查，此时并发情况不管你数据库更没更新，我都是从数据库拿到的正确数据
2.  更新数据库
3.  再次删除缓存  
    PS：第二次删除前我们会让线程先睡眠一段时间，假如存在并发情况又对缓存被回写了，此时 1S 的睡眠能保证 1S 内脏数据被删除（3.1.2 中情况）

## 4.2 代码实现 ##

### 4.2.1 定义函数式接口 ###

**OperatorInterface：**

/** * 数据库操作方法 */
    @FunctionalInterface
    public interface OperatorInterface<T> { 
        T apply();
    }

### 4.2.2 初始化线程池 ###

线程池使用的是 hutool 线程池，也可以自己写

<dependency>
        <groupId>cn.hutool</groupId>
        <artifactId>hutool-all</artifactId>
        <version>5.3.10</version>
    </dependency>

/** * 线程池配置 * * @Author: linjinp * @Date: 2020/9/29 10:54 */
    @Slf4j
    @Component
    public class ExecutorConfig { 
    
        // 初始线程数量
        private final static int DEFAULT_NUM = 5;
    
        // 最大线程数
        private final static int MAX_NUM = 10;
    
        // 最大等待线程数
        private final static int MAX_WAITING = 100;
    
        @Bean
        public ExecutorService createExecutor() { 
            ExecutorService executor = ExecutorBuilder.create()
                    // 默认初始化 5 个线程
                    .setCorePoolSize(DEFAULT_NUM)
                    // 最大线程数 10
                    .setMaxPoolSize(MAX_NUM)
                    // 最大等待线程数 100
                    .setWorkQueue(new LinkedBlockingQueue<>(MAX_WAITING))
                    .build();
            log.info("\n初始化线程池\n默认初始线程数：{}\n最大线程数：{}\n最大等待线程数：{}", DEFAULT_NUM, MAX_NUM, MAX_WAITING);
            return executor;
        }
    }

### 4.2.2 删除缓存 ###

不管你是更新数据库数据还是删除数据库数据，缓存都应该删除

二次删除目的是为了保证正确性，因此使用多线程进行删除，应用系统就无须等待二次删除中的睡眠等操作，提高效率

**RedisRemoveFunction：**

/** * Redis 双删策略 * @author: linjinp * @create: 2020-06-18 11:19 **/
    @Slf4j
    @Component
    public class RedisRemoveFunction { 
    
        @Resource
        private RedisTemplate redisTemplate;
    
        @Resource
        private ExecutorService executorService;
    
        // 延迟删除 1s
        private final static long REMOVE_DELAY = 1000;
    
        // 失败重试次数
        private final static int FAIL_RETRY = 5;
    
        /** * 删除 Redis 缓存 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @return */
        public void remove(String key) { 
            remove(key, null);
        }
    
        /** * 删除 Redis 缓存，并进行数据库操作 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @param operatorInterface 接口函数（数据库更新） * @param <T> 返回数据类型 * @return */
        public <T> T remove(String key, OperatorInterface<T> operatorInterface) { 
            // 第一次删除，清除缓存
            redisTemplate.delete(key);
            T data = null;
            // 为空说明没进行数据库操作
            if (operatorInterface != null) { 
                data = operatorInterface.apply();
            }
    
            // 创建线程进行二次删除，清除 1S 内的脏数据
            executorService.execute(() -> { 
                // 失败重试
                int failRetry= FAIL_RETRY;
                while (failRetry-- > 0) { 
                    try { 
                        Thread.sleep(REMOVE_DELAY);
                        redisTemplate.delete(key);
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            });
            return data;
        }
    }

### 4.2.3 获取缓存数据 ###

逻辑：

1.  判断 Redis 中是否能找到 Key
2.  如果能找到，判断是否是空标识，是的话返回 null，不是则返回数据
3.  如果找不到 Key，调用接口函数获取数据
4.  如果查到数据，保存数据到缓存并返回数据
5.  如果没找到数据，保存空值标识，设置过期，返回 null

**RedisValueFunction：**

/** * Redis 获取数据 * @author: linjinp * @create: 2020-06-18 11:19 **/
    @Slf4j
    @Component
    public class RedisValueFunction { 
    
        @Resource
        private RedisTemplate redisTemplate;
    
        // 默认空值超时时间，5S
        private final static long DEFAULT_EMPTY_TIMEOUT = 5;
    
        // 默认数据超时时间，30分钟
        private final static long DEFAULT_DATA_TIMEOUT = 30;
    
        /** * 获取 Redis 参数，数据不会设置过期时间 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @param operatorInterface 接口函数（数据库查询） * @param <T> 返回数据类型 * @return */
        public <T> T getNotTimeout(String key, OperatorInterface<T> operatorInterface) { 
            return get(key, 0, null, operatorInterface);
        }
    
        /** * 获取 Redis 参数 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @param operatorInterface 接口函数（数据库查询） * @param <T> 返回数据类型 * @return */
        public <T> T get(String key, OperatorInterface<T> operatorInterface) { 
            return get(key, DEFAULT_DATA_TIMEOUT, operatorInterface);
        }
    
        /** * 获取 Redis 参数 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @param timeout 超时时间，单位分钟，当值 <= 0 时表示永不过期 * @param operatorInterface 接口函数（数据库查询） * @param <T> 返回数据类型 * @return */
        public <T> T get(String key, long timeout, OperatorInterface<T> operatorInterface) { 
                return get(key, timeout, TimeUnit.MINUTES, operatorInterface);
        }
    
        /** * 获取 Redis 参数 * 如果未获取到数据，从数据库中获取 * 如果数据库中也获取不到，则设置空值标识 * @param key Redis Key * @param timeout 超时时间，当值 <= 0 时表示永不过期 * @param timeUnit 超时时间单位 * @param operatorInterface 接口函数（数据库查询） * @param <T> 返回数据类型 * @return */
        public <T> T get(String key, long timeout, TimeUnit timeUnit, OperatorInterface<T> operatorInterface) { 
            if (redisTemplate.hasKey(key)) { 
                Object obj = redisTemplate.opsForValue().get(key);
                if (obj != null && obj instanceof String && "EmptyMark".equals(obj)) { 
                    log.info("{}：EmptyMark", key);
                    return null;
                }
                return (T) obj;
            } else { 
                // 调用数据库查询接口函数
                T data = operatorInterface.apply();
                if (data == null) { 
                    // 设置空标识
                    redisTemplate.opsForValue().set(key, "EmptyMark", DEFAULT_EMPTY_TIMEOUT, TimeUnit.SECONDS);
                    log.info("{}：EmptyMark", key);
                    return null;
                }
                if (timeout <= 0) { 
                    redisTemplate.opsForValue().set(key, data);
                } else { 
                    redisTemplate.opsForValue().set(key, data, timeout, timeUnit);
                }
                return data;
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3MTQzNjcz_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/af9445774fd34b319704d2596268aa77.png