Redis+Caffeine多级缓存数据一致性解决方案_Redis

问题分析

通过redis+caffeine，似乎可以完成一级、二级缓存中数据的同步，如果在单节点项目中是没有问题的，但是，在分布式场景下是有问题的，看下图：

说明：

部署了2个transport-info微服务节点，每个微服务都有自己进程级的一级缓存，都共享同一个redis作为二级缓存
假设，所有节点的一级和二级缓存都是空的，此时，用户通过节点1查询运单物流信息，在完成后，节点1的caffeine和redis中都会有数据
接着，系统通过节点2更新了数据，此时节点2中的caffeine和redis都是更新后的数据
用户还是进行查询动作，依然是通过节点1查询，此时查询到的将是旧的数据，也就是出现了一级缓存与二级缓存之间的数据不一致的问题

解决方案

如何解决该问题呢？可以通过消息的方式解决，就是任意一个节点数据更新了数据，发个消息出来，通知其他节点，其他节点接收到消息后，将自己caffeine中相应的数据删除即可。

关于消息的实现，可以采用rabbitmq，也可以采用redis的消息订阅发布来实现，在这里为了应用技术的多样化，所以采用redis的订阅发布来实现。

方案概述

redis 发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息

当有新消息通过 publish 命令发送给频道 channel1 时，这个消息就会被发送给订阅它的三个客户端。

redis的订阅发布功能与传统的消息中间件（如：rabbitmq）相比，相对轻量一些，针对数据准确和安全性要求没有那么高的场景可以直接使用。

代码实现

在redisconfig增加订阅的配置：

	/**
     * 配置订阅，用于解决caffeine一致性的问题
     *
     * @param connectionfactory 链接工厂
     * @param listeneradapter 消息监听器
     * @return 消息监听容器
     */
    @bean
    public redismessagelistenercontainer container(redisconnectionfactory connectionfactory,
                                                   messagelisteneradapter listeneradapter) {
        redismessagelistenercontainer container = new redismessagelistenercontainer();
        container.setconnectionfactory(connectionfactory);
        container.addmessagelistener(listeneradapter, new channeltopic(channel_topic));
        return container;
    }

编写redismessagelistener用于监听消息，删除caffeine中的数据。

/**
 * redis消息监听，解决caffeine一致性的问题
 */
@slf4j
@component
public class redismessagelistener extends messagelisteneradapter {

    @resource
    private cache<string, transportinfodto> transportinfocache;

    @override
    public void onmessage(message message, byte[] pattern) {
        // 获取到消息中的运单id
        string transportorderid = convert.tostr(message);
        log.info("redis消息监听缓存变更，运单id:{}", transportorderid);
        // 将本jvm中的缓存删除掉
        this.transportinfocache.invalidate(transportorderid);
    }
}

更新数据后向redis发送消息：

    @resource
    private stringredistemplate stringredistemplate;

    @override
    @cacheput(value = "transport-info", key = "#p0")
    public transportinfoentity saveorupdate(string transportorderid, transportinfodetail infodetail) {
		// 省略代码
 
        // 清除缓存中的数据
        // this.transportinfocache.invalidate(transportorderid);
        // caffeine本地缓存一致性，发布订阅消息到redis，通知订阅者更新缓存
        this.stringredistemplate.convertandsend(redisconfig.channel_topic, transportorderid);
        // 保存/更新到mongodb
        return this.mongotemplate.save(transportinfoentity);
    }