基础

Redis是完全开源的，遵守BSD协议，是一个高性能的key-value数据库。

Redis与其他key-value缓存产品有以下三个特点：

• Redis支持数据的持久化，可以将内存中的数据保存在磁盘中，重启的时候可以再次加载进行使用。
• Redis不仅仅支持简单的key-value类型的数据，同时还提供list，set，zset，hash等数据结构的存储。
• Redis支持数据的备份，即master-slave模式的数据备份。

c语言编写，内存存储，key-value保存，没有表
一秒写8w次，读11w次
NoSQL = Not Only SQL
端口 6379
默认16个数据库，默认使用第0个
启动时，先启动服务端，再启动客户端；
可视化客户端RedisDestopManager

为什么我们要用Redis？
性能极高 – Redis能读的速度是110000次/s,写的速度是81000次/s。
丰富的数据类型 – Redis支持二进制案例的 Strings, Lists, Hashes, Sets 及 Ordered Sets 数据类型操作。
原子性 – Redis的所有操作都是原子性的，意思就是要么成功执行要么失败完全不执行。单个操作是原子性的。多个操作也支持事务，即原子性，通过MULTI和EXEC指令包起来。
丰富的特性 – Redis还支持publish/subscribe, 通知, key过期等等特性。

Redis与其他key-value存储有什么不同？
Redis有着更为复杂的数据结构并且提供对他们的原子性操作，这是一个不同于其他数据库的进化路径。Redis的数据类型都是基于基本数据结构的同时对程序员透明，无需进行额外的抽象。

Redis运行在内存中但是可以持久化到磁盘，所以在对不同数据集进行高速读写时需要权衡内存，因为数据量不能大于硬件内存。在内存数据库方面的另一个优点是，相比在磁盘上相同的复杂的数据结构，在内存中操作起来非常简单，这样Redis可以做很多内部复杂性很强的事情。同时，在磁盘格式方面他们是紧凑的以追加的方式产生的，因为他们并不需要进行随机访问。

Windows安装

1.下载windows客户端

https://gitee.com/liuq1991v/redis-for-windows/tree/master

2.下载好后解压到某个目录，以d:/Programs Files/redis为例，先进入这个目录，在地址栏直接输入cmd打开命令行：输入redis-server.exe启动（或者指定配置文件redis-server xxx.conf），如果出现下图证明启动成功。

[4760] 25 Dec 10:59:48.328 # Warning: no config file specified, using the default config. In order to specify a config file use redis-server.exe /path/to/redis.conf
                _._
           _.-``__ ''-._
      _.-``    `.  `_.  ''-._           Redis 2.6.12 (00000000/0) 64 bit
  .-`` .-```.  ```\/    _.,_ ''-._
 (    '      ,       .-`  | `,    )     Running in stand alone mode
 |`-._`-...-` __...-.``-._|'` _.-'|     Port: 6379
 |    `-._   `._    /     _.-'    |     PID: 4760
  `-._    `-._  `-./  _.-'    _.-'
 |`-._`-._    `-.__.-'    _.-'_.-'|
 |    `-._`-._        _.-'_.-'    |           http://redis.io
  `-._    `-._`-.__.-'_.-'    _.-'
 |`-._`-._    `-.__.-'    _.-'_.-'|
 |    `-._`-._        _.-'_.-'    |
  `-._    `-._`-.__.-'_.-'    _.-'
      `-._    `-.__.-'    _.-'
          `-._        _.-'
              `-.__.-'

[4760] 25 Dec 10:59:48.336 # Server started, Redis version 2.6.12
[4760] 25 Dec 10:59:48.336 * The server is now ready to accept connections on port 6379

3.启动另一个命令行界面，在该Redis目录下，使用命令：redis-cli.exe 启动客户端,连接服务器。出现下图效果，证明连接服务器成功。

C:\Environment\redis64-2.6.12.1>redis-cli.exe
redis 127.0.0.1:6379>

精简模式：redis-cli.exe
指定模式：redis-cli.exe -h 127.0.0.1 -p 6379 -a requirepass
(-h 服务器地址  -p 指定端口号 -a 连接数据库的密码[可以在redis.windows.conf中配置]，默认无密码)

4.由于上面虽然启动了Redis服务，但是，只要一关闭命令行窗口，Redis服务就关闭了。所以，把Redis设置为一个Windows服务。安装之前，Windows服务是不包含Redis服务的。

5.安装为Windows服务

安装命令: redis-server.exe --service-install redis.conf 使用命令

why

C:\Environment\redis64-2.6.12.1>redis-server.exe --service-install

*** FATAL CONFIG FILE ERROR ***
Reading the configuration file, at line 2
>>> 'service-install'
Bad directive or wrong number of arguments

C:\Environment\redis64-2.6.12.1>redis-server.exe --service-install redis.conf

*** FATAL CONFIG FILE ERROR ***
Reading the configuration file, at line 2
>>> 'service-install "redis.conf"'
Bad directive or wrong number of arguments

在打开的运行窗口中，输入命令services.msc可以查看是否确实开启

卸载服务：redis-server --service-uninstall
开启服务：redis-server --service-start
停止服务：redis-server --service-stop

redis - 无注释版.conf

daemonize no

pidfile /var/run/redis.pid

port 6379

timeout 0

loglevel notice

logfile stdout

databases 16

save 900 1
save 300 10
save 60 10000

stop-writes-on-bgsave-error yes

rdbcompression yes

rdbchecksum yes

dbfilename dump.rdb

dir ./

slave-serve-stale-data yes

slave-read-only yes

slave-priority 100

appendonly no

appendfsync everysec

no-appendfsync-on-rewrite no

auto-aof-rewrite-percentage 100
auto-aof-rewrite-min-size 64mb

lua-time-limit 5000

slowlog-log-slower-than 10000

slowlog-max-len 128

hash-max-ziplist-entries 512
hash-max-ziplist-value 64

list-max-ziplist-entries 512
list-max-ziplist-value 64

set-max-intset-entries 512

zset-max-ziplist-entries 128
zset-max-ziplist-value 64

activerehashing yes

client-output-buffer-limit normal 0 0 0
client-output-buffer-limit slave 256mb 64mb 60
client-output-buffer-limit pubsub 32mb 8mb 60

Linux安装

download：http://download.redis.io/releases/
wget http://download.redis.io/releases/redis-6.2.6.tar.gz
注意事项和常见问题：
1.如果执行wget提示“wget 未找到该命令”，则先执行yum -y install wget后再执行上面的命令。
2.如果执行make提示“cc 未找到该命令”则先运行yum -y install gcc automake autoconf libtool make命令,
再删除redis-6.2.6文件夹，重新执行解压缩及以后的步骤即可。
3./home/admin/myapps/redis 可以更换为任意你想安装redis的路径。

1.将下载好的压缩包放到/usr/local目录下
tar xzf redis-3.0.2.tar.gz
cd redis-3.0.2 
make
(执行安装，安装到/home/admin/myapps/redis目录
make PREFIX=/home/admin/myapps/redis install)
提示错误 make: cc: Command not found make: *** [adlist.o] Error 127
没有安装gcc环境，需要安装gcc
yum install gcc-c++
yum install gcc
安装后检查是否安装成功
rpm -qa |grep gcc
之后重新make
make  # 耗时大约3分钟
make install

2.编译完成后，在src目录下，有四个可执行文件redis-server、redis-benchmark、
redis-cli和redis.conf将其拷贝到一个目录下。
# 建议复制config文件到指定目录，后期使用这个新的config文件
# mkdir /usr/redis
# cp redis-server  /usr/redis
# cp redis-benchmark /usr/redis
# cp redis-cli  /usr/redis
# cd ..
# cp redis.conf  /usr/redis
# cd /usr/redis

vim redis.conf 
daemonize yes # 改为后台启动方式
redis-server redis.conf  # 启动服务，不指定配置文件默认使用redis目录下的redis.conf
redis-cli  -p 6379 # 客户端连接，默认 -h localhost -p 6379

ps -ef|grep redis   # 查看redis进程是否开启，一个服务端，一个客户端
root     21736  2965  0 02:47 ?        00:01:16 redis-server 127.0.0.1:6379
root     28975 26023  0 19:30 pts/3    00:00:00 redis-cli -p 6379
root     29140 24347  0 19:30 pts/1    00:00:00 grep --color=auto redis

kill -9 21736

压力测试

src/redis-benchmark  -h localhost -p 6379 -c 100 -n 100000
====== SET ======  # 这里只是截取了set，还有get 等很多命令
  100000 requests completed in 2.01 seconds  # 对100000个请求进行写入测试
  100 parallel clients # 100个并发客户端
  3 bytes payload # 每次写入3字节
  keep alive: 1 # 只有一台服务器来处理这些请求，单击性能
  host configuration "save": 900 1 300 10 60 10000
  host configuration "appendonly": no
  multi-thread: no
23.81% <= 1 milliseconds
91.56% <= 2 milliseconds
95.31% <= 3 milliseconds
97.77% <= 4 milliseconds
98.51% <= 5 milliseconds  # 5秒后处理了98.51%
100.00% <= 20 milliseconds # 20秒后处理了100%
49751.24 requests per second # 每秒处理49751.24个请求

基本命令

select index # 切换数据库 0-15
dbsize # 查看当前数据库大小
keys * # 查看所有key
keys s* # 查看所有以s开头的key
del k
flushdb 
flushall
exists k # 是否存在
move k index # 移动到 指定数据库 中
rename oldk newk 
renamenx oldk newk # 当newk不存在时才会执行
ttl key  # 查看过期时间 单位秒 -1永远不过期 -2已过期
persist k # 设置k永不过期
expire key 10 # 设置过期时间，单位s
expireat k # 设置过期时间，单位时间戳
type key # 查看类型
randomkey # 随机返回一个k

Redis 为什么单线程还那么快？
单线程，基于内存操作，性能瓶颈依据机器内存和网络宽带
误区1：高性能不一定是多线程
误区2：多线程（CPU上下文切换）一定比单线程快
核心：redis是将所有的数据全部放在内存中，所以说使用单线程去操作效率已经是最高，
多线程会多次上下文切换，很耗时；对于内存系统来说，没有上下文切换就是理论最快，内存情况下这样就是最佳方案！

数据类型

http://doc.redisfans.com/index.html【在线手册】

String 字符串

set k v   
set k v ex 20 # set同时过期时间20s，默认s
get k
setex k 30 v # 新建key，同时设置过期时间，单位s
setnx k v #  当k不存在时set
getset k v # 没有k，新建且赋值v 返回错误信息：(nil)，有k，获取v

ttl k # 查看剩余过期时间 -2表示已过期
mset k1 v1 k2 v2 # 同时设置多个key
msetnx k1 v1 k2 v2 # k不存在时候执行；原子操作，所有k必须都不存在，才能顺利执行
mget k1 k2
append k # 给key的value追加str，key不存在，会创建key
strlen k # 查看length

incr k # 自增1，只针对数字
incrby k 10 # 增加10， key不存在时先被初始化为0，然后再执行
decr key # 月incr相反
decrby key 7

getrange k start end #  index从0开始 (包括 start 和 end 在内)
getrange k 1 -1 # -1表示截取到尾部
setrange k index  xxxxx # index从0开始，从index开始替换五个字符

应用场景：

mset user:1 {name:zhangan,age:24}
上面的name age都在一个key里面，下面的写法优势是可以设置/更新部分字段
mset user:1:name zhangsan user:1:age 24
mget user:1:name user:1:age

setnx分布式锁
优势：因为是setnx所以第一个才能设置成功，达到锁的目的

incr article:readcount:{} 文章阅读数+1、点赞统计，类似计数器

List 列表

• 实际上是一个链表，有节点概念，left right before after 都可以操作
• key不存在，会创建新的链表
• 移除了所有值，空链表，便会消失
• 两侧操作效率高，中间操作相对慢一点
• 变相可实现消息 （消息队列 lpush rpop）（栈 lpush lpop）
• 命令全部L开头 先进先出 value可重复，有序

lpush k v1 v2 v3 v4 v5
lrange k start end # 截取一部分，index从0开始，-1表示尾部，-2表示倒数第二个
lpushx k v1 v2 # key存在时插入
rpush k v
rpushx k v # k不存在，push无效

lpop k # 得到一个元素且移除该元素
rpop k
blpop [k1, k2, k3...]  timeout  # k1 k2 k3 按顺序查找，返回匹配的第一个k和v，等待时间timeout毫秒 ？？？

lindex k index # 根据index获取指定元素，也可以-1 -2倒叙获取
llen k # 获取个数

lrem k count v # 从左边开始移除count个匹配的value
lrem k -2 morning    # 移除从表尾到表头，共计-2的绝对值个 morning
lrem k 0 hello      # 移除表中所有 hello，0表示所有

ltrim key start end # 裁掉范围之外元素，含头不含尾
rpoplpush source destination # 剪切一个元素到目标，右弹左入
lset key index value # key不存在会报错，更新指定index的value
linsert k before/after targetValue myValue #在指定value前/后插入 value

Set 无序集合

不重复。随机获取

sadd k v1 v2 v3 v4 v5
smembers k # 查看时顺序并不是按照插入顺序，但每次查出后的顺序是固定的
sismember k v # 查看v在不在k中
srem k v # 移除指定的value
scard k # 统计元素个数
spop k # 弹出一个 且 移除
smove source destination v # 从源key剪切一个value存入目标key
srangemember k # 随机获取一个元素，并不移除该元素
srangemember k count # 随机获取count个元素，并不移除该元素

sdiff k1 k2 k3 k4 # 以k1为基准，向右逐个比较，取出差集
sdiffstore destination k1 k2 k3 # 取出差集，存入新的key中
sinter k1 k2 k3 # 交集
sinterstore destination k1 k2 k3 # 取出交集，存入新的key中
sunion k1 k2 k3 # 并集
sunionstore destination k1 k2 k3 # 取出并集，存入新的key中

抽奖
sadd k zs lisi wangwu # 将用户全部sadd进去，
smembers k # smembers查看所有用户
spop key # spop抽出一个幸运用户，该用户会移除该队伍，避免重复中奖
srangemember k 5 # 随机抽5名用户

点赞
sadd like:消息ID zhangsan # 张三点赞了，加入该消息的like列表
srem like:消息ID zhangsan # 张三取消点赞
sismember like:消息ID zhangsan # 检查张三是否给该消息点过赞
smenbers like:消息ID # 获取全部点赞的人
scard like:消息ID # 获取点赞总数

关注模型
sinter Mata Faker  # Mata和Faker的共同关注
sismember jay G.E.M # 我关注的G.E.M是否也关注jay （我关注的人也关注他）
sdiff k1 k2 # 差集 （你认识的，我不认识的 = 可能认识的人）

Zset 有序集合

比set多了一个属性score，根据score排序，默认从小到大
场景：给每一个进入排队的人给一个顺序号，来的早的号码最小，小号先开始结账
就这样保证了顺序

redis 127.0.0.1:6379> ZADD runoobkey 1 redis
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 2 mongodb
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 3 mysql
(integer) 1
redis 127.0.0.1:6379> ZADD runoobkey 3 mysql
(integer) 0
redis 127.0.0.1:6379> ZADD runoobkey 4 mysql
(integer) 0
redis 127.0.0.1:6379> ZRANGE runoobkey 0 10 WITHSCORES
1) "redis"
2) "1"
3) "mongodb"
4) "2"
5) "mysql"
6) "4"


zadd k score1 v1 # key没有则创建
zadd k score1 v1 score2 v2

zrange k 0 -1 [withscores]
zrevrange k 0 -1 # 反排序
zrevrangebyscore k max min # 按分数从大到小获取

zrank k v # 获取v在k中index
zrevrank k v # 获取v的index

zscore k v  # 获取分数
zincrby k step v # 给v增加step，只针对数字

zadd sal 5000 zhangsan
zadd sal 7800 xiaohong
zadd sal 9000 bob

zrangebyscore k min max
zrangebyscore k -inf +inf
zrangebyscore k -inf +inf withscores
zrangebyscore k -inf 2500 withscores # 包括2500

zrem k v
zremrangebyscore k min max 
zremrangebyrank k start end 

zcard k
zcount k min max


zunionstore newKey count k1 k2 k3 
# 将三个k组合得到交集放入newkey，count是期望组合的key的数量
# 返回的newkey为空时，会删除newkey

积分排行榜
zadd hotnews:202010001 点击数 新闻标题 # 新建一则新闻   key=hotnews:202010001   score=点击数    v=新闻标题
zincrby hotnews:202010001 1 新闻标题  # 点击次数+1
zrevrange hotnews:202010001 0 9 withscores # 获取热点新闻前十名 做排行
zunionstore newKey count k1 k2 k3 # 三天三个knewkey，选出交集，newkey就是三天全部新闻 再使用zrevrange newkey 0 4 withscore 得到前5

Hash 哈希

hset k filed1 value1 # 一个key，可以存多个字段，都有对应的value
hsetnx f k v # 不存在就执行
hmset f k1 v1 k2 v2
hget f k1
hmget f k1 k2
hgetall k # 取出key的全部filed value
hkeys f # 取出key的全部filed
hvals f# 取出key的全部value

hdel f k1
hlen f # key中filed的总数
hexits f k1

hincrby f k num # 增加num
hincrby f k -1 # 减少1

存储user对象
hset user:1 name zhangsan
hmset user {userid}:name zs {userid}:age 24
hmget user 9527:name 9527:age
hmset userid name zs age 24
hmget userid name age

购物车
用户id：9527      商品id：10087    商品总数：value
hset 9527 10087 1 # 给用户 9527 添加 1 件商品 10087
hset 9529 10088 1
hset 9628 10089 1
hincrby 9527 10087 10 # 给10087商品数量增加10
hlen 9527 # 查看商品总数
hdel 9527 10087 #删除用户的10087商品
hgetall 9527 # 获取全部商品

同类数据整合归类，方便管理
相比string消耗内存与cpu更小，更节省空间

过期时间只能用在key上，不能用再filed
Redis集群架构下不适合大规模使用（大量用户的数据不能全放在一个user下）

geospatial 地理位置

有效经度 -180 180

有效纬度 -85.05112878 85.05112878

geoadd china:city 116.40 39.90 beijing
geoadd china:city 121.47 31.23 shanghai
geoadd china:city 106.50 29.53 chongqing 114.05 22.52 shenzhen
geoadd china:city 120.16 30.24 hangzhou 108.96 34.26 xian

geopos china:city beijing
geopos china:city beijing chongqin

geodist china:city beijing shanghai km  # m km mi ft

georadius china:city 110 30 1000 km # 以110 30未中心寻找1000km的城市
georadius china:city 110 30 1000 km withdist # 以110 30为中心寻找1000km的城市，且显示距离
georadius china:city 110 30 1000 km withcoord # 以110 30为中心寻找1000km的城市，显示经纬信息
georadius china:city 110 30 1000 km withdist withcoord count 1 # 只显示1条

georadiusbymember china:city beijing 1000 km # 与georadius类似，不过以现有值作为基准，此处为“beijing”

geohash china:city beijing chongqing # 将多个元素转为字符串，如“wx4fbxxfke0”

geo底层实现就是用的zset，所以可以使用zset命令操作
zrange china:city 0 -1 # 查看全部元素
zrem china:city beijing # 删除指定元素

Hyperloglog 基数

统计计数而已，可以接受误差

pfadd k a b c d e f g
pfcount k
pfmerge targetKey k1 k2

Bitmap

位存储，只关注状态，比如登陆和未登录，关注和未关注，打卡和未打卡, 只能存0或者1

setbit kaoqin 0 1
setbit kaoqin 1 0
setbit kaoqin 2 1
setbit kaoqin 3 0
setbit kaoqin 4 1

getbit kaoqin 2
bitcount kaoqin # 统计key=kaoqin中 value=1 的总数

事务

multi
sql1
sql2
...
sql3
exec
discard

redis事务没有原子性，单条命令才有
编译时错误，exec时全部sql不执行，直接报错
运行时错误，会跳过该sql，其他sql照样执行，换句话说redis没有事务原子性

watch
unwatch
exec时会检查监视对象有无变化，有则执行失败

持久化

rdb

指定的时间间隔内将内存中数据集快照写入磁盘
dump.rdb
file dump.db # data数据类型文件  二进制

触发机制：
	1，正常关闭数据库时
	2，flushall
	3，save 900 1 等等（900s内有一个变化）
优点：
	1，效率高
	2，适合大规模数据恢复
	3，对数据的完整性要求不高
缺点：
	1，处于时间间隔中的数据在意外宕机后会丢失
	2，fork进程时会占用一定的内存空间

aof

记录每一句命令，然后执行命令达到恢复数据的目的

appendonly no # 默认不开启aof模式，大部分情况下rdb已经够用
appendfilename "appendonly.aof"

appendfsync always # 每次修改都会sync，消耗性能
appendfsync everysec # 每一秒sync一次，最多丢失1s的数据
appendfsync no # 不执行sync，系统会自动同步，这种速度最快 ？？？

优点：
	数据文件相对比较大，aof远大于rdb，恢复数据速度也较慢，因为记录的是每一条命令
	运行效率也比rdb慢，所以默认时rdb持久化方式

发布/订阅

客户端a订阅频道，客户端b向频道发送消息，a即可自动收到消息，频道名称可自定义

客户端1：
SUBSCRIBE channel1 # 表示订阅自定义频道channel1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "channel1"
3) (integer) 1

客户端2：
PUBLISH channel1 "hello world" # 往channel1频道发送消息
(integer) 1
127.0.0.1:6379> PUBLISH channel1 "22222"
(integer) 1
发送之后，客户端1自动接收：
1) "message"
2) "kuang"
3) "hello world"
1) "message"
2) "kuang"
3) "22222"
使用场景：
	1，实时消息系统，比如订阅某up主后会立刻受到up主设置的问候消息
	2，实时聊天系统，订阅，关注场景
稍微复杂的场景要使用消息队列MQ

主从复制

主从复制，是指将一台Redis服务器的数据，复制到其他的Redis服务器。前者称为主节点
(master)，后者称为从节点(slave)；数据的复制是单向的，只能由主节点到从节点。
Master以写为主，Slave 以读为主。
默认情况下，每台Redis服务器都是主节点；且一个主节点可以有多个从节点(或没有从节点)，但一个从节点只能有一个主节点。

主从复制作用：

• 数据的热备份：主从复制实现了数据的热备份，是持久化之外的一种数据冗余方式。
• 故障转移：当主节点出现问题时，可以由从节点提供服务，实现快速的故障恢复
• 负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，由从节点提供读服务（即写Redis数据时应用连接主节点，读Redis数据时应用连接从节点），分担服务器负载；尤其是在写少读多的场景下，通过多个从节点分担读负载，可以大大提高Redis服务器的并发量。
• 高可用：除了上述作用以外，主从复制还是哨兵和集群能够实施的基础，因此说主从复制是
  Redis高可用的基础。

Redis数量推荐
1、从结构上，单个Redis服务器会发生单点故障，并且一台服务器需要处理所有的请求负载，压力较大；
2、从容量上，单个Redis服务器内存容量有限，大型项目不满足。

对于这种场景，我们可以使如下这种架构：
（一般至少3个，一主二从）

• 多配置几个服务器分散压力；服务器有好几个，那么客户端就得好几个；
• 主服务器负责写入数据，其他服务器作为从服务器，同步master的数据，负责读取数据；这就是：读写分离；
• 80%都是读操作；
• 主服务器写入数据后，从服务器salveof master后，会自动同步master的数据，包括master的历史数据

环境搭建

复制三个conf文件
cp redis.conf redis6379.conf
cp redis.conf redis6380.conf
cp redis.conf redis6381.conf
然后逐个修改3个文件的内容：
	port 改为6379
	daemonize yes （是否可后台运行）
	pidfile redis_6379.pid
	logfile "6379.log"
	dbfilename dump6379.rdb
ps -ef | grep redis # 3个服务都启动后，要能看到3个server
 
配置从机（关联主机）（不配置主从关系前，默认都是主节点）
slaveof 主库ip 主库端口 # 配置主从
info replication # 查看信息，包括角色是不是master，连接的从机有几个
 
主机可以写入，从机只能读（读写分离）
slaveof  ip 6379
命令来配置所属主服务器时手动操作，一次性的，需要永久生效，就配置conf文件，但后期如果修改的话就得改文件


层层链路（自己起的名字）
上一个Slave可以是下一个Slave的Master，Slave同样可以接收其他slaves的连接和同步请求，
那么该slave作为了链条中下一个slave的Master，如此可以有效减轻Master的写压力。
如果slave中途变更转向，会清除之前的数据，重新建立最新的。


测试：主机挂了，查看从机信息无影响，主机恢复，再次查看信息又恢复正常

全量复制：从机salveof master时，会连同历史数据全部发送至从机
增量复制：只同步新增的数据

b机器既是主机，也是从机，info查询显示实际上是slave，也不能写入数据

当Master挂掉后，Slave可键入命令 slaveof no one使当前从机独立，自然就成了Master；
但如果不修改从机的master，主机回来后依然可以恢复master的身份进行主从复制。

sync原理：
1、Slave启动成功连接到master后会发送一个sync命令；
2、Master接到命令启动后的存盘进程，同时收集所有接收到的用于修改数据集命令，
在后台进程执行完毕之后，master将传送整个数据文件到slave，以完成一次完全同步；
3、全量复制：而slave服务在数据库文件数据后，将其存盘并加载到内存中；
4、增量复制：Master继续将新的所有收集到的修改命令依次传给slave，完成同步；
5、但是只要是重新连接master，一次完全同步（全量复制）将被自动执行。

数据同步延时
由于所有的写操作都是在Master上操作，然后同步更新到Slave上，所以从Master同步到Slave机器有一定的延迟，
当系统很繁忙的时候，延迟问题会更加严重，Slave机器数量的增加也会使得这个问题更加严重。

哨兵模式

手动版的升级版，即自动版解决主从关系

就是将主从复制功能配置为自动化，哨兵机制是一个独立进程，独立运行，所以要新开启一个服务，专门启动哨兵，通过发送命令等待Redis服务器的响应，从而达到监控的目的。

哨兵模式是一种特殊模式，首先Redis提供了哨兵命令，哨兵是一个独立的进程，它会独立运行。其原理是哨兵通过发送命令等待Redis服务器响应，从而监控运行的多个Redis实例。

哨兵的作用：
1.通过发送命令，让Redis服务器返回监控其运行状态，包括主服务器和从服务器。
2.当哨兵检测到master宕机，会自动从slave中选举一个为master，然后通过发送订阅模式通知其他的从服务器，修改配置文件，让他们切换主机。

一个哨兵进程对Redis服务器进行监控，可能会出现问题，所以我们使用多个哨兵进行监控。各个哨兵之间还会进行监控，这样就形成了多个哨兵模式。

故障切换的过程：
假设主服务器宕机，哨兵1先检测到这个结果，系统不会马上进行failover过程，仅仅时哨兵1直观认为主服务器不可用，这个现象叫主观下线。当后面的哨兵也检测到主服务器不可用，并且数量达到一定值时，那么哨兵会进行一次投票，投票的结果由一个哨兵发起，进行failover操作。切换成功后，就会通过发布订阅模式，让各个哨兵把自己监控的从服务器实现切换主机，这个过程成为客观下线。（主观下线的数量到达一定后，才会客观下线）

反客为主的自动版，能够后台监控Master库是否故障，如果故障了根据投票数自动将slave库转换为主库。
一组sentinel能同时监控多个Master。
使用步骤：
1、在Master对应redis.conf同目录下新建sentinel.conf文件，名字绝对不能错；
2、配置哨兵，在sentinel.conf文件中填入内容：
sentinel monitor 被监控数据库名字（自己起名字） ip port 1
说明：上面最后一个数字1，表示主机挂掉后slave投票看让谁接替成为主机，得票数多少后成为主机。
3、启动哨兵模式：
命令键入：redis-sentinel /myredis/sentinel.conf
注：上述sentinel.conf路径按各自实际情况配置

哨兵配置说明

# Example sentinel.conf
# 哨兵sentinel实例运行的端口 默认26379
port 26379
# 哨兵sentinel的工作目录
dir /tmp
# 哨兵sentinel监控的redis主节点的 ip port
# master-name 可以自己命名的主节点名字 只能由字母A-z、数字0-9 、这三个字符".-_"组成。
# quorum 配置多少个sentinel哨兵统一认为master主节点失联 那么这时客观上认为主节点失联了
# sentinel monitor <master-name> <ip> <redis-port> <quorum>
sentinel monitor mymaster 127.0.0.1 6379 2

# 当在Redis实例中开启了requirepass foobared 授权密码 这样所有连接Redis实例的客户端都
要提供密码
# 设置哨兵sentinel 连接主从的密码 注意必须为主从设置一样的验证密码
# sentinel auth-pass <master-name> <password>
sentinel auth-pass mymaster MySUPER--secret-0123passw0rd
# 指定多少毫秒之后 主节点没有应答哨兵sentinel 此时 哨兵主观上认为主节点下线 默认30秒
# sentinel down-after-milliseconds <master-name> <milliseconds>
sentinel down-after-milliseconds mymaster 30000
# 这个配置项指定了在发生failover主备切换时最多可以有多少个slave同时对新的master进行 同
步，
这个数字越小，完成failover所需的时间就越长，
但是如果这个数字越大，就意味着越 多的slave因为replication而不可用。
可以通过将这个值设为 1 来保证每次只有一个slave 处于不能处理命令请求的状态。
# sentinel parallel-syncs <master-name> <numslaves>
sentinel parallel-syncs mymaster 1
# 故障转移的超时时间 failover-timeout 可以用在以下这些方面：
#1. 同一个sentinel对同一个master两次failover之间的间隔时间。
#2. 当一个slave从一个错误的master那里同步数据开始计算时间。直到slave被纠正为向正确的
master那里同步数据时。
#3.当想要取消一个正在进行的failover所需要的时间。 
#4.当进行failover时，配置所有slaves指向新的master所需的最大时间。不过，即使过了这个超
时，slaves依然会被正确配置为指向master，但是就不按parallel-syncs所配置的规则来了
# 默认三分钟
# sentinel failover-timeout <master-name> <milliseconds>
sentinel failover-timeout mymaster 180000
# SCRIPTS EXECUTION
#配置当某一事件发生时所需要执行的脚本，可以通过脚本来通知管理员，例如当系统运行不正常时发邮
件通知相关人员。
#对于脚本的运行结果有以下规则：
#若脚本执行后返回1，那么该脚本稍后将会被再次执行，重复次数目前默认为10
#若脚本执行后返回2，或者比2更高的一个返回值，脚本将不会重复执行。
#如果脚本在执行过程中由于收到系统中断信号被终止了，则同返回值为1时的行为相同。
#一个脚本的最大执行时间为60s，如果超过这个时间，脚本将会被一个SIGKILL信号终止，之后重新执
行。
#通知型脚本:当sentinel有任何警告级别的事件发生时（比如说redis实例的主观失效和客观失效等
等），将会去调用这个脚本，这时这个脚本应该通过邮件，SMS等方式去通知系统管理员关于系统不正常
运行的信息。调用该脚本时，将传给脚本两个参数，一个是事件的类型，一个是事件的描述。如果
sentinel.conf配置文件中配置了这个脚本路径，那么必须保证这个脚本存在于这个路径，并且是可执
行的，否则sentinel无法正常启动成功。
#通知脚本
# sentinel notification-script <master-name> <script-path>
sentinel notification-script mymaster /var/redis/notify.sh
# 客户端重新配置主节点参数脚本
# 当一个master由于failover而发生改变时，这个脚本将会被调用，通知相关的客户端关于master
地址已经发生改变的信息。
# 以下参数将会在调用脚本时传给脚本:
# <master-name> <role> <state> <from-ip> <from-port> <to-ip> <to-port>
# 目前<state>总是“failover”,
# <role>是“leader”或者“observer”中的一个。
# 参数 from-ip, from-port, to-ip, to-port是用来和旧的master和新的master(即旧的
slave)通信的

# 这个脚本应该是通用的，能被多次调用，不是针对性的。
# sentinel client-reconfig-script <master-name> <script-path>
sentinel client-reconfig-script mymaster /var/redis/reconfig.sh

缓存穿透、击穿、雪崩

Redis缓存的使用，极大的提升了应用程序的性能和效率，特别是数据查询方面。但同时，它也带来了一
些问题。其中，最要害的问题，就是数据的一致性问题，从严格意义上讲，这个问题无解。如果对数据
的一致性要求很高，那么就不能使用缓存。
另外的一些典型问题就是，缓存穿透、缓存雪崩和缓存击穿。目前，业界也都有比较流行的解决方案。

穿透：(查不到数据)

程序在处理缓存时，一般是先从缓存查询，如果缓存没有这个key获取为null，则会从DB中查询，并设置到缓存中去。
按这种做法，那查询一个一定不存在的数据值，由于缓存是不命中时需要从数据库查询，查不到数据则不写入缓存，这将导致这个不存在的数据每次请求都要到数据库去查询，造成缓存穿透。


但是这种方法会存在两个问题：
1、如果空值能够被缓存起来，这就意味着缓存需要更多的空间存储更多的键，因为这当中可能会有很多
的空值的键；
2、即使对空值设置了过期时间，还是会存在缓存层和存储层的数据会有一段时间窗口的不一致，这对于
需要保持一致性的业务会有影响。

击穿(查询量太大)

一个key非常热点，在不停的扛着大并发，大并发集中访问，当这个key在失效的瞬间，持续的大并发就穿破缓存，直接请求数据库，并且回写缓存，会导使数据库瞬间压力过大。

解决方案:

1. 加互斥锁：通过synchronized+双重检查机制：某个key只让一个线程查询，阻塞其它线程(缺点: 会阻塞其它线程)
2. 设置value永不过期(是最可靠的，最安全的但是占空间，内存消耗大，并且不能保持数据最新 这个需要根据具体的业务逻辑来做 )

雪崩

缓存雪崩，是指在某一个时间段，缓存集中过期失效

雪崩就是指缓存中大批量热点数据过期后系统涌入大量查询请求，因为大部分数据在Redis层已经失效，请求渗透到数据库层，大批量请求引起数据库宕机。

解决办法：
1，将缓存失效时间分散开，比如每个key的过期时间是随机，防止同一时间大量数据过期现象发生，这样不会出现同一时间全部请求都落在数据库层，如果缓存数据库是分布式部署，将热点数据均匀分布在不同Redis和数据库中，有效分担压力，别一个人扛。
2，简单粗暴，让Redis数据永不过期（如果业务准许，比如不用更新的名单类）。当然，如果业务数据准许的情况下可以，比如中奖名单用户，每期用户开奖后，名单不可能会变了，无需更新。

产生雪崩的原因之一，比如在写本文的时候，马上就要到双十二零点，很快就会迎来一波抢购，这波商
品时间比较集中的放入了缓存，假设缓存一个小时。那么到了凌晨一点钟的时候，这批商品的缓存就都
过期了。而对这批商品的访问查询，都落到了数据库上，对于数据库而言，就会产生周期性的压力波
峰。于是所有的请求都会达到存储层，存储层的调用量会暴增，造成存储层也会挂掉的情况。

其实集中过期，倒不是非常致命，比较致命的缓存雪崩，是缓存服务器某个节点宕机或断网。因为自然形成的缓存雪崩，一定是在某个时间段集中创建缓存，这个时候，数据库也是可以顶住压力的。无非就是对数据库产生周期性的压力而已。而缓存服务节点的宕机，对数据库服务器造成的压力是不可预知的，很有可能瞬间就把数据库压垮。

3种解决方案
redis高可用
这个思想的含义是，既然redis有可能挂掉，那我多增设几台redis，这样一台挂掉之后其他的还可以继续
工作，其实就是搭建的集群。
限流降级
这个解决方案的思想是，在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对
某个key只允许一个线程查询数据和写缓存，其他线程等待。
数据预热
数据加热的含义就是在正式部署之前，我先把可能的数据先预先访问一遍，这样部分可能大量访问的数
据就会加载到缓存中。在即将发生大并发访问前手动触发加载缓存不同的key，设置不同的过期时间，让
缓存失效的时间点尽量均匀。

区别

穿透是单个key在缓存中不存在，从而进入数据库中查询，此时DB产生压力；
击穿是指热点key的大量查询，此时DB压力巨大，面临随时宕机可能；
雪崩是大量的热点key全部涌入服务器，此时DB已经扛不住而宕机，即使重启也会再次宕机。

层级关系、目录形式存储数据

        jedis.set("user:01", "user_zhangsan");
        jedis.set("user:02", "lisi");
        System.out.println(jedis.get("user:01"));

Jedis & redisTemplate

Jedis

    <dependency>
      <groupId>redis.clients</groupId>
      <artifactId>jedis</artifactId>
      <version>3.2.0</version>
    </dependency>

Jedis jedis = new Jedis("39.99.164.237", 6379);
System.out.println(jedis.ping()); // pong
//jedis.auth("root");// 设置认证密码
jedis.select(0);// 指定数据库 默认是 0

//批量插入
JSONObject jsonObject = new JSONObject();
jsonObject.put("age",24);
jsonObject.put("sex","female");
jsonObject.put("name","Faker");
Transaction multi = jedis.multi();
String str = jsonObject.toJSONString();
try {
    multi.set("user1", str);
    multi.set("user2", str);
    multi.exec();
} catch (Exception e){
    multi.discard();
    e.printStackTrace();
} finally {
    System.out.println(jedis.get("user1"));
    System.out.println(jedis.get("user2"));
    jedis.close();
}

Set<String> keys = jedis.keys("*");
System.out.println(keys.toString());

System.out.println(jedis.dbSize());

jedis.flushAll();//清空整个 Redis 服务器的数据
jedis.flush();//清空当前数据库中的所有 key

Transaction tx = jedis.multi();//multi标记事务开始
tx.set("tel", "10010");
// tx.exec();// 提交事务
tx.discard();// 回滚事务

//JedisUtil工具类
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class JedisUtil {

    private static String ADDR = "192.168.75.163";
    private static int PORT = 6379;
    //private static String AUTH = "root";
    //可用连接实例的最大数目，默认值为 8；
    //如果赋值为-1，则表示不限制；如果 pool 已经分配了 maxActive 个 jedis 实例，则此时 pool 的状态为 exhausted(耗尽)。
    private static int MAX_ACTIVE = 1024;
    //控制一个 pool 最多有多少个状态为 idle(空闲的)的 jedis 实例，默认值也是 8。
    private static int MAX_IDLE = 200;
    //等待可用连接的最大时间，单位毫秒，默认值为-1，表示永不超时。如果超过等待时间，则直接抛出 JedisConnectionException；
    private static int MAX_WAIT = 10000;
    private static int TIMEOUT = 10000;
    //在 borrow 一个 jedis 实例时，是否提前进行 validate 操作；如果为 true，则得到的 jedis 实例均是可用的；
    private static boolean TEST_ON_BORROW = true;
    private static JedisPool jedisPool = null;

    /**
     * 初始化 Redis 连接池
     */
    static {
        try {
            JedisPoolConfig config = new JedisPoolConfig();
            config.setMaxTotal(MAX_ACTIVE);
            config.setMaxIdle(MAX_IDLE);
            config.setMaxWaitMillis(MAX_WAIT);
            config.setTestOnBorrow(TEST_ON_BORROW);
            //jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT, AUTH);
            jedisPool = new JedisPool(config, ADDR, PORT, TIMEOUT);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public synchronized static Jedis getJedis() {
        try {
            if (jedisPool != null)
                return jedisPool.getResource();
            return null;
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    // 释放 jedis 资源
    public static void returnResource(final Jedis jedis) {
        if (jedis != null) {
            jedisPool.close();
        }
    }
}

SpringBoot整合Redis

springboot 2.x 之后spring-boot-starter-data-redis底层的 jedis 被替代为 letture

jedis
Spring-data-redis
Spring-boot-starter-data-Redis
三者关系是后者封装前者，jedis最原生类似jdbc

jedis:采用直连，多线程时不安全，使用jedis pool
lettuce：采用netty，实例在多线程中共享，不存在不安全情况

pom

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-redis</artifactId>
		</dependency>
    	<dependency>
      		<groupId>com.alibaba</groupId>
      		<artifactId>fastjson</artifactId>
      		<version>1.2.62</version>
    	</dependency>

yml

spring:
 redis:
  host: 39.99.164.237

SpringBootTest

@SpringBootTest
class DemoApplicationTests {
	@Autowired
	private RedisTemplate redisTemplate;
	@Test
	void contextLoads() {
		redisTemplate.opsForValue().set("age","24");
		// 如果 set 对象，该 entity 必须实现序列化
		System.out.println(redisTemplate.opsForValue().get("age"));
	}
}

如果有中文，会出现乱码，解决方法是增加自定义序列化配置文件

@Configuration
public class Config {
    @Autowired
    private RedisTemplate redisTemplate;
    @Bean
    public RedisTemplate redisTemplateInit() {
        StringRedisSerializer stringSerializer = new StringRedisSerializer();
        redisTemplate.setKeySerializer(stringSerializer);
        redisTemplate.setValueSerializer(stringSerializer);
        redisTemplate.setHashKeySerializer(stringSerializer);
        redisTemplate.setHashValueSerializer(stringSerializer);
        return redisTemplate;
    }
}

RedisTemplate

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-redis</artifactId>
		</dependency>

boolean redisTemplate.hasKey(key) //判断是否有key所对应的值
boolean redisTemplate.expire(key,time,TimeUnit.MINUTES); // 指定key的失效时间
Long expire = redisTemplate.getExpire(key); //    根据key获取过期时间
boolean redisTemplate.delete(key); //
// 删除多个key
public void deleteKey (String ...keys){
    redisTemplate.delete(keys);
}

redisTemplate.opsForValue().set("k","v");
redisTemplate.opsForValue().get("k")'

redisTemplate.opsForHash().put("pig","name","peiqi");

redisTemplate.opsForList().rightPush("k","v");

redisTemplate.opsForSet().add("hobby","giter","pingpang");
System.out.println(redisTemplate.opsForSet().members("hobby")); // [giter, pingpang]

redisTemplate.opsForZSet().add("k","v1",12);
redisTemplate.opsForZSet().add("k","v2",5);
redisTemplate.opsForZSet().add("k","v3",152);
System.out.println(redisTemplate.opsForZSet().range("k",0,-1)); // [v2, v1, v3]

API列表

//String类型
redisTemplate.dump(key)  将当前传入的key值序列化为byte[]类型

设置过期时间
public Boolean expire(String key, long timeout, TimeUnit unit) {
    return redisTemplate.expire(key, timeout, unit);
}   
public Boolean expireAt(String key, Date date) {
    return redisTemplate.expireAt(key, date);
}

查找匹配的key值，返回一个Set集合类型
public Set<String> getPatternKey(String pattern) {
    return redisTemplate.keys(pattern);
}

修改redis中key的名称
 public void renameKey(String oldKey, String newKey) {
    redisTemplate.rename(oldKey, newKey);
}

返回传入key所存储的值的类型
public DataType getKeyType(String key) {
    return redisTemplate.type(key);
}

如果旧值存在时，将旧值改为新值
public Boolean renameOldKeyIfAbsent(String oldKey, String newKey) {
    return redisTemplate.renameIfAbsent(oldKey, newKey);
}

从redis中随机取出一个key
redisTemplate.randomKey()

返回当前key所对应的剩余过期时间
public Long getExpire(String key) {
    return redisTemplate.getExpire(key);
}

返回剩余过期时间并且指定时间单位
public Long getExpire(String key, TimeUnit unit) {
    return redisTemplate.getExpire(key, unit);
}

将key持久化保存
public Boolean persistKey(String key) {
    return redisTemplate.persist(key);
}

将当前数据库的key移动到指定redis中数据库当中
public Boolean moveToDbIndex(String key, int dbIndex) {
    return redisTemplate.move(key, dbIndex);
}

设置当前的key以及value值
redisTemplate.opsForValue().set(key, value)

设置当前的key以及value值并且设置过期时间
redisTemplate.opsForValue().set(key, value, timeout, unit)

返回key中字符串的子字符
public String getCharacterRange(String key, long start, long end) {
    return redisTemplate.opsForValue().get(key, start, end);
}

将旧的key设置为value，并且返回旧的key
public String setKeyAsValue(String key, String value) {
    return redisTemplate.opsForValue().getAndSet(key, value);
}

批量获取值
public List<String> multiGet(Collection<String> keys) {
    return redisTemplate.opsForValue().multiGet(keys);
}

在原有的值基础上新增字符串到末尾
redisTemplate.opsForValue().append(key, value)

以增量的方式将double值存储在变量中
public Double incrByDouble(String key, double increment) {
    return redisTemplate.opsForValue().increment(key, increment);
}

通过increment(K key, long delta)方法以增量方式存储long值（正值则自增，负值则自减）
public Long incrBy(String key, long increment) {
    return redisTemplate.opsForValue().increment(key, increment);
}

如果对应的map集合名称不存在，则添加否则不做修改
Map valueMap = new HashMap();  
valueMap.put("valueMap","map1");  
valueMap.put("valueMap2","map2");  
valueMap.put("valueMap3","map3");  
redisTemplate.opsForValue().multiSetIfAbsent(valueMap); 

设置map集合到redis
Map valueMap = new HashMap();  
valueMap.put("valueMap1","map1");  
valueMap.put("valueMap2","map2");  
valueMap.put("valueMap3","map3");  
redisTemplate.opsForValue().multiSet(valueMap);  

获取字符串的长度
redisTemplate.opsForValue().size(key)

用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始
redisTemplate.opsForValue().set(key, value, offset)

重新设置key对应的值，如果存在返回false，否则返回true
redisTemplate.opsForValue().setIfAbsent(key, value)

将值 value 关联到 key,并将 key 的过期时间设为 timeout
redisTemplate.opsForValue().set(key, value, timeout, unit)

将二进制第offset位值变为value
redisTemplate.opsForValue().setBit(key, offset, value)

对key所储存的字符串值，获取指定偏移量上的位(bit)
redisTemplate.opsForValue().getBit(key, offset)

//Hash类型
Redis hash 是一个string类型的field和value的映射表，hash特别适合用于存储对象。
Redis 中每个 hash 可以存储 2^32 - 1 键值对（40多亿）。

获取变量中的指定map键是否有值,如果存在该map键则获取值，没有则返回null。
redisTemplate.opsForHash().get(key, field)

获取变量中的键值对
public Map<Object, Object> hGetAll(String key) {
    return redisTemplate.opsForHash().entries(key);
}

新增hashMap值
redisTemplate.opsForHash().put(key, hashKey, value)

以map集合的形式添加键值对
public void hPutAll(String key, Map<String, String> maps) {
    redisTemplate.opsForHash().putAll(key, maps);
}

仅当hashKey不存在时才设置
public Boolean hashPutIfAbsent(String key, String hashKey, String value) {
    return redisTemplate.opsForHash().putIfAbsent(key, hashKey, value);
}

删除一个或者多个hash表字段
public Long hashDelete(String key, Object... fields) {
    return redisTemplate.opsForHash().delete(key, fields);
}

查看hash表中指定字段是否存在
public boolean hashExists(String key, String field) {
    return redisTemplate.opsForHash().hasKey(key, field);
}

给哈希表key中的指定字段的整数值加上增量increment

public Long hashIncrBy(String key, Object field, long increment) {
    return redisTemplate.opsForHash().increment(key, field, increment);
}

public Double hIncrByDouble(String key, Object field, double delta) {
    return redisTemplate.opsForHash().increment(key, field, delta);
}

获取所有hash表中字段
redisTemplate.opsForHash().keys(key)

获取hash表中字段的数量
redisTemplate.opsForHash().size(key)

获取hash表中存在的所有的值
public List<Object> hValues(String key) {
    return redisTemplate.opsForHash().values(key);
}

匹配获取键值对，ScanOptions.NONE为获取全部键对
public Cursor<Entry<Object, Object>> hashScan(String key, ScanOptions options) {
    return redisTemplate.opsForHash().scan(key, options);
}

//List类型
通过索引获取列表中的元素
redisTemplate.opsForList().index(key, index)

获取列表指定范围内的元素(start开始位置, 0是开始位置，end 结束位置, -1返回所有)
redisTemplate.opsForList().range(key, start, end)

存储在list的头部，即添加一个就把它放在最前面的索引处
redisTemplate.opsForList().leftPush(key, value)

把多个值存入List中(value可以是多个值，也可以是一个Collection value)
redisTemplate.opsForList().leftPushAll(key, value)

List存在的时候再加入
redisTemplate.opsForList().leftPushIfPresent(key, value)

如果pivot处值存在则在pivot前面添加
redisTemplate.opsForList().leftPush(key, pivot, value)

按照先进先出的顺序来添加(value可以是多个值，或者是Collection var2)
redisTemplate.opsForList().rightPush(key, value)
redisTemplate.opsForList().rightPushAll(key, value)

在pivot元素的右边添加值
redisTemplate.opsForList().rightPush(key, pivot, value)

设置指定索引处元素的值
redisTemplate.opsForList().set(key, index, value)

移除并获取列表中第一个元素(如果列表没有元素会阻塞列表直到等待超时或发现可弹出元素为止)
redisTemplate.opsForList().leftPop(key)
redisTemplate.opsForList().leftPop(key, timeout, unit)

移除并获取列表最后一个元素
redisTemplate.opsForList().rightPop(key)
redisTemplate.opsForList().rightPop(key, timeout, unit)

从一个队列的右边弹出一个元素并将这个元素放入另一个指定队列的最左边
redisTemplate.opsForList().rightPopAndLeftPush(sourceKey, destinationKey)
redisTemplate.opsForList().rightPopAndLeftPush(sourceKey, destinationKey, timeout, unit)

删除集合中值等于value的元素(index=0, 删除所有值等于value的元素; index>0, 
从头部开始删除第一个值等于value的元素; index<0, 从尾部开始删除第一个值等于value的元素)
redisTemplate.opsForList().remove(key, index, value)

将List列表进行剪裁
redisTemplate.opsForList().trim(key, start, end)

获取当前key的List列表长度
redisTemplate.opsForList().size(key)

//Set类型
添加元素
redisTemplate.opsForSet().add(key, values)

移除元素(单个值、多个值)
redisTemplate.opsForSet().remove(key, values)

删除并且返回一个随机的元素
redisTemplate.opsForSet().pop(key)

获取集合的大小
redisTemplate.opsForSet().size(key)

判断集合是否包含value
redisTemplate.opsForSet().isMember(key, value)

获取两个集合的交集(key对应的无序集合与otherKey对应的无序集合求交集)
redisTemplate.opsForSet().intersect(key, otherKey)

获取多个集合的交集(Collection var2)
redisTemplate.opsForSet().intersect(key, otherKeys)

key集合与otherKey集合的交集存储到destKey集合中(其中otherKey可以为单个值或者集合)
redisTemplate.opsForSet().intersectAndStore(key, otherKey, destKey)

key集合与多个集合的交集存储到destKey无序集合中
redisTemplate.opsForSet().intersectAndStore(key, otherKeys, destKey)

获取两个或者多个集合的并集(otherKeys可以为单个值或者是集合)
redisTemplate.opsForSet().union(key, otherKeys)

key集合与otherKey集合的并集存储到destKey中(otherKeys可以为单个值或者是集合)
redisTemplate.opsForSet().unionAndStore(key, otherKey, destKey)

获取两个或者多个集合的差集(otherKeys可以为单个值或者是集合)
redisTemplate.opsForSet().difference(key, otherKeys)

差集存储到destKey中(otherKeys可以为单个值或者集合)
redisTemplate.opsForSet().differenceAndStore(key, otherKey, destKey)

随机获取集合中的一个元素
redisTemplate.opsForSet().randomMember(key)

获取集合中的所有元素
redisTemplate.opsForSet().members(key)

随机获取集合中count个元素
redisTemplate.opsForSet().randomMembers(key, count)

获取多个key无序集合中的元素（去重），count表示个数
redisTemplate.opsForSet().distinctRandomMembers(key, count)

遍历set类似于Interator(ScanOptions.NONE为显示所有的)
redisTemplate.opsForSet().scan(key, options)

// zSet类型
ZSetOperations提供了一系列方法对有序集合进行操作
添加元素(有序集合是按照元素的score值由小到大进行排列)
redisTemplate.opsForZSet().add(key, value, score)

删除对应的value,value可以为多个值
redisTemplate.opsForZSet().remove(key, values)

增加元素的score值，并返回增加后的值
redisTemplate.opsForZSet().incrementScore(key, value, delta)

返回元素在集合的排名,有序集合是按照元素的score值由小到大排列
redisTemplate.opsForZSet().rank(key, value)

返回元素在集合的排名,按元素的score值由大到小排列
redisTemplate.opsForZSet().reverseRank(key, value)

获取集合中给定区间的元素(start 开始位置，end 结束位置, -1查询所有)
redisTemplate.opsForZSet().reverseRangeWithScores(key, start,end)

按照Score值查询集合中的元素，结果从小到大排序
redisTemplate.opsForZSet().reverseRangeByScore(key, min, max)
redisTemplate.opsForZSet().reverseRangeByScoreWithScores(key, min, max)

从高到低的排序集中获取分数在最小和最大值之间的元素
redisTemplate.opsForZSet().reverseRangeByScore(key, min, max, start, end)

根据score值获取集合元素数量
redisTemplate.opsForZSet().count(key, min, max)

获取集合的大小
redisTemplate.opsForZSet().size(key)
redisTemplate.opsForZSet().zCard(key)

获取集合中key、value元素对应的score值
redisTemplate.opsForZSet().score(key, value)

移除指定索引位置处的成员
redisTemplate.opsForZSet().removeRange(key, start, end)

移除指定score范围的集合成员
redisTemplate.opsForZSet().removeRangeByScore(key, min, max)

获取key和otherKey的并集并存储在destKey中（其中otherKeys可以为单个字符串或者字符串集合）
redisTemplate.opsForZSet().unionAndStore(key, otherKey, destKey)

获取key和otherKey的交集并存储在destKey中（其中otherKeys可以为单个字符串或者字符串集合）
redisTemplate.opsForZSet().intersectAndStore(key, otherKey, destKey)

遍历集合（和iterator一模一样）
Cursor<TypedTuple<Object>> scan = opsForZSet.scan("test3", ScanOptions.NONE);
while (scan.hasNext()){
    ZSetOperations.TypedTuple<Object> item = scan.next();
    System.out.println(item.getValue() + ":" + item.getScore());
}