Redis订阅发布你会用了吗？

news2024/11/29 9:37:10

点击下方“JavaEdge”，选择“设为星标”

第一时间关注技术干货！

免责声明~

任何文章不要过度深思！

万事万物都经不起审视，因为世上没有同样的成长环境，也没有同样的认知水平，更「没有适用于所有人的解决方案」；

不要急着评判文章列出的观点，只需代入其中，适度审视一番自己即可，能「跳脱出来从外人的角度看看现在的自己处在什么样的阶段」才不为俗人。

怎么想、怎么做，全在乎自己「不断实践中寻找适合自己的大道」

Redis实现多级缓存时的数据一致性保证

在实际应用中，为了提高数据访问的效率和减轻数据库的压力，通常会在系统中使用多级缓存策略。Redis作为一个常用的分布式缓存系统，可以通过多级缓存来保证数据的一致性。本文将介绍两种主要措施来保证本地缓存和Redis缓存数据的一致性：

主动措施：通过Redis的订阅发布模型来实现。当Redis中的数据发生变更时，所有订阅者会收到通知，从而更新本地缓存。
被动措施：本地缓存设置失效时间，当缓存失效后，主动从Redis中获取最新的数据。

本文主要通过代码示例展示如何在Spring Boot项目中实现Redis的订阅发布机制和本地缓存的使用。

1. Redis的订阅发布

1.1 添加依赖

首先，在项目的pom.xml文件中添加Spring Boot Data Redis和Guava的依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!--guava依赖-->
<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>27.0.1-jre</version>
</dependency>

1.2 订阅者配置

Redis使用list结构实现订阅发布。订阅者和发布者可以在不同的项目中，但必须使用同一个Redis实例。

下图展示了Redis的订阅发布模型：

订阅发布模型

在代码中使用了Guava的Cache来实现本地缓存，并设置了失效时间。当监听到Redis修改时，主动修改本地缓存。同时，缓存失效后也会去Redis中获取数据并更新本地缓存。

package com.javagpt.back.interceptor;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.apache.commons.lang3.StringUtils;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.HandlerInterceptor;

import javax.annotation.Resource;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Component
public class FuseInterceptor implements HandlerInterceptor {

    @Resource
    private StringRedisTemplate stringRedisTemplate;

    // 设置超时时间的Guava Cache
    public static Cache<String, Object> moduleCache = CacheBuilder.newBuilder()
            // 写入缓存后，10秒后过期
            .expireAfterWrite(10, TimeUnit.SECONDS)
            .build();

    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws ExecutionException {
        // 先在本地缓存中获取内容，若获取不到，直接查询Redis并放入本地缓存（线程安全）
        String moduleStatus = (String) moduleCache.get("topic:service_xx:module", () -> {
            // 若本地缓存获取不到，那么去Redis查询数据并放入缓存。
            String value = stringRedisTemplate.opsForValue().get("service_xx:module");
            return StringUtils.isBlank(value) ? "close" : value;
        });
        // 拦截器处理
        return true;
    }

    /**
     * 订阅发布获取到最新的Redis内容
     *
     * @param type    监听队列的名
     * @param message 监听到的信息
     */
    public void refreshCache(String type, String message) {
        if ("topic:service_xx".equals(type)) {
            // 若是这个队列的消息，那么将其转换为Set<String>结构，并放入缓存中
            Set<String> patterns = JSONObject.parseObject(message,
                    new TypeReference<LinkedHashSet<String>>() {}.getType());
            moduleCache.put(type, patterns);
        } else {
            moduleCache.put(type, message);
        }
    }
}

1.3 消息发布者配置

在需要发布消息的地方，使用以下代码进行消息发布：

// 发布消息
stringRedisTemplate.convertAndSend("topic:service_xx:module","发送的消息");
// 将消息存储到Redis中，以便本地缓存失效后查询Redis缓存
stringRedisTemplate.opsForValue().set("service_xx:module","发送的消息");

通过上述代码配置，当Redis中的数据发生变更时，订阅者可以及时收到通知并更新本地缓存，从而保证数据的一致性。同时，当本地缓存失效后，会主动查询Redis获取最新数据，以保证数据的一致性。