内容转载自JavaGuide 介绍 简单介绍一下 Redis 简单来说 Redis 就是一个使用C 语言开发的数据库，不过与传统数据库不同的是 Redis 的数据是存在内存中的 ，也就是它是内存数据库，所以读写速度非常快，因此 Redis 被广泛应用于缓存方向。 另外，Redis 除了做缓存之外，也经常用来做分布式锁，甚至是消息队列。 Redis 提供了多种数据类型来支持不同的业务场景。Redis 还支持事务、持久化、Lua 脚本、多种集群方案。 分布式缓存方案 分布式缓存的话，使用比较多的主要是 Memcached 和 Redis 共同点： 都是基于内存的数据库，一般都用来当做缓存使用。 都有过期策略。 两者的性能都非常高。 区别： Redis 支持更丰富的数据类型（支持更复杂的应用场景）。Redis 不仅仅支持简单的 k/v 类型的数据，同时还提供 list，set，zset，hash 等数据结构的存储。Memcached 只支持最简单的 k/v 数据类型。 Redis 支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用,而 Memecache 把数据全部存在内存之中。 Redis 有灾难恢复机制。 因为可以把缓存中的数据持久化到磁盘上。 Redis 在服务器内存使用完之后，可以将不用的数据放到磁盘上。但是，Memcached 在服务器内存使用完之后，就会直接报异常。 Redis 目前是原生支持 cluster 模式。Memcached 没有原生的集群模式，需要依靠客户端来实现往集群中分片写入数据 Redis 使用单线程的多路 IO 复用模型。Memcached 是多线程，非阻塞 IO 复用的网络模型。 Redis 支持发布订阅模型、Lua 脚本、事务等功能，而 Memcached 不支持。并且，Redis 支持更多的编程语言。 Memcached 过期数据的删除策略只用了惰性删除，而 Redis 同时使用了惰性删除与定期删除。 为什么使用redis 保证用户下一次再访问这些数据的时候就可以直接从缓存中获取了。操作缓存就是直接操作内存，所以速度相当快。 一般像 MySQL 这类的数据库的 QPS 大概都在 1w 左右（4 核 8g） ，但是使用 Redis 缓存之后很容易达到 10w+，甚至最高能达到 30w+（就单机 redis 的情况，redis 集群的话会更高）。 redis的其他用途 分布式锁: 通过 Redis 来做分布式锁是一种比较常见的方式。通常情况下，我们都是基于 Redisson来实现分布式锁。 限流 ：一般是通过 Redis + Lua 脚本的方式来实现限流。 消息队列：Redis 自带的 list 数据结构可以作为一个简单的队列使用。Redis5.0 中增加的 Stream 类型的数据结构更加适合用来做消息队列。它比较类似于 Kafka，有主题和消费组的概念，支持消息持久化以及 ACK 机制。 复杂业务场景 ：通过 Redis 以及 Redis 扩展（比如 Redisson）提供的数据结构，我们可以很方便地完成很多复杂的业务场景比如通过 bitmap 统计活跃用户、通过 sorted set 维护排行榜。 redis中的数据结构 键key常用命令 keys * //查看当前库所有key (匹配：keys *1) exists key //判断某个key是否存在 type key //查看你的key是什么类型 del key //删除指定的key数据 unlink key //根据value选择非阻塞删除，仅将keys从keyspace元数据中删除，真正的删除会在后续异步操作。 expire key 10 //10秒钟：为给定的key设置过期时间 ttl key //查看还有多少秒过期，-1表示永不过期，-2表示已过期 select //命令切换数据库 dbsize //查看当前数据库的key的数量 flushdb //清空当前库 flushall //通杀全部库 String 介绍：string 数据结构是简单的 key-value 类型。 String类型是二进制安全的。意味着Redis的string可以包含任何数据。比如jpg图片或者序列化的对象。 虽然 Redis 是用 C 语言写的，但是 Redis 并没有使用 C 的字符串表示，而是自己构建了一种 简单动态字符串(Simple Dynamic String,缩写SDS)。 是可以修改的字符串，内部结构实现上类似于Java的ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配。 当字符串长度小于1M时，扩容都是加倍现有的空间，如果超过1M，扩容时一次只会多扩1M的空间。一个Redis中字符串value最多可以是512M。 常用命令：set,get,strlen,exists,decr,incr,setex 等等 set <key><value> //添加键值对 get <key> //查询对应键值 append <key><value> //将给定的<value> 追加到原值的末尾 strlen <key> //获得值的长度 setnx <key><value> //只有在 key 不存在时 设置 key 的值 incr <key> //将 key 中储存的数字值增1，只能对数字值操作，如果为空，新增值为1 decr <key> //将 key 中储存的数字值减1，只能对数字值操作，如果为空，新增值为-1 incrby / decrby <key><步长> //将 key 中储存的数字值增减。自定义步长。 mset <key1><value1><key2><value2> ..... //同时设置一个或多个 key-value对 mget <key1><key2><key3> ..... //同时获取一个或多个 value getrange <key><起始位置><结束位置> //获得值的范围，类似java中的substring，前包，后包 setrange <key><起始位置><value> //用 <value> 覆写<key>所储存的字符串值，从<起始位置>开始(索引从0开始)。 setex <key><过期时间><value> //设置键值的同时，设置过期时间，单位秒。 应用场景： 一般常用在需要计数的场景，比如用户的访问次数、热点文章的点赞转发数量等等。 list 介绍： Redis 的 list 是简单的字符串列表，按照插入顺序排序，实现为一个快速链表 quickList （双向链表），首先在列表元素较少的情况下会使用一块连续的内存存储，这个结构是ziplist，也即是压缩列表。 它将所有的元素紧挨着一起存储，分配的是一块连续的内存。当数据量比较多的时候才会改成quicklist。（因为普通的链表需要的附加指针空间太大，会比较浪费空间） Redis将链表和ziplist结合起来组成了quicklist。也就是将多个ziplist使用双向指针串起来使用。这样既满足了快速的插入删除性能，又不会出现太大的空间冗余。 常用命令：rpush,lpop,lpush,rpop,lrange,llen lpush/rpush <key><value1><value2><value3> .... //从左边/右边插入一个或多个值。 lpop/rpop <key> //从左边/右边吐出一个值。值在键在，值光键亡。 rpoplpush <key1><key2> //从<key1>列表右边吐出一个值，插到<key2>列表左边。 lrange <key><start><stop> //按照索引下标获得元素(从左到右) lrange mylist 0 -1 //0左边第一个，-1右边第一个，（0-1表示获取所有） lindex <key><index> //按照索引下标获得元素(从左到右) llen <key> //获得列表长度 linsert <key> before <value><newvalue> //在<value>的后面插入<newvalue>插入值 lrem <key><n><value> //从左边删除n个value(从左到右) lset<key><index><value> //将列表key下标为index的值替换成value 应用场景：发布与订阅或者说消息队列、慢查询。 hash 介绍：hash 是一个 string 类型的 field 和 value 的映射表，是一个键值对集合。 特别适合用于存储对象，类似Java里面的Map<String,Object>，后续操作的时候，你可以直接仅仅修改这个对象中的某个字段的值。 比如我们可以 hash 数据结构来存储用户信息，商品信息等等。 通过 key(用户ID) + field(属性标签) 就可以操作对应属性数据了，既不需要重复存储数据，也不会带来序列化和并发修改控制的问题 常用命令：hset,hmset,hexists,hget,hgetall,hkeys,hvals hset <key><field><value> //给<key>集合中的 <field>键赋值<value> hget <key1><field> //从<key1> 集合<field>取出 value hmset <key1><field1><value1><field2><value2>... //批量设置hash的值 hexists<key1><field> //查看哈希表 key 中，给定域 field 是否存在。 hkeys <key> //列出该hash集合的所有field hvals <key> //列出该hash集合的所有value hincrby <key><field><increment> //为哈希表 key 中的域 field 的值加上增量 1 -1 hsetnx <key><field><value> //将哈希表 key 中的域 field 的值设置为 value ，当且仅当域 field 不存在 . 应用场景：系统中对象数据的存储。 set 介绍：Redis 中的 set 类型是一种无序集合(自动去重)，集合中的元素没有先后顺序。它底层数据结构是dict字典，字典是用hash表实现的，是一个value为null的hash表，所以添加，删除，查找的复杂度都是O(1)。 可以基于 set 轻易实现交集、并集、差集的操作。比如：你可以将一个用户所有的关注人存在一个集合中，将其所有粉丝存在一个集合。Redis 可以非常方便的实现如共同关注、共同粉丝、共同喜好等功能。这个过程也就是求交集的过程。 常用命令：sadd,spop,smembers,sismember,scard,sinterstore,sunion等。 sadd <key><value1><value2> ..... //将一个或多个 member 元素加入到集合 key 中，已经存在的 member 元素将被忽略 smembers <key> //取出该集合的所有值。 sismember <key><value> //判断集合<key>是否为含有该<value>值，有1，没有0 scard<key> //返回该集合的元素个数。 srem <key><value1><value2> .... //删除集合中的某个元素。 spop <key> //随机从该集合中吐出一个值。 srandmember <key><n> //随机从该集合中取出n个值。不会从集合中删除 。 smove <destination>value //把集合中一个值从一个集合移动到另一个集合 sinter <key1><key2> //返回两个集合的交集元素。 sunion <key1><key2> //返回两个集合的并集元素。 sdiff <key1><key2> //返回两个集合的差集元素(key1中的，不包含key2中的) 应用场景: 需要存放的数据不能重复以及需要获取多个数据源交集和并集等场景 Zset(sorted set) 介绍： 和 set 相比，sorted set 增加了一个权重参数 score（可重复），使得集合中的元素能够按 score 进行有序排列，还可以通过 score 的范围来获取元素的列表。有点像是 Java 中 HashMap 和 TreeSet 的结合体。 zset底层使用了两个数据结构： hash，hash的作用就是关联元素value和权重score，保证元素value的唯一性，可以通过元素value找到相应的score值。 跳跃表，跳跃表的目的在于给元素value排序，根据score的范围获取元素列表。 常用命令： zadd,zcard,zscore,zrange,zrevrange,zrem 等。 zadd <key><score1><value1><score2><value2>… //将一个或多个 member 元素及其 score 值加入到有序集 key 当中。 zrange <key><start><stop> [WITHSCORES] //返回有序集 key 中，下标在<start><stop>之间的元素带WITHSCORES，可以让分数一起和值返回到结果集。 zrangebyscore key minmax [withscores] [limit offset count] //返回有序集 key 中，所有 score 值介于 min 和 max 之间(包括等于 min 或 max )的成员。有序集成员按 score 值递增(从小到大)次序排列。 zrevrangebyscore key maxmin [withscores] [limit offset count] //同上，改为从大到小排列。 zincrby <key><increment><value> //为元素的score加上增量 zrem <key><value> //删除该集合下，指定值的元素 zcount <key><min><max> //统计该集合，分数区间内的元素个数 zrank <key><value> //返回该值在集合中的排名，从0开始。 应用场景： 需要对数据根据某个权重进行排序的场景。比如在直播系统中，实时排行信息包含直播间在线用户列表，各种礼物排行榜，弹幕消息（可以理解为按消息维度的消息排行榜）等信息。 案例：如何利用zset实现一个文章访问量的排行榜？ bitmap 介绍：bitmap 存储的是连续的二进制数字（0 和 1），通过 bitmap, 只需要一个 bit 位来表示某个元素对应的值或者状态，key 就是对应元素本身 。我们知道 8 个 bit 可以组成一个 byte，所以 bitmap 本身会极大的节省储存空间。 Bitmaps本身不是一种数据类型， 实际上它就是字符串（key-value） ， 但是它可以对字符串的位进行操作。 可以把Bitmaps想象成一个以位为单位的数组， 数组的每个单元只能存储0和1， 数组的下标在Bitmaps中叫做偏移量offset。（offset:偏移量从0开始） 常用命令： setbit 、getbit 、bitcount、bitop setbit<key><offset><value> //设置Bitmaps中某个偏移量的值（0或1） getbit<key><offset> //获取Bitmaps中某个偏移量的值 bitcount<key>[start end] //统计字符串从start字节到end字节比特值为1的数量 bitop and(or/not/xor) <destkey> [key…] //bitop是一个复合操作， 它可以做多个Bitmaps的and（交集） 、 or（并集） 、 not（非） 、 xor（异或） 操作并将结果保存在destkey中。 应用场景： 适合需要保存状态信息（比如是否签到、是否登录...）并需要进一步对这些信息进行分析的场景。比如用户签到情况、活跃用户情况、用户行为统计（比如是否点赞过某个视频）。 实例： 每个独立用户是否访问过网站存放在Bitmaps中， 将访问的用户记做1， 没有访问的用户记做0， 用偏移量作为用户的id。 设置键的第offset个位的值（从0算起） ， 假设现在有20个用户，userid=1， 6， 11， 15， 19的用户对网站进行了访问， 那么当前Bitmaps初始化结果如图 Bitmaps与set对比 假设网站有1亿用户， 每天独立访问的用户有5千万， 如果每天用集合类型和Bitmaps分别存储活跃用户可以得到表: 很明显， 这种情况下使用Bitmaps能节省很多的内存空间， 尤其是随着时间推移节省的内存还是非常可观的 但Bitmaps并不是万金油， 假如该网站每天的独立访问用户很少， 例如只有10万（大量的僵尸用户） ， 那么两者的对比如下表所示， 很显然， 这时候使用Bitmaps就不太合适了， 因为基本上大部分位都是0。 HyperLogLog [scode type="share"]基数问题：在工作当中，我们经常会遇到与统计相关的功能需求，比如统计网站PV（PageView页面访问量）,可以使用Redis的incr、incrby轻松实现。 但像UV（UniqueVisitor，独立访客）、独立IP数、搜索记录数等需要去重和计数的问题如何解决？这种求集合中不重复元素个数的问题称为基数问题。[/scode] Redis HyperLogLog 是用来做基数统计的算法，HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。 pfadd <key>< element> [element ...] //添加指定元素到 HyperLogLog 中 pfcount<key> [key ...] //计算HLL的近似基数，可以计算多个HLL，比如用HLL存储每天的UV，计算一周的UV可以使用7天的UV合并计算即可 pfmerge<destkey><sourcekey> [sourcekey ...] //将一个或多个HLL合并后的结果存储在另一个HLL中，比如每月活跃用户可以使用每天的活跃用户来合并计算可得 6.3. Geospatial Redis 3.2 中增加了对GEO类型的支持。 GEO，Geographic，地理信息的缩写。该类型，就是元素的2维坐标，在地图上就是经纬度。 redis基于该类型，提供了经纬度设置，查询，范围查询，距离查询，经纬度Hash等常见操作。 geopos <key><member> [member...] //获得指定地区的坐标值 geodist<key><member1><member2> [m|km|ft|mi ] //获取两个位置之间的直线距离 georadius<key>< longitude><latitude>radius m|km|ft|mi //以给定的经纬度为中心，找出某一半径内的元素 发布订阅 Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。 Redis 客户端可以订阅任意数量的频道。 命令 SUBSCRIBE channel1 //客户端订阅channel1 publish channel1 hello //另一个客户端，给channel1发布消息hello 线程模型 Redis单线程模型详解 Redis 基于 Reactor 模式来设计开发了自己的一套高效的事件处理模型 （Netty 的线程模型也基于 Reactor 模式，Reactor 模式不愧是高性能 IO 的基石），这套事件处理模型对应的是 Redis 中的文件事件处理器（file event handler）。由于文件事件处理器（file event handler）是单线程方式运行的，所以我们一般都说 Redis 是单线程模型。 既然是单线程，那怎么监听大量的客户端连接呢？ Redis 通过IO 多路复用程序 来监听来自客户端的大量连接（或者说是监听多个 socket），它会将感兴趣的事件及类型（读、写）注册到内核中并监听每个事件是否发生。 这样的好处非常明显： I/O 多路复用技术的使用让 Redis 不需要额外创建多余的线程来监听客户端的大量连接，降低了资源的消耗（和 NIO 中的 Selector 组件很像）。 另外， Redis 服务器是一个事件驱动程序，服务器需要处理两类事件：1. 文件事件; 2. 时间事件。 时间事件不需要多花时间了解，我们接触最多的还是 文件事件（客户端进行读取写入等操作，涉及一系列网络通信）。 《Redis 设计与实现》有一段话是如是介绍文件事件的，我觉得写得挺不错。 [scode type="green"]Redis 基于 Reactor 模式开发了自己的网络事件处理器：这个处理器被称为文件事件处理器（file event handler）。文件事件处理器使用 I/O 多路复用（multiplexing）程序来同时监听多个套接字，并根据套接字目前执行的任务来为套接字关联不同的事件处理器。[/scode] [scode type="green"]当被监听的套接字准备好执行连接应答（accept）、读取（read）、写入（write）、关 闭（close）等操作时，与操作相对应的文件事件就会产生，这时文件事件处理器就会调用套接字之前关联好的事件处理器来处理这些事件。[/scode] [scode type="green"]虽然文件事件处理器以单线程方式运行，但通过使用 I/O 多路复用程序来监听多个套接字，文件事件处理器既实现了高性能的网络通信模型，又可以很好地与 Redis 服务器中其他同样以单线程方式运行的模块进行对接，这保持了 Redis 内部单线程设计的简单性。[/scode] 可以看出，文件事件处理器（file event handler）主要是包含 4 个部分： 多个 socket（客户端连接） IO 多路复用程序（支持多个客户端连接的关键） 文件事件分派器（将 socket 关联到相应的事件处理器） 事件处理器（连接应答处理器、命令请求处理器、命令回复处理器） 多线程 Redis 6.0 之前主要还是单线程处理。Redis6.0 之前 为什么不使用多线程？ 单线程编程容易并且更容易维护； Redis 的性能瓶颈不在 CPU ，主要在内存和网络； 多线程就会存在死锁、线程上下文切换等问题，甚至会影响性能。 Redis6.0 之后为何引入了多线程？ Redis6.0 引入多线程主要是为了提高网络 IO 读写性能，因为这个算是 Redis 中的一个性能瓶颈（Redis 的瓶颈主要受限于内存和网络）。 Redis6.0 的多线程默认是禁用的，只使用主线程。开启多线程后，还需要设置线程数，否则是不生效的。 过期 Redis 中除了字符串类型有自己独有设置过期时间的命令 setex 外，其他方法都需要依靠 expire 命令来设置过期时间 。另外， persist 命令可以移除一个键的过期时间。 如何判断过期 Redis 通过一个叫做过期字典（可以看作是 hash 表）来保存数据过期的时间。过期字典的键指向 Redis 数据库中的某个 key(键)，过期字典的值是一个 long long 类型的整数，这个整数保存了 key 所指向的数据库键的过期时间（毫秒精度的 UNIX 时间戳）。 删除策略 惰性删除 ：只会在取出 key 的时候才对数据进行过期检查。这样对 CPU 最友好，但是可能会造成太多过期 key 没有被删除。 定期删除 ： 每隔一段时间抽取一批 key 执行删除过期 key 操作。并且，Redis 底层会通过限制删除操作执行的时长和频率来减少删除操作对 CPU 时间的影响。 定期删除对内存更加友好，惰性删除对 CPU 更加友好。两者各有千秋，所以 Redis 采用的是 定期删除+惰性/懒汉式删除 。 淘汰机制 Redis 提供 6 种数据淘汰策略： volatile-lru（least recently used）：从已设置过期时间的数据集（server.db[i].expires）中挑选最近最少使用的数据淘汰 volatile-ttl：从已设置过期时间的数据集（server.db[i].expires）中挑选将要过期的数据淘汰 volatile-random：从已设置过期时间的数据集（server.db[i].expires）中任意选择数据淘汰 allkeys-lru（least recently used）：当内存不足以容纳新写入数据时，在键空间中，移除最近最少使用的 key（这个是最常用的） allkeys-random：从数据集（server.db[i].dict）中任意选择数据淘汰 no-eviction：禁止驱逐数据，也就是说当内存不足以容纳新写入数据时，新写入操作会报错。这个应该没人使用吧！ 4.0 版本后增加以下两种： volatile-lfu（least frequently used）：从已设置过期时间的数据集（server.db[i].expires）中挑选最不经常使用的数据淘汰 allkeys-lfu（least frequently used）：当内存不足以容纳新写入数据时，在键空间中，移除最不经常使用的 key 持久化 Redis 的一种持久化方式叫快照（snapshotting，RDB,Redis DataBase），另一种方式是只追加文件（append-only file, AOF） RDB快照 Redis 可以通过创建快照来获得存储在内存里面的数据在某个时间点上的副本。在指定的时间间隔内将内存中的数据集快照写入磁盘， 也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里 Redis 创建快照之后，可以对快照进行备份，可以将快照复制到其他服务器从而创建具有相同数据的服务器副本（Redis 主从结构，主要用来提高 Redis 性能），还可以将快照留在原地以便重启服务器的时候使用。 快照持久化是 Redis 默认采用的持久化方式 过程 Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到 一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。 整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能 如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。 RDB的缺点是最后一次持久化后的数据可能丢失。 命令 save :只管保存，其它不管，全部阻塞。手动保存。不建议。 bgsave:Redis会在后台异步进行快照操作， 快照同时还可以响应客户端请求。 总结 优势： 适合大规模的数据恢复 对数据完整性和一致性要求不高更适合使用 节省磁盘空间 恢复速度快 缺点： Fork的时候，内存中的数据被克隆了一份，大致2倍的膨胀性需要考虑 虽然Redis在fork时使用了写时拷贝技术,但是如果数据庞大时还是比较消耗性能。 在备份周期在一定间隔时间做一次备份，所以如果Redis意外down掉的话，就会丢失最后一次快照后的所有修改。 AOF（append-only file）持久化 以日志的形式来记录每个写操作（增量保存），将Redis执行过的所有写指令记录下来(读操作不记录)， 只许追加文件但不可以改写文件 redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作 与快照持久化相比，AOF 持久化的实时性更好，因此已成为主流的持久化方案。 AOF默认不开启，AOF和RDB同时开启，系统默认取AOF的数据（数据不会存在丢失） 过程 客户端的请求写命令会被append追加到AOF缓冲区内 AOF缓冲区根据AOF持久化策略[always,everysec,no]将操作sync同步到磁盘的AOF文件中 AOF文件大小超过重写策略或手动重写时，会对AOF文件rewrite重写，压缩AOF文件容量 Redis服务重启时，会重新load加载AOF文件中的写操作达到数据恢复的目的 AOF同步频率设置： appendfsync always ：始终同步，每次Redis的写入都会立刻记入日志；性能较差但数据完整性比较好 appendfsync everysec ：每秒同步，每秒记入日志一次，如果宕机，本秒的数据可能丢失。 appendfsync no ：redis不主动进行同步，把同步时机交给操作系统。 Rewrite压缩 AOF采用文件追加方式，文件会越来越大为避免出现此种情况，新增了重写机制, 当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩， 只保留可以恢复数据的最小指令集 总结 优势： 备份机制更稳健，丢失数据概率更低。 可读的日志文本，通过操作AOF稳健，可以处理误操作。 缺点： 比起RDB占用更多的磁盘空间。 恢复备份速度要慢。 每次读写都同步的话，有一定的性能压力。 总结 官方推荐两个都启用。 如果对数据不敏感，可以选单独用RDB。 不建议单独用 AOF，因为可能会出现Bug。 如果只是做纯内存缓存，可以都不用。 事务 Redis事务是一个单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。 Redis事务的主要作用就是串联多个命令防止别的命令插队。 Redis 可以通过 MULTI，EXEC，DISCARD 和 WATCH 等命令来实现事务(transaction)功能。 Multi、Exec、discard 从输入Multi命令开始，输入的命令都会依次进入命令队列中，但不会执行，直到输入Exec后，Redis会将之前的命令队列中的命令依次执行。 这个过程是这样的： 开始事务（MULTI）。 命令入队(批量操作 Redis 的命令，先进先出（FIFO）的顺序执行)。 执行事务(EXEC)。 你也可以通过 DISCARD 命令取消一个事务，它会清空事务队列中保存的所有命令。 事务的错误处理 组队中某个命令出现了报告错误，执行时整个的所有队列都会被取消。 如果执行阶段某个命令报出了错误，则只有报错的命令不会被执行，而其他的命令都会执行，不会回滚。 Redis 是不支持 roll back 的，因而不满足原子性的（而且不满足持久性）。 WATCH key 在执行multi之前，先执行watch key1 [key2],可以监视一个(或多个) key 如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。 unwatch 取消 WATCH 命令对所有 key 的监视。 如果在执行 WATCH 命令之后，EXEC 命令或DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。 事务三特性 单独的隔离操作： 事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。 没有隔离级别的概念： 队列中的命令没有提交之前都不会实际被执行，因为事务提交前任何指令都不会被实际执行 不保证原子性 事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚 redis乐观锁解决超卖 //增加乐观锁 jedis.watch(qtkey); //判断库存 String qtkeystr = jedis.get(qtkey); if(qtkeystr==null || "".equals(qtkeystr.trim())) { System.out.println("未初始化库存"); jedis.close(); return false ; } int qt = Integer.parseInt(qtkeystr); if(qt<=0) { System.err.println("已经秒光"); jedis.close(); return false; } //增加事务 Transaction multi = jedis.multi(); //减少库存 //jedis.decr(qtkey); multi.decr(qtkey); //加入 //jedis.sadd(usrkey, uid); multi.sadd(usrkey, uid); //执行事务 List<Object> list = multi.exec(); //判断事务提交是否失败 if(list==null || list.size()==0) { System.out.println("秒杀失败"); jedis.close(); return false; } System.err.println("秒杀成功"); jedis.close(); LUA Lua 是一个小巧的脚本语言，Lua脚本可以很容易的被C/C++ 代码调用，也可以反过来调用C/C++的函数，Lua并没有提供强大的库，一个完整的Lua解释器不过200k，所以Lua不适合作为开发独立应用程序的语言，而是作为嵌入式脚本语言。 在redis中的应用 将复杂的或者多步的redis操作，写为一个脚本，一次提交给redis执行，减少反复连接redis的次数。提升性能。 LUA脚本是类似redis事务，有一定的原子性，不会被其他命令插队，可以完成一些redis事务性的操作。 利用lua脚本淘汰用户，解决超卖问题。 redis 2.6版本以后，通过lua脚本解决争抢问题，实际上是redis 利用其单线程的特性，用任务队列的方式解决多任务并发问题。 解决库存遗留问题 local userid=KEYS[1]; local prodid=KEYS[2]; local qtkey="sk:"..prodid..":qt"; local usersKey="sk:"..prodid.":usr'; local userExists=redis.call("sismember",usersKey,userid); if tonumber(userExists)==1 then return 2; end local num= redis.call("get" ,qtkey); if tonumber(num)<=0 then return 0; else redis.call("decr",qtkey); redis.call("sadd",usersKey,userid); end return 1; 缓存穿透 缓存穿透说简单点就是大量请求的 key 根本不存在于缓存中，导致请求直接到了数据库上，根本没有经过缓存这一层。 举个例子：某个黑客故意制造我们缓存中不存在的 key 发起大量请求，导致大量请求落到数据库。 解决方法 最基本的就是首先做好参数校验，一些不合法的参数请求直接抛出异常信息返回给客户端。比如查询的数据库 id 不能小于 0、传入的邮箱格式不对的时候直接返回错误消息给客户端等等。 缓存无效key 尽量将无效的 key 的过期时间设置短一点比如 1 分钟。 布隆过滤器 是一种基于概率的数据结构，判断当前某个元素是否在该集合中，运行速度快。 一个非常大的二进制位数组（数组中只存在 0 和 1） 拥有若干个哈希函数（Hash Function） 在空间效率和查询效率都非常高 布隆过滤器不会提供删除方法，在代码维护上比较困难。 有一定的误判 每个布隆过滤器对应到 Redis 的数据结构里面就是一个大型的位数组和几个不一样的无偏 hash 函数。 所谓无偏就是能够把元素的 hash 值算得比较均匀。 向布隆过滤器中添加 key 时，会使用多个 hash 函数对 key 进行 hash 算得一个整数索引值然后对位数组长度进行取模运算得到一个位置，每个 hash 函数都会算得一个不同的位置。 再把位数组的这几个位置都置为 1 就完成了 add 操作。 （ 每一个 key 都通过若干的hash函数映射到一个巨大位数组上，映射成功后，会在把位数组上对应的位置改为1。） 缓存击穿 原因： 一个“冷门”key，突然被大量用户请求访问。 一个“热门”key，在缓存中时间恰好过期，这时有大量用户来进行访问。 导致大并发请求直接请求数据库，瞬间对数据库的访问压力增大。 解决： 我们常用的解决方案是加锁。 对于key过期的时候，当key要查询数据库的时候加上一把锁，这时只能让第一个请求进行查询数据库，然后把从数据库中查询到的值存储到缓存中，对于剩下的相同的key，可以直接从缓存中获取即可。 如果我们是在单机环境下：直接使用常用的锁即可（如：Lock、Synchronized等） 在分布式环境下我们可以使用分布式锁，如：基于数据库、基于Redis或者zookeeper 的分布式锁。 还有就是：对于访问特别频繁的热点数据，不设置过期时间 缓存雪崩 Redis 服务不可用： 缓存在同一时间大面积的失效，后面的请求都直接落到了数据库上，造成数据库短时间内承受大量请求。 Redis 大面积过期：有一些被大量访问数据（热点缓存）在某一时刻大面积失效，导致对应的请求直接落到了数据库上。 解决 针对 Redis 服务不可用的情况： 采用 Redis 集群，避免单机出现问题整个缓存服务都没办法使用。 限流，避免同时处理大量的请求。 针对热点缓存失效的情况： 设置不同的失效时间比如随机设置缓存的失效时间。 缓存永不失效。 数据预热 另外还有三个建议： 针对缓存雪崩，合理地设置数据过期时间，以及搭建高可靠缓存集群。 针对缓存击穿，在缓存访问非常频繁的热点数据时，不要设置过期时间。 针对缓存穿透，提前在入口前端实现恶意请求检测，或者规范数据库的数据删除操作，避免误删除。 保证缓存和数据库数据的一致性 读写策略 Cache Aside Pattern（旁路缓存模式） ：更新 DB，然后直接删除 cache 。 如果更新数据库成功，而删除缓存这一步失败的情况的话，简单说两个解决方案： 缓存失效时间变短（不推荐，治标不治本）：我们让缓存数据的过期时间变短，这样的话缓存就会从数据库中加载数据。另外，这种解决办法对于先操作缓存后操作数据库的场景不适用。 增加 cache 更新重试机制（常用）： 如果 cache 服务当前不可用导致缓存删除失败的话，我们就隔一段时间进行重试，重试次数可以自己定。如果多次重试还是失败的话，我们可以把当前更新失败的 key 存入队列中，等缓存服务可用之后，再将缓存中对应的 key 删除即可。 读写策略 Cache Aside Pattern Cache Aside Pattern （旁路缓存模式） 是我们平时使用比较多的一个缓存读写模式，比较适合读请求比较多的场景。 Cache Aside Pattern 中服务端需要同时维系 DB 和 cache，并且是以 DB 的结果为准。 写： 先更新 DB 然后直接删除 cache 。 读： 从 cache 中读取数据，读取到就直接返回 cache中读取不到的话，就从 DB 中读取数据返回 再把数据放到 cache 中。 问题 在写数据的过程中，可以先删除 cache ，后更新 DB 么？ 不行：请求1先把cache中的A数据删除 -> 请求2从DB中读取数据->请求1再把DB中的A数据更新。 在写数据的过程中，先更新DB，后删除cache就没有问题了么？ 不一定： 请求1从DB读数据A->请求2写更新数据 A 到数据库并把删除cache中的A数据->请求1将数据A写入cache。 缺陷 首次请求数据一定不在 cache 的问题：可以将热点数据可以提前放入cache 中。 写操作比较频繁的话导致cache中的数据会被频繁被删除，这样会影响缓存命中率 ： 数据库和缓存数据强一致场景 ：更新DB的时候同样更新cache，不过我们需要加一个锁/分布式锁来保证更新cache的时候不存在线程安全问题。 可以短暂地允许数据库和缓存数据不一致的场景 ：更新DB的时候同样更新cache，但是给缓存加一个比较短的过期时间，这样的话就可以保证即使数据不一致的话影响也比较小。 Read/Write Through Pattern Read/Write Through Pattern （读写穿透）中服务端把 cache 视为主要数据存储，从中读取数据并将数据写入其中。 cache 服务负责将此数据读取和写入 DB，从而减轻了应用程序的职责。 写： 先查 cache，cache 中不存在，直接更新 DB。 cache 中存在，则先更新 cache，然后 cache 服务自己更新 DB（同步更新 cache 和 DB）。 读(Read Through)： 从 cache 中读取数据，读取到就直接返回 。 读取不到的话，先从 DB 加载，写入到 cache 后返回响应。 Write Behind Pattern Write Behind Pattern （异步缓存写入）和 Read/Write Through Pattern 很相似，两者都是由 cache 服务来负责 cache 和 DB 的读写。 但是，两个又有很大的不同：Read/Write Through 是同步更新 cache 和 DB，而 Write Behind Caching 则是只更新缓存，不直接更新 DB，而是改为异步批量的方式来更新 DB。 很明显，这种方式对数据一致性带来了更大的挑战，比如cache数据可能还没异步更新DB的话，cache服务可能就就挂掉了。 Write Behind Pattern 下 DB 的写性能非常高，非常适合一些数据经常变化又对数据一致性要求没那么高的场景，比如浏览量、点赞量。 缓存异常 缓存和数据库的数据一致性包含两种情况： 缓存中有数据，缓存的数据值需要和数据库中的值相同 缓存中没有数据，数据库中的值必须是最新值 数据不一致是如何发生的 读写缓存：有两种写回策略，同步直写和异步写回。如果要保证数据一致，就要采用同步直写策略。但需要保证缓存和数据库的更新具有原子性，即要么都成功，要么都失败。 只读缓存：分新增数据和删改数据两种情况说明。 新增数据：数据直接写到数据库中，不对缓存做任何操作，符合一致性的第2种情况。 删改数据：发生删改操作时，既要更新数据库，也要在缓存里删除数据。因为缓存和数据库是不同的系统，这里分两种情况： 先删除缓存，再更新数据库：数据库更新失败，导致请求再次访问缓存时，发现缓存失败，再读数据库时，从数据库中读取旧值。 先更新数据库，再删除缓存：缓存删除失败，导致请求再次访问缓存时，发现缓存命中，并从缓存中读取到旧值。 如何解决数据不一致 使用重试机制，指把删除的缓存值或者是要更新的数据库值暂存到消息队列中（例如使用Kafka消息队列）。 当应用没有能够成功地删除缓存值或者是更新数据库值时，从消息队列中重新读取这些值，然后再次进行删除或更新。 如果成功删除，就从消息队列中删除，以免重复操作。否则就要进行重试，如果重试超过一定次数，就要向业务层发送报错信息。 对于只读缓存来说，建议优先使用先更新数据库，再删除缓存。 主从 读写分离，性能扩展，容灾快速恢复
通过Redis的自增set实现“乐观锁”，从而控制并发
[scode type="blue"]这还要从项目提出的静默签到开始讲起[/scode] 学长说我们的新院统战要对用户的登录进行记录，也就是签到，但是不是用户手动去页面点击签到，而是用户今天登录了，或者说向服务器发送请求了，那么就算是签过到了，数据库有他今天的登录记录。 emmm，一开始还真是让我头大，因为不是新写一个接口去进行签到，而是通过请求进行签到，一天只能签一次，那么首先这个判断的位置只能是放在过滤器中，当用户请求进来时，拦截下来，拿token，获取用户id。 ok，拿到用户id了，但是不能去通过查数据库来检测用户是否这一天已经登陆过，因为慢，而且请求是时时刻刻都有的，查数据库显然不理想，那肯定就是去查redis里有没有今天的记录，把userId当作key，value随意，就有了一开始的写法： //如果缓存里没有 if (!redisUtils.hasKey(key)) { //写入数据库 ... //写入缓存，第二天凌晨过期 redisUtils.set(key, value, DateUtil.getTomorrowSeconds(new Date())); } 一开始这样写，还没有什么太大的问题，但是之后就发现了数据库第一时刻记录了多条用户登录记录，经过排查之后，可以初步判定是多条请求同时进来，而 if (!redisUtils.hasKey(key)) 需要时间并没有全部拦截，所以上面的写法是有问题的。 那我就想了，既然redis读取有延时，我直接用 ConcurrentHashMap 来记录做外层第一次拦截,同时写一个定时任务清空，同时里面再加第二层判断拦截记录到redis（考虑到服务器重启），这样延迟足够小了吧，应该可以拦下并发的请求访问的，可结果告诉我还是不行，即使是用了map也拦截不全。 [scode type="blue"]通过redis的自增set做乐观锁[/scode] 现在可以明确是要控制并发访问了，这一块我之前有写过悲观锁和乐观锁，但很明显这个不能使用悲观锁，太慢了。那就使用校验字段来加乐观锁，通过查阅知道redis是单线程的，所以在写入操作一定是线程安全的！那么就要使用到redis的自增set了： /** * 缓存自增放入（乐观锁） * * @param key 键 * @param value 自增大小（Long） * @return 返回已增长度 */ public Long setnc(String key, Object value) { try { return redisTemplate.opsForValue().increment(key, (Long)value); } catch (Exception e) { log.error(e.getMessage(), e); return 0L; } } 这个可以保证每一次进入的请求都会增加该条缓存的增量记录值，由于redis是线程安全的，所以可以保证每个请求最后返回的自增长度是不同的，那么怎们就对第一次自增后的长度做一个判断： //如果自增长度小于等于1（只有第一次，后面的都会越来越大） if (redisUtils.setnc(key, 1L) <= 1L) { //存入数据库... //设置过期时间 redisUtils.expire(SIGN_KEY + userId, DateUtil.getTomorrowSeconds(new Date())); } 经过测试，可以控制redis的并发访问所带来的差错，这样，静默签到算是完成了。
Mybatis结合Redis缓存以注解开发时的查询优化方案——通过AOP和反射重新实现@One、@Many注解
前言 [scode type="share"]现在算是终于有空来写这篇文章了，由于考研的问题一直没有时间来补这篇文章[/scode] 该问题最早提出是在9月份开发新院统战项目时提出的，也就是我刚刚搭完第一版后端框架，其中持久层使用的框架是mybatis plus，缓存使用的是sprinboot缓存注解+redis，其中将缓存注解加在service层。 当时在使用mybatis plus并以注解开发时，由于过度使用@One与@Many注解进行属性的封装，例如： @Select("SELECT id,dept_id,username,nick_name,gender,phone,email,avatar_path,password," + "is_admin,enabled,create_by,update_by,pwd_reset_time,create_time,update_time" + " FROM sys_user u WHERE u.id = #{id} AND is_deleted=0") @Results({ @Result(column = "id", property = "id"), @Result(column = "dept_id", property = "deptId"), @Result(column = "dept_id", property = "dept", one = @One(select = "marchsoft.modules.system.mapper.DeptMapper.selectById", fetchType = FetchType.EAGER)), @Result(column = "id", property = "roles", many = @Many(select = "marchsoft.modules.system.mapper.RoleMapper.findWithMenuByUserId", fetchType = FetchType.EAGER)), @Result(column = "id", property = "jobs", many = @Many(select = "marchsoft.modules.system.mapper.JobMapper.findByUserId", fetchType = FetchType.EAGER)) }) UserBO findUserDetailById(Long id); 但是没有考虑到一句@One就是一条新的sql，一句@Many就又是n条sql，在没有缓存的情况下效率是非常低的，这就牵扯到之后的不断探究与讨论了。 探究 [scode type="share"]在之后的不断讨论与探究中，主要从以下几个方面就行抉择：[/scode] [scode type="blue"]选择 springboot + redis 缓存[/scode] 保持与原先与eladmin相同的缓存方法，通过@CacheConfig与@Cacheable将缓存加在service中，仅仅只缓存以id或某字段为主键的记录（类似数据库中一条一条的数据），不缓存分页查询或自己封装后的数据。同时自己对哪些数据加了缓存，还要需要自己手动去维护。 @Service @AllArgsConstructor @CacheConfig(cacheNames = "role") public class RoleServiceImpl implements IRoleService { private final RoleMapper roleMapper; @Override @Cacheable(key = "'id:' + #p0") public Role getOneById(Serializable id) { return this.roleMapper.selectById(id); } @Override public List<Role> getRolesById(Serializable id) { return roleMapper.getRoleByUser(id); } } 这样加缓存的好处是缓存键值清晰，便于存取和删除，还可以建立缓存的命名空间，使缓存结构清晰易于管理。 [scode type="blue"]springboot缓存方式与@One、@Many之间的矛盾[/scode] 上述的缓存方式虽然很理想，但是会导致@One与@Many的查询无法正常走缓存，首先是@One与@Many方法只能调用mapper层的查询方法(调用service层会报错)，这就必须要求@CacheConfig与@Cacheable需要加在mapper层的接口方法上（springboot不建议将缓存加在接口方法上） [scode type="yellow"]其次即使将缓存注解加在了mapper层上，@One与@Many中所调用的查询方法仍然无法走缓存[/scode] 原因是@One与@Many在实现原理上并不是直接通过反射真正的去调用所指定的方法，而是与当前目标方法结合在了一起。 到这里就可以发现第一个令人头疼的地方，就是如果使用原生@One与@Many进行开发时，是必定走不了springboot的缓存方式，这里就需要考虑第二种缓存方式了。 [scode type="blue"]mybatis的二级缓存[/scode] 能使原生@One与@Many以及其所在的方法走缓存的唯一方式就是使用mybatis的二级缓存，当然mybatis自己实现的缓存是存在项目本身的内存中，这当然不符合我们的要求，这里是结合redis实现ibatis的cache接口来开启二级缓存。 @CacheNamespace(implementation = MybatisRedisCache.class, eviction = MybatisRedisCache.class) @Component public interface UserMapper extends BasicMapper<User> {} 但其实使用了mybatis的二级缓存之后，看起来貌似解决了@One与@Many不走缓存的问题，但是却又带了许多其他的小问题： [scode type="yellow"]mybatis plus的分页插件有的时候显示总条数与总页数为0，但结果数据还有[/scode] 该问题网上也有提到过，但是唯一的建议就是关闭mybatis的二级缓存，说是二级缓存与分页插件之间有冲突。 [scode type="yellow"]缓存没有命名空间，键名是一串随机数字，不易管理与维护[/scode] 使用了二级缓存结合redis之后，redis中的缓存就变成了这个样子： 这相较于原先使用springboot + redis缓存的结果乱了很多，而且根本不知道哪个key到底存了什么。 [scode type="yellow"]二级缓存的自动维护会删除该mapper下的所有缓存[/scode] 手动通过redis实现ibatis的cache接口有以下几个需要实现的方法： public class MyRedisCache implements Cache { //当前放入缓存的mapper的namespace private final String id; //必须要有这个构造方法 public MyRedisCache(String id){ this.id = id; } //返回cache唯一标识————即namespace @Override public String getId() { return this.id; } @Override public void putObject(Object key, Object value) { System.out.println("key : " + key.toString() + "--- value : " + value.toString()); //使用redishash类型作为缓存存储模型 key hashkey value getRedisTemplate().opsForHash().put(id.toString(), key.toString(), value); } @Override public Object getObject(Object key) { return getRedisTemplate().opsForHash().get(id.toString(), key.toString()); } @Override public Object removeObject(Object o) { return null; } //清空整个id的所有key缓存,mybatis默认走这个 @Override public void clear() { getRedisTemplate().delete(id.toString()); } @Override public int getSize() { return getRedisTemplate().opsForHash().size(id).intValue(); } private RedisTemplate getRedisTemplate(){ RedisTemplate redisTemplate = SpringContextHolder.getBean("redisTemplate"); redisTemplate.setKeySerializer(new StringRedisSerializer()); redisTemplate.setHashKeySerializer(new StringRedisSerializer()); return redisTemplate; } } 其中mybatis二级缓存存储一个记录需要有id与key来标识，其中id就是所在的mapper，key就是键名 [scode type="yellow"]mybatis的二级缓存在遇到删改方法时会自动清理缓存，走的是cache接口中的clear方法，会将该id下（就是该mapper）的所有缓存全部删除，而不是根据具体的某个key去删除缓存。[/scode] 相较于之前springboot + redis的手动维护缓存，是你删除和更新了某条数据，才仅仅将这条数据的缓存删除，而mybatis二级缓存这样直接清除力度过大，如果说为什么不清单个，原因应该有以下两个： removeObject(Object o)还未实现，这是mybatis自身待完善的地方 由于二级缓存存储时键名是随机的，维护难度过大。 上述就是使用二级缓存时又带来的困扰 [scode type="blue"]不使用@One、@Many注解，手动封装结果，调用service层方法走springboot缓存[/scode] 这个是讨论的最后一种方式，就是不使用@One、@Many注解直接在mapper层进行数据封装，而是第一次查到主体数据，在业务层再次调用其他service层的方法对剩余属性进行封装，这样调用的其他方法就可以走springboot的缓存方式了，这个方案是一个折中的方案，虽然没有@One、@Many的方便开发效率，但是至少维持了适中的执行效率和便于维护的缓存存储结构与方式。 最终方案 [scode type="share"]由于博主本人需要考研的原因，11月之后不再参与框架的更新迭代，缓存的问题也一直遗留了下来，目前第二版框架SMPE已经基本成型，最大的遗留问题就是缓存问题了[/scode] [scode type="blue"]2021年1月10日博主再次参与框架的更新迭代中，并且再次接手框架缓存的解决方案[/scode] [scode type="blue"]2021年1月10日晚与同学星星讨论后，定下尝试解决方案——重新实现@One与@Many注解，使新注解可以调用service层/mapper层的方法，从而走springboot缓存，其目标为既可以到达原生@One与@Many的开发效率，又可以使用最理想的缓存存储方案！[/scode] 幸运的是，我们的初步想法是可行的，通过1天的努力与尝试，运用springboot切面编程的思想与java反射原理开发了新的注解——@Query与@Queries，通过新注解@Query与@Queries实现了原先@One和@Many的效果，最重要的调用的方法是通过反射去真真正正调用了一条方法，因此成功走了springboot的缓存。 [scode type="green"]注解源码展示[/scode] 首先是 @Query // 重新实现 @One、@Many注解,关联查询注解，用于mapper层查询方法上,仅需关联查询一个属性时，可以直接使用@Query // 如果需要使用多个@Query，请先使用@Queries套在外层（类似 @Results）参数、作用基本和原@One保持一致 // 既可以用于一对一也可以一对多 // // @author Wangmingcan // Date: 2021/01/12 09:35 // @Documented @Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) @Repeatable(Queries.class) public @interface Query { /** 被关联的列，一般为id，请保证和实体类中属性名称一致，驼峰下划线都可以 * 如关联的列为dept_id，填deptId和dept_id都可以*/ String column() default ""; /** 关联的属性 ，和实体类中需要封装的属性名称保持一致*/ String property() default ""; /** 执行的查询方法，建议填写mapper层的全限定方法名 ，方法返回值必须和property类型一致*/ String select() default ""; } 其次是 @Queries // // 多个关联查询注解，用于mapper层查询方法上 // 使用方法类比 @Results ，里面套多个 @Query // Queries({ // @ Query(),@Query(),... // }) // // Date: 2021/01/12 09:35 // @Documented @Retention(RetentionPolicy.RUNTIME) @Target({ElementType.METHOD}) public @interface Queries { Query[] value() default {}; } 最后是注解的具体实现 // Query注解具体实现 // // Date: 2021/01/12 09:35 // @Aspect @Component public class QueryAspect { /** Query注解配置切面织入点 */ @Pointcut("@annotation(marchsoft.annotation.Query)") public void queryPointcut() {} /** Queries注解配置切面织入点 */ @Pointcut("@annotation(marchsoft.annotation.Queries)") public void queriesPointcut() {} // // Date : 2021-01-12 09:50 // @param joinPoint 切入点 // @param result 目标方法返回值 // 方法执行后切入并修改目标方法返回值 // @AfterReturning(value = "queryPointcut()", returning = "result") public void queryAround(JoinPoint joinPoint, Object result) { MethodSignature signature = (MethodSignature) joinPoint.getSignature(); Method signatureMethod = signature.getMethod(); Query query = signatureMethod.getAnnotation(Query.class); //如果目标方法返回类型为 Collection的实现类，如List,则通过迭代器遍历 //如果目标方法返回类型为 IPage，则先获取getRecords()的List，再通过迭代器遍历 if (result instanceof Collection || result instanceof IPage) { Iterator iter; if (result instanceof Collection) { iter = ((Collection) result).iterator(); }else { iter = ((IPage) result).getRecords().iterator(); } while(iter.hasNext()){ execute(query, iter.next()); } }else { //如果目标方法返回类型即为实体类 execute(query, result); } } @AfterReturning(value = "queriesPointcut()", returning = "result") public void queriesAround(JoinPoint joinPoint, Object result) { MethodSignature signature = (MethodSignature) joinPoint.getSignature(); Method signatureMethod = signature.getMethod(); Queries queries = signatureMethod.getAnnotation(Queries.class); if (result instanceof Collection || result instanceof IPage) { Iterator iter; if (result instanceof Collection) { iter = ((Collection) result).iterator(); }else { iter = ((IPage) result).getRecords().iterator(); } while(iter.hasNext()){ Object o = iter.next(); //针对每个实体处理多个@query for (Query query : queries.value()) { execute(query, o); } } }else { for (Query query : queries.value()) { execute(query, result); } } } // // Date : 2021-01-11 15:38 // @param query 注解query // @param result 目标方法返回值 // private void execute(Query query, Object result) { //获取需要被赋值的属性 String property = query.property(); //获取需要执行的方法 String select = query.select(); //获取关联的列，并转驼峰 String column = StringUtils.toCamelCase(query.column()); //关联属性的类型（也为select的参数类型） Class<?> columnType; //关联列的值 Serializable columnValue; //select方法返回值类型 Class<?> selectMethodReturnType; //select方法返回结果 Object selectResult; try { //目标方法返回值类型class Class<?> resultClass = result.getClass(); //通过反射获取get方法并调用来获取关联列的值 try { Method getMethod = resultClass.getMethod("get" + column.substring(0, 1).toUpperCase() + column.substring(1)); columnValue = (Serializable) getMethod.invoke(result); columnType = getMethod.getReturnType(); }catch (NoSuchMethodException get) { throw new BadRequestException("找不到get方法 : " + get.getMessage() + "，请检查column参数" + column + "是否填写正确"); } //通过反射获取该select方法所在的类 int pointIndex = select.lastIndexOf("."); Class<?> selectMethodClass = Class.forName(select.substring(0, pointIndex)); //通过反射获取需要调用的select方法 try{ Method selectMethod = selectMethodClass.getMethod(select.substring(pointIndex+1), columnType); //调用并获取select方法的结果 selectResult = selectMethod.invoke(SpringContextHolder.getBean(selectMethodClass), columnValue); selectMethodReturnType = selectMethod.getReturnType(); }catch (NoSuchMethodException selectExc) { throw new BadRequestException("找不到select方法 : " + selectExc.getMessage() + "，请检查select方法的名称和参数类型"); } //通过反射获取set方法并调用来完成赋值 try{ Method setMethod = resultClass.getMethod("set" + property.substring(0, 1).toUpperCase() + property.substring(1), selectMethodReturnType); setMethod.invoke(result, selectResult); }catch (NoSuchMethodException set) { throw new BadRequestException("找不到set方法 : " + set.getMessage() + "，请检查" + property + "的类型与方法" + select + "的返回值是否一致"); } }catch (Exception e) { e.printStackTrace(); } } } 其中字符串转驼峰的toCamelCase方法来自StringUtils工具类，获取调用select中需要传入的service/mapper实例通过SpringContextHolder来获取。 其中对IPage的判断是使用了mybatis plus的分页插件的情况下需要考虑的一种情况，如果没有使用mybatis plus或者是其分页插件，把对IPage的判断去除即可。 @Select("SELECT id,dept_id,username,nick_name,gender,phone,email,avatar_path,password," + "is_admin,enabled,create_by,update_by,pwd_reset_time,create_time,update_time" + " FROM sys_user u WHERE u.id = #{id} AND is_deleted=0") @Results({ @Result(column = "id", property = "id"), @Result(column = "dept_id", property = "deptId"), }) @Queries({ @Query(column = "id", property = "roles", select = "marchsoft.modules.system.mapper.RoleMapper.findWithMenuByUserId"), @Query(column = "id", property = "jobs", select = "marchsoft.modules.system.mapper.JobMapper.findByUserId"), @Query(column = "dept_id", property = "dept", select = "marchsoft.modules.system.mapper.DeptMapper.selectById") }) UserBO findUserDetailById(Long id); 加缓存方式与原先一致： @Component @CacheConfig(cacheNames = "dept") public interface DeptMapper extends BasicMapper<Dept> { @Cacheable(key = "'id:' + #p0") Dept selectById(Long id); } 无论返回结果是UserBO，还是List，还是Set,还是IPage，最终的查询结果都是和原先使用@One与@Many的时候是一致的，而且缓存存储结构与以前也是一样的： 总结 上述就是目前想到并实现的最佳缓存解决方案，即具备了原先@One、@Many注解的开发效率，又保留了缓存的存储结构，虽然仍然需要手动维护缓存，但和学长一致认为这是合理的，谁对哪一个方法加了缓存，就有义务自己去维护他，况且难度也比较低。 但上述注解仅仅是实现了原有@One、@Many的最基本的查询封装功能，类似于@Many的fetchType = FetchType.EAGER/LEAZY等功能还需后续不断继续完善。 自此，长达数月的缓存大问题拉下帷幕。 缓存的添加与维护 springboot缓存注解的复习————SpringBoot2.0的@Cacheable(Redis)缓存失效时间解决方案 根据上述所说可知，我们是把缓存注解加在了mapper层，所以首先我要说明同时在service层和mapper加缓存时的问题： [scode type="blue"]service与mapper加相同的命名空间时，即CacheConfig(cacheNames = "xx")相同，此时会将service和mapper缓存合并，但是如果出现service和mapper中缓存的键名相同，即@Cacheable(key = "XXX")相同，此时service层的覆盖mapper层的缓存。[/scode] 所以如果出现了缓存加在service和mapper层都可以的情况，我们考虑优先加在mapper层（考虑之后其他的mapper层可能会通过@Query调用）。但是service层也要加命名空间CacheConfig(cacheNames = "xx")，考虑有可能会在service层加其他缓存。但是删除缓存的操作均在service层。 下面我以用户user与岗位job（多对多）两个实体类来详细说明缓存的添加与维护。 情况一：单一实体缓存，key和value都仅与该实体相关(1:1) 首先是针对一个实体的单条信息进行缓存，key和value都和一个实体有关。如缓存job表的每一条记录，以id和键名，Job实体为value存入缓存： [scode type="blue"]在JobMapper加上命名空间 @CacheConfig(cacheNames = "job")[/scode] 首先在JobMapper加上命名空间： @CacheConfig(cacheNames = "job") public interface JobMapper extends BasicMapper<Job> { 虽然BaseMapper已经提供了根据id查询—— T selectById(Serializable id); [scode type="blue"]重写selectById，并加上@Cacheable[/scode] 但是咱们需要加上缓存，所以需要重新在mapper层写该方法（无需写sql），并加上@Cacheable，此时键名key的命名规则为统一的 "'id:' + #p0" ，id指Job的id，#p0指方法的第一个参数： @Cacheable(key = "'id:' + #p0") Job selectById(Long id); 此时springboot会将第一查询的结果放入缓存，之后无论是从service调mapper的该方法，还是从mapper的@Query中调用该方法都可以走缓存。 缓存情况如下图： [scode type="blue"]将键名除id外以静态变量存入CacheKey[/scode] 可以看到我们成功把id为1的job存入了redis缓存，该缓存的value即为Job实体类，键名为 job::id:1 ，其中为了后期维护缓存方便，我们将字符串 "job::id:" 作为静态变量存入 CacheKey.class中，在做维护时只需拼接一个id即可： public interface CacheKey { /** * 用户 */ String USER_ID = "user::id:"; /** * 数据 */ String DATA_USER = "data::user:"; /** * 菜单 */ String MENU_ID = "menu::id:"; String MENU_USER = "menu::user:"; String MENU_ROLE = "menu::role:"; /** * 角色授权 */ String ROLE_AUTH = "role::auth:"; String ROLE_USER = "role::user:"; /** * 角色信息 */ String ROLE_ID = "role::id:"; /** * 部门信息 */ String DEPT_ID = "dept::id:"; String DEPT_ROLE = "dept::role:"; /** * 岗位信息 */ String JOB_ID = "job::id:"; //示例 String JOB_USER = "job::user:"; } [scode type="blue"]在对应的service层手动维护缓存[/scode] 这里为了统一建议手动维护缓存，即在JobServiceImpl下手写一个方法来清理缓存，在相应的更新、删除Job的地方调用该方法即可： /** * 清理缓存 * @param id job_id */ private void delCaches(Long id){ redisUtils.del(CacheKey.JOB_ID + id); } 你也可以在相应的更新、删除的方法上加上springboot删除缓存的注解（不推荐，后期缓存情况复杂） @CacheEvict(key = "'id:' + #p0.id") public void update(Job resources) {} [scode type="yellow"]对于单一实体的缓存，仅缓存 selectById方法，对于该实体的分页查询等其他条件查询一般不做缓存，除非有特殊要求[/scode] 情况二：缓存内容关联两个及以上的实体（1:n,n:m） 这类缓存是比较令人头疼的地方，我们之前有规定，@Query中所调用的方法必须要有缓存，这里user与job是多对多的关系，有一张中间sys_users_jobs，当我们在查询一个用户时要连带将其所有的岗位也查找出来： //UserMapper中查询一个用户 @Select("SELECT id,dept_id,username,nick_name,gender,phone,email,avatar_path,password," + "is_admin,enabled,create_by,update_by,pwd_reset_time,create_time,update_time" + " FROM sys_user u WHERE u.id = #{id} AND is_deleted=0") @Results({ @Result(column = "id", property = "id"), @Result(column = "dept_id", property = "deptId"), }) @Queries({ @Query(column = "id", property = "roles", select = "marchsoft.modules.system.mapper.RoleMapper.findWithMenuByUserId"), @Query(column = "id", property = "jobs", select = "marchsoft.modules.system.mapper.JobMapper.findByUserId"), //这里 @Query(column = "dept_id", property = "dept", select = "marchsoft.modules.system.mapper.DeptMapper.selectById") }) @Cacheable(key = "'id:' + #p0") UserBO findUserDetailById(Long id); 既然通过@Query调用了JobMapper.findByUserId方法，我们就需要对其加缓存（其他的同理） //JobMapper中根据用户id查询岗位 @Select("SELECT j.id, j.name, j.enabled, j.job_sort, j.create_by, j.update_by, j.create_time, j.update_time " + "FROM sys_job j, sys_users_jobs uj WHERE j.id = uj.job_id AND uj.user_id = ${id} AND j.is_deleted=0") @Cacheable(key = "'user:' + #p0") Set<Job> findByUserId(Long id); 此时缓存的命名规则为 key = "'user:' + #p0" ，这里的user指user_id的意思，后面的参数#p0就是user_id，我们可以发现此时的redis里缓存全称为： job::user:2 意思即为user_id为2的用户所拥有的job，从value中可以看出该用户有两个job。我们也需要将 "job::user:" 存入 CacheKey中，命名为 JOB_USER （可以回看上面的）。 [scode type="blue"]从每个与该缓存相关的实体的service层对其进行维护[/scode] 以 job::user:2 缓存维护为例 [scode type="blue"]Job发生删改时，可能需要对改缓存进行维护[/scode] 当我们修改了一个id为1的job，可能id为2的user有这个job，也可能没有，有的话必须删除该缓存，即我们需要通过sys_users_jobs关系表找到含有job_id=2的user_id的集合，是一个List： //UserMapper下 //根据job_id查询用户Id (清理job缓存时调用) @Select("SELECT user_id FROM sys_users_jobs WHERE job_id = #{id} group by user_id") List<Long> findByJobId(Long id); 注意，这里是因为job的变动导致需要清理该缓存，所以该部分清理缓存的代码写在 JobServiceImpl下的delcache方法中: //JobServiceImpl private void delCaches(Long id){ List<Long> userIds = userMapper.findByJobId(id); redisUtils.delByKeys(CacheKey.JOB_USER, new HashSet<>(userIds)); //第二种情况所删的缓存 redisUtils.del(CacheKey.JOB_ID + id); //第一种情况所删的缓存 } [scode type="blue"]当中间表sys_users_jobs发生变动时，一定要清理缓存！[/scode] 当我们修改了user的job时，如上图user_id = 2的用户有两个job，我们将其调整为一个job，此时也需要删除 job::user:2 这条缓存，由于是因为更新了user才导致删除这条缓存，所以该部分代码写在 UserServiceImpl的更新、删除中: //UserServiceImpl下的更新user方法 public void updateUserWithDetail(UserInsertOrUpdateDTO userInsertOrUpdateDTO) { //... //如果岗位发生变化 if (! CollectionUtils.isEqualCollection(jobIds, userInsertOrUpdateDTO.getJobs())) { //... //清除缓存 redisUtils.del(CacheKey.JOB_USER + userInsertOrUpdateDTO.getId()); } //... } 这样，我们才算是对 job::user:2这条缓存做了完全的维护。 [scode type="yellow"]对于情况一的缓存只涉及一个实体（表），缓存维护比较好实现，针对第二种情况需要我们考虑全面，在哪些地方需要去维护，这个需要考虑清除，不要随意加缓存[/scode] [scode type="red"]谁加了哪个缓存，谁就有义务去维护它！[/scode]