1 redis概述

1.1 什么是redis

  Redis的全称是REmote Dictionary Server。redis 是一种基于对（key-value）的NoSQL数据库。redis所有数据都存放在内存中，所以Redis的读写速度非常惊人，还可以将内存中的数据以快走和日志的形式保存在硬盘上，这样断电等机器故障不会使内存中的数据“丢失”。redis还提供键过期、发布订阅、事务、流水线、Lua脚本等功能。

1.2 redis的作者是谁

   redis的作者Salvatore Sanfilippo 在开发一个叫LLOOGG的网站时需要实现一个高性能的队列功能，一开始是使用MySql 来现实，后来发现无论怎么优化SQL语句都不能使网站的性能提高上去，于是他决定自己做一个专属于LLOOGG的数据库，这个就是Redis的前身。后来，Salvatore Sanfilippo将Redis1.0的源码开放到GitHub上，可能连他自己都没想到，Redis后来如此受欢迎。

1.3 有谁在用redis

  从Redis的官方公司统计来看，有很多重量级的公司都在使用Redis，如国外的Twitter、Instagram、Stack Overflow、GitHub等，国内就更多了，如果单单从体量来统计，新浪微博可以说是全球最大的Redis使用者，除了新浪微博，还有像阿里巴巴、腾讯、百度、搜狐、优酷土豆、美团、小米、唯品会等公司都是Redis的使用者。

2 redis 和 mySQL的区别

2.1 什么是mySQL 数据库

1. mySQL是一个关系型数据库管理系统，由瑞典MySQL AB公司开发，目前属于Oracle公司。
2. mySQL是一种关联数据库管理系统，关联数据库将数据保存在不同的表中，而不是将所有数据放在一个大仓库内，这样就增加了速度并提高了灵活性。
3. mySQL是开源的，Mysql支持大型的数据库。可以处理拥有上千万条记录的大型数据库。MySQL使用标准的SQL数据语言形式。
4. mySQL可以允许于多个系统上，并且支持多种语言。这些编程语言包括C、C++、Python、Java、Perl、PHP、Eiffel、Ruby和Tcl等。

2.2 两者的区别

  redis是内存数据库，数据保存在内存中，读写速度快。
  mySQL是关系型数据库，功能强大，数据访问也就慢。

3 redis特性

3.1 速度快

  正常情况下，Redis执行命令的速度非常快，官方给出的数字是读写性能可以达到10万/秒。原因可以大致归纳为以下四点：
  1. Redis的所有数据都是存放在内存中的，所以把数据放在内存中是Redis速度快的最主要原因。
  2. Redis是用C语言实现的，一般来说C语言实现的程序“距离”操作系统更近，执行速度相对会更快。
  3. Redis使用了单线程架构，预防了多线程可能产生的竞争问题。

3.2 基于键值对的数据结构服务器

  Redis的全称是REmote Dictionary Server，它主要提供了5种数据结构：字符串、哈希、列表、集合、有序集合，同时在字符串的基础之上演变出了位字符串、哈希、列表、集合、有序集合，同时在字符串的基础之上演变出了位图（Bitmaps）和HyperLogLog两种神奇的“数据结构”，并且随着LBS（Location Based Service，基于位置服务）的不断发展，Redis3.2版本中加入有关GEO（地理信息定位）的功能。

3.3 丰富的功能

  除了5种数据结构，Redis还提供了许多额外的功能：
    ·提供了键过期功能，可以用来实现缓存。
    ·提供了发布订阅功能，可以用来实现消息系统。
    ·支持Lua脚本功能，可以利用Lua创造出新的Redis命令。
    ·提供了简单的事务功能，能在一定程度上保证事务特性。
    ·提供了流水线（Pipeline）功能，这样客户端能将一批命令一次性传到Redis，减少了网络的开销。

3.4 简单稳定

  Redis的简单主要表现在三个方面。首先，Redis的源码很少，早期版本的代码只有2万行左右，3.0版本以后由于添加了集群特性，代码增至5万行左右。其次，Redis使用单线程模型，这样不仅使得Redis服务端处理模型变得简单，而且也使得客户端开发变得简单。最后，Redis不需要依赖于操作系统中的类库，Redis自己实现了事件处理的相关功能。
  Redis虽然很简单，但是不代表它不稳定。以笔者维护的上千个Redis为例，没有出现过因为Redis自身bug而宕掉的情况。

3.5 客户端语言多

  Redis提供了简单的TCP通信协议，很多编程语言可以很方便地接入到Redis，并且由于Redis受到社区和各大公司的广泛认可，所以支持Redis的客户端语言也非常多，几乎涵盖了主流的编程语言，例如Java、PHP、Python、C、C++、Nodejs等。

3.6 持久化

  通常看，将数据放在内存中是不安全的，一旦发生断电或者机器故障，重要的数据可能就会丢失，因此Redis提供了两种持久化方式：RDB和AOF，即可以用两种策略将内存的数据保存到硬盘中（，这样就保证了数据的可持久性。

3.7 主从复制

  Redis提供了复制功能，实现了多个相同数据的Redis副本，复制功能是分布式Redis的基础。

3.8 高可用和分布式

  Redis从2.8版本正式提供了高可用实现Redis Sentinel，它能够保证Redis节点的故障发现和故障自动转移。Redis从3.0版本正式提供了分布式实现Redis Cluster，它是Redis真正的分布式实现，提供了高可用、读写和容量的扩展性。

4 安装Redis

  安装参考 https://www.jianshu.com/p/150307f7e43f

4.1 安装依赖

yum install gcc gcc-c++ make
yum install wget

4.2 下载redis软件安装包

wget http://download.redis.io/redis-stable.tar.gz

4.3 解压安装

tar xvzf redis-stable.tar.gz
cd redis-stable
mv redis-stable /usr/local/redis
cd /usr/local/redis
make
make install

编译过程中可能会有如下报错：

在包含自 adlist.c：34 的文件中:
zmalloc.h:50:31: 错误：jemalloc/jemalloc.h：没有那个文件或目录
zmalloc.h:55:2: 错误：#error "Newer version of jemalloc required"
make[1]: *** [adlist.o] 错误 1
make[1]: Leaving directory `/usr/local/redis/src'

需要执行如下命令：

make MALLOC=libc && make install

2.4 修改配置文件

复制配置文件

cp redis.conf /etc/

修改配置文件

vim /etc/redis.conf

配置修改如下：

    pidfile /var/run/redis.pid  
    bind 127.0.0.1 本机ip #bind ip,绑定本机ip即可
    daemonize yes       #后台运行
    loglevel notice 
    logfile /data/logs/redis/redis.log  
    #设置log目录，需手动创建/data/logs/redis/目录，mkdir /data/logs/redis/
    dir /data/redis/
    #设置文件目录，需手动创建mkdir /data/redis/
    databases 16 
    save 900 1 
    save 300 10 
    save 60 10000 
    rdbcompression yes 
    dbfilename dump.rdb #设置db文件
    appendonly no

创建数据目录 并更改权限

mkdir /data/redis/
chmod -R 755 /data

4.5 验证是否安装成功

使用 /etc/redis.conf 配置文件启动redis

redis-server /etc/redis.conf 

查看端口：

netstat -ntlp |grep 6379

image.png

如果显示端口正在使用则说明安装成功，没有就没安装成功。
链接redis 测试：

/usr/local/redis/src/redis-cli

存值

set keyTest "hello"

取值

get keyTest

image.png

关闭redis：

/usr/local/redis/src/redis-cli shutdown

5 redis API

5.1 常用命令

1. 查看所有键
   keys pattern
   注：keys 后面接查找表示式，在库中有大量key时，使用keys * 需要谨慎，可能会引发线程阻塞。
2. 查看所有键
   dbsize
3. 查看所有键
   exists key
4. 删除键
   del key [key ...]
   可同时删除多个key，返回成功删除键的个数
5. 键过期
   expire key seconds
   设置键的过期时间，当键过期后，用 get key 取键值，会返回nil。
6. 键过期查询命令
   ttl key
   返回值为大于等于0的整数：键过期的剩余时间。
   返回值为-1: 键没有过期时间。
   返回值为-2: 键以过期或不存在。
7. 键的结构类型
   type key
8. 设置键值
   set key value [ex seconds] [px milliseconds] [nx|xx]
   [ex seconds] 为秒级过期时间
   [px milliseconds] 为毫秒级过期时间
   [nx] 当键不存在时才设置，用于添加
   [xx] 当键存在时才设置，用于更新
9. 获取键值
   get key
10. incr 自增计数
    incr key
    注：仅对整数有效，否则返回错误
11. decr 自减计数
    decr key
    注：仅对整数有效，否则返回错误
12. 键重命名
    rename oldkey newkey
13. 迁移键
    move key db
    把键移动db库

6 主从复制

6.1 建立主从方式

建立主从复制方式有三种：

1. 在从redis配置文件中加入 slaveof {masterHost} {masterHost}，redis启动生效。
2. 启动从redis后，在命令中执行 --slaveof {masterHost} {masterHost}。
3. 直接使用 slaveof {masterHost} {masterHost}。

查看主从复制状态：info replication

6.2 断开复制

在从节点上执行命令 ：slaveof no one
切换主方式：slaveof {newMaster}
注：切换主节点后从节点上的数据会被清空，线上慎用 slaveof命令。

6.3 复制原理

复制流程大致分为一下步骤：

1. 保存master信息
   slave 上执行slaveof 命令后，便把master 的信息保存起来。 master 的 master_link_status 变为下线状态。
2. slave 内部通过每秒运行的定时任务维护复制相关逻辑任务，发现新master 节点后，会尝试与该节点建立网络链接。
3. 发送ping命令
   发送ping命令，当master 回复 pong 时，说明建立成功。
4. 权限验证
5. 数据集同步
   在和master 正常通信后，首次建立链接，master会把持有的所有数据发给slave，这次为全量同步。
6. 命令持续复制
   在首次与master 同步完成后，后续master会持续把写命令发送给slave，保证数据一致性，为部分同步。

6.4 心跳

在主从建立复制后，他们之间维护着长链接并彼此发送心跳命令。

1. 主从彼此都有心跳检测机制，可以通过client list命令查看信息，主节点链接状态为 flags=N，从节点链接状态为flags=S。
2. 主节点默认每 10s 对从节点发送ping命令。可通过 repl-ping-slave-period 控制发送频率。
3. 从节点每 1s 给主节点发送 replconf ack {offset} 命令，上报自身当前的复制偏移量。

7 内存管理与优化

7.1 内存管理

7.1.1 设置内存上限

maxmemory 参数可限制最大可使用内存，主要目的：

1. 用于缓存场景，当超出内存上限时，使用LRU等删除策略释放空间。
2. 防止所用内存超过服务器内存。

注：由于内存碎片率的存在，实际消耗的内存可能会比maxmemor设置的大。

7.1.2 动态调整内存上限

config set maxmemory 1GB
该命令可动态调整内存上限

注：为了防止极端情况下导致系统内存被耗尽，建议每个redis进程都设置maxmemory参数，默认无限大。

7.1.3 内存回收策略

内存回收策略体现在以下两个方面：
. 删除到期键。
. 内存使用达到maxmemory上限触发控制策略。

1. 删除过期键
   1）惰性删除
     redis 读取带过期属性的键时，当该键已过期，会执行删除命令并返回空。
   2）定时任务删除
      redis 内部维护一个定时任务，默认每秒运行10次，可通过 hz 参数配置。每次任务随机检查20个键，当发现过期键时删除对应键；
      如果超过25%的键过期，循环执行回收逻辑直到不足25%或超时为止，默认超时时间为25ms；该次为慢模式。
     如果回收逻辑超时，则在触发内部事件之前再次以快模式回收过期键。
     快慢两种模式删除逻辑相同，只是执行超时时间不同。慢模式超时为25ms，快模式超时为 1ms 且每 2 s内只能运行一次。
2. 内存溢出控制策略
   当redis内存使用达到maxmemory上限触发控制策略。
   redis 支持 6种控制策略，由maxmemory-policy参数控制。
   1）noeviction：默认策略。不删除任何数据，redis变为只读模式，写入报错。
   2）volatile-lru：根据LRU算法删除设置超时属性的键，直到空间足够为止。没有可删除键时，回退到noeviction策略。
   3）allkeys-lru：根据LRU算法删除键，不管键有没有设置超时属性，直到空间足够为止。
   4）allkeys-random：随机删除所有键，直到空间足够为止。
   5）volatile-random：随机删除过期键，直到空间足够为止。
   6）volatile-ttl：根据值的ttl属性，删除将要过期的数据。没有可删除键，回退到noeviction策略。

内存溢出控制策略 可采用 config set maxmemory-policy {policy} 动态设置。

7.2 内存优化

redis 内存优化的常用的方法有：

1. redisObject对象
2. 缩减键值对象
3. 共享对象池
4. 字符串优化
5. 编码优化
6. 控制键的数量

8 redis cluster 集群

8.1 cluster 数据分布

8.1.1 分布式数据分区规则

  分布式数据库会把整个数据集按照分区规则将数据集划分到多个节点上，每个节点负责一部分数据集。
  常用的分区规则由哈希分区和顺序分区。redis cluster 采用的是哈希分区规则。
常见的哈希分区规则有以下几种：

1. 节点取余分区
     将key（键数量）% N（节点数）计算出哈希值，决定该数据映射到哪个节点上。
     优点：简单。
     缺点：当节点数 N 发生变化时，映射关系会重新计算，可能会导致数据迁移。
   为了避免数据迁移，通常节点数扩张或缩减会成倍数关系。
2. 一致性哈希分区
     思路是为系统中每个节点分配一个token，范围一般在0～，这写token构成一个哈希环。数据读写时，先根据key计算hash值，然后顺时针找到第一个大于等于该哈希值的token节点。
3. 虚拟槽分区
     虚拟槽分区使用哈希空间，用哈希函数把所有数据映射到一个固定的范围整数集合中，整数定义为槽（slot）。这个范围远大于节点数。槽是集群内数据管理和迁移的基本单位。
     redis cluster 槽的范围 0～16383。当有五个节点时，每个节点大约负责管理3276个槽。

8.1.2 redis cluster 数据分区

  redis cluster采用虚拟槽分区规则，所有键根据哈希函数映射到0～16383个槽上。计算公式：slot=CRC16(key)&16383。

redis 虚拟槽分区特点：

1. 数据和节点解藕，简化节点扩容和缩减难度。
2. 不需要客户端或代理服务维护槽元数据，自身就能完成。
3. 支持节点、槽、键之间的映射查询。

8.2 redis cluster 功能限制

相对单例，cluster上存在以下功能限制：

1. 只支持相同slot值的key 执行批量操作，如mset、mget。
2. 只支持多key 在同一个节点上的事务操作。
3. 不能将大键值对象如hash、list映射到不同节点上。
4. 不支持多数据库空间，只有一个db0 数据库。单例有db0～db15 共16个数据空间。
5. 复制结构只支持一层。

8.3 redis cluster 创建

8.3.1 服务器配置说明

以三主三从为例：
    只有三台服务器，所以每台服务上创建两个redis 实例，用作一主一从。

注：需每台服务器上先安装好redis服务，可参考步骤4 安装Redis

8.3.2 安装依赖

安装ruby
    yum install ruby
安装 gem
    yum install rubygems
安装 gem  
    gem install redis

注：redis 安装需要ruby 版本大于 2.2.2

升级ruby方法可参考 https://www.jianshu.com/p/a1a4d59490d7

8.3.3 创建实例

注：以一台服务器为例，主从实例的端口分别为 6379、6479
复制配置文件

cp /usr/local/redis/redis.conf /etc/redis_6379.conf
cp /usr/local/redis/redis.conf /etc/redis_6479.conf

注：每台服务器应先装好redis服务，

修改配置文件，修改项如下：

port 6379       #对应实例的端口
pidfile /var/run/redis.pid  
bind 127.0.0.1 本机ip #bind ip,绑定本机ip即可
daemonize yes

loglevel notice 

logfile /data/logs/redis/redis_6379.log 
#设置log目录，需手动创建/data/logs/redis目录，mkdir /data/logs/redis

dir /data/redis/redis_6379/
#设置文件目录，需手动创建mkdir /data/redis/redis_6379/

databases 16 
save 900 1 
save 300 10 
save 60 10000 
rdbcompression yes 
dbfilename dump_6379.rdb
#设置db文件
appendonly no

8.3.4 启动实例

使用对应的配置文件启动

redis-server /etc/redis_6379.conf
redis-server /etc/redis_6479.conf

查看线程，端口验证是否启动正好

ps -ef|grep redis
netstat -tnlp | grep redis

8.3.5 创建集群

/usr/local/redis/src/redis-trib.rb create --replicas 1 主机1:6379 主机2:6379 主机3:6379 主机1:6479 主机2:6479 主机3:6479

注:只需在一台服务器上创建集群。redis-trib.rb 会尽可能的将主从节点分配在不同的机器上，所以顺序会重新分配。

8.3.6 验证集群

链接集群，加 –c 参数意味链接集群

redis-cli -h 172.31.78.3 -c -p 6379

8.3.7 集群完整性检查

redis-trib.rb check 172.31.78.3:6379

该命令可检查集群的完整性，只需检查任意一个节点。返回结果会列出集群节点信息

8.4 节点通信

8.4.1 通信流程

1. 集群每个节点都会开辟一个TCP通道，用于节点间通信。通信端口为基础端口上加10000.
2. 每个节点在固定周期内通过特定规则选择几个节点发送ping消息。
3. 收到ping消息的节点回复pong消息作为响应。

注：因为通信端口 = 基础端口 + 10000。所以如果基础端口为 6379，这防火墙需要放开 6379 和 16379 这两个端口才可让集群正常通信。

8.4.2 节点选择

  redis cluster 节点内通信采用固定频率，每秒执行10次定时任务。ping/pong消息携带当前节点和其他节点的状态数据，频繁通信势必会加重计算和带宽的负担。所以每次定时任务选择通信的节点非常重要。

1. 选择发送信息的节点数量
     集群每秒会随机选取5个节点，找出最久没通信的节点发送ping消息。每100毫秒会扫描本地节点列表，发现节点最近接受pong消息的消息大于cluster_node_timeout/2 时会立即发送ping消息。所以每个节点每秒发送的ping消息量 = 1 + 10 * num (node.pong_receives>cluster_node_timeout/2)；
2. 消息数据量
     每个ping消息的数据体现在消息头和消息体中，消息头的大小为 2KB，消息体需看伪代码。

8.5 伸缩

新 node 加入集群方法。
可使用redis-trib.rb 工具直接添加。

// 添加新节点
redis-trib.rb add-node newhost1:port newhost2:port 

//添加新节点并设置为master-id的从节点
redis-trib.rb add-node newhost:port --slave --master-id  

8.6 请求重定向

  redis 接受到key命令时先计算key对应的槽，再找出槽对应的节点。如果节点是本身就处理命令，否则回复MOVED重定向错误，通知客户端请求正确节点。

image.png

8.6 故障转移

8.6.1 故障发现

  redis 通过ping/pong消息实现节点通信，同时携带节点的消息，如主从状态、节点故障、槽信息等。

  每个节点定期会给其他节点发送ping消息，如果在 cluster-node-timeout时间内收不到其他节点回复的pong消息，则会将该节点标记为主观下线状态（pfail）。
  当某个节点判断另一个节点为主观下线后，会将相应节点状态信息在集群内传播。当超过半数的节点的任务该节点下线时，会将该节点标记为客观下线状态。这时该节点就会下线。

注：如果在cluster-node-time*2 时间内无法收集到一半以上节点的下线报告，那么之前的下线报告将会过期。也就是说主观下线报告永远赶不上客观下线报告，那么故障节点就不会被标记客观下线而导致故障转移失败。所以cluster-node-time不建议设置过小。

8.6.2 故障恢复

   当故障节点变为客观下线后，如果该节点为主节点则需要在从节点中选出一个节点替补上。
恢复流程如下：

1. 资格检查
   如果从节点与主节点断线的时间超过 cluster-node-time*cluster-slave-validity-factor，则该节点没有资格。cluster-slave-validity-factor 为从节点有效因子，默认为10。
2. 准备选举时间
   当从节点拥有资格后，更新触发选举时间（failover_auth_time），达到该时间继续后续流程。
3. 发起选举
   发起选举流程：
   1）更新配置纪元
   2）广播选举消息

4. 选举投票
   只有主节点才会处理故障选举消息。

   
   
   image.png
5. 替换主节点
   1）从节点取消复制变为主节点
   2）执行clusterDelSlot操作撤销故障主节点负责的槽，并执行clustrAddSlot把这些槽给自己
   3）在集群中广播自己pong消息，说自己登基了，不是备胎了。

8.6.3 故障转移时间

故障发现到故障恢复花销的时间大致可以由以下估计：

1. 主观下线识别时间 = cluster-node-timeout
2. 主观下线传播时间 <= cluster-node-timeout/2
3. 从节点转移时间 <= 1000毫秒

注：cluster-node-timeout 的默认时间为 15秒

8.7 集群运维

8.7.1 集群完整性

集群上每个槽（slot）都需要有一个非故障的主节点管理负责，否则该集群会变为不可用。

8.7.2 集群倾斜

集群倾斜是指不同节点间数据量和请求量出现明显差异，这会增大负载均衡和开发运维难度。

1. 数据倾斜
   1）节点和槽分配严重不均。
   2）槽对应键的数量差异过大。
   3）集合对象包含大量元素。
   4）内存相关配置不一致。
   针对 1）可使用redis-trib.rb info host:ip 命令定位。
   针对 2）可使用cluster countkeysinslot {slot} 先获取对应槽的键数，在 使用 cluster getkeysinslot {slot} {count} 循环迭代出槽下所以键。
   针对 3）可使用redis-cli --bigkeys 识别大集合对象，在进行数据转移。

8.7.3 数据迁移

将单机redis数据转移到集群环境下：

redis-trib.rb import host:port --form <arg> --copy --replace

注：

1. 只能从单机转移值集群。
2. 不支持在线转移，数据提供方应先停止读写。
3. 不支持断点续传。
4. 单线程进行数据迁移，大数据量时迁移速度慢。

8.7 springboot + jedis + RedisTemplate java客户端接入集群demo

配置

// 从配置文件中获取集群配置信息
   @Value("${spring.redis.cluster.nodes}")
    private String clusterNodes;  // 节点格式 “host1:port,host2:port,host3:port,host4:port”
    @Value("${spring.redis.timeout}")
    private String timeOut;
    @Value("${spring.redis.password}")  // 集群密码
    private String password;
    @Value("${spring.redis.jedis.pool.max-active}")
    private int maxActive;
    @Value("${spring.redis.jedis.pool.max-wait}")
    private long maxWait;
    @Value("${spring.redis.jedis.pool.max-idle}")
    private int maxIdle;
    @Value("${spring.redis.jedis.pool.min-idle}")
    private int minIdle;

    @Bean
    public RedisTemplate<String, Object> redisTemplate() {

        Map<String, Object> source = new HashMap<String, Object>();
        source.put("spring.redis.cluster.nodes", clusterNodes);
        source.put("spring.redis.cluster.timeout", timeOut);

        RedisClusterConfiguration redisClusterConfiguration = new RedisClusterConfiguration(new MapPropertySource("RedisProperties", source));
        redisClusterConfiguration.setPassword(RedisPassword.of(password));

        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxIdle(maxIdle);
        config.setMinIdle(minIdle);
        config.setMaxWaitMillis(maxWait);
        config.setMaxTotal(maxActive);

        JedisConnectionFactory factory = new JedisConnectionFactory(redisClusterConfiguration,config);
        factory.getClientConfiguration();
        factory.afterPropertiesSet();

        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericToStringSerializer<String>(String.class));

        redisTemplate.setConnectionFactory(factory);

        return redisTemplate;
    }

使用方法

   @Autowired
    private RedisTemplate redisTemplate; // 自动注入

    /**  默认过期时长，单位：秒 */
    public final static long DEFAULT_EXPIRE = 60 * 60 * 24 * 20;
    /**  当天就过期，单位：秒 */
    public final static long DATE_EXPIRE = 60 * 60 * 24;
    /**  不设置过期时长 */
    public final static long NOT_EXPIRE = -1;

    public void set(String key, String value, long expire){
        valueOperations.set(key, value);
        if (expire != NOT_EXPIRE) {
            redisTemplate.expire(key, expire, TimeUnit.SECONDS);
        }
    }
    public void set(String key, String value){
        set(key, value, NOT_EXPIRE);
    }
    public boolean del(String key){
        return redisTemplate.delete(key);
    }
    public String get(String key, long expire) {
        String value = valueOperations.get(key);
        if(expire != NOT_EXPIRE){
           redisTemplate.expire(key, expire, TimeUnit.SECONDS);
        }
        return value;
    }
    public <T> T get(String key, Class<T> clazz, long expire) {
        String value = valueOperations.get(key);
        if (expire != NOT_EXPIRE) {
            redisTemplate.expire(key,expire,TimeUnit.SECONDS);
        }
        return value == null ? null : fromJson(value, clazz);
    }

参考文献：

1. 《Redis开发与运维》作者: 付磊 张益军