Redis 全笔记
Table of Contents generated with DocToc (opens new window)
- Redis记录
# Redis记录
# 概述
- Redis是noSQL(not only SQL)数据库 即非关系型数据库
- 以key-value形式缓存数据在内存中，读取速度极快支持多种数据结构如string,list,set,zset,hash等
- 这些数据都支持push/pop,add/remove以及取交集并集差集等操作，且这些操作都是原子性的也可进行数据持久化，
- 将数据存储到硬盘：RDB和AOF方式
- Redis与memcached区别：Redis会周期性的把更新的数据写入磁盘或者把修改操作写入追加的记录文件,在此基础上，Redis实现了主从同步
# 安装Redis
先给Linux安装gcc环境 yum -y install gcc automake autoconf libtool make
将gcc升级到9
# gcc -v # 查看gcc版本 # yum -y install centos-release-scl # 升级到9.1版本 # yum -y install devtoolset-9-gcc devtoolset-9-gcc-c++ devtoolset-9-binutils # scl enable devtoolset-9 bash1
2
3
4在/目录下新建soft文件夹，将redis的tar.gz包放入，使用tar -zxvf redis-xx解压(如果没安装tar，可以使用 yum install tar -y安装)
在redis的解压目录下输入make指令
在redis的解压目录下输入 make install PREFIX=/usr/local/redis(prefix表示安装的目录，不指定的话默认安装到/usr/local/bin里面)
观察/usr/local/redis中的文件结构，需要安装tree命令软件(yum install tree -y)
然后启动redis服务：/usr/local/redis/bin/redis-server
再打开一个终端窗口使用redis客户端操作：/usr/local/redis/bin/redis-cli
在redis服务启动的窗口使用ctrl+c或shutdown(推荐)或使用kill -9 进程id杀死redis进程。shutdown命令后的参数可以是save/nosave，一般用save进行本地保存以便下次启动时恢复数据
通过ps -aux|grep redis可查看redis服务端的进程
firewall-cmd --zone=public --add-port=6379/tcp --permanent：开启6379防火墙
firewall-cmd --reload：重新加载防火墙
# Redis配置允许远程访问
配置redis访问密码
port 6500 daemonize yes pidfile /var/run/redis_6500.pid logfile "/usr/local/redis/bin/redis-6500.log" dbfilename dump-6500.rdb requirepass 1234561
2
3
4
5
6指定监听的ip地址
port 6500 daemonize yes pidfile /var/run/redis_6500.pid logfile "/usr/local/redis/bin/redis-6500.log" dbfilename dump-6500.rdb bind 192.168.3.421
2
3
4
5
6直接禁用保护模式(不推荐、危险)
port 6500 daemonize yes pidfile /var/run/redis_6500.pid logfile "/usr/local/redis/bin/redis-6500.log" dbfilename dump-6500.rdb protected-mode no1
2
3
4
5
6推荐使用设置密码+设置监听IP的方式
port 6500 daemonize yes pidfile /var/run/redis_6500.pid logfile "/usr/local/redis/bin/redis-6500.log" dbfilename dump-6500.rdb bind 192.168.3.42 requirepass 1234561
2
3
4
5
6
7
# 其他的Redis配置
笔记
loglevel：redis的日志记录等级，默认为notice，共有四级(debug<verbose<notice<warning)
databases：redis服务中数据库的数量，通过select+数字切换，默认为0
timeout：超时关闭连接的时间，当连接redis服务的客户端timeout秒没有活动的时候就关闭这个连接。若设置为0则表示永远不关闭
save：指出在多长时间内，有多少次更新操作，就将数据同步到数据文件
rdbcompression：指定存储至本地数据库时是否压缩数据，默认为 yes
Redis 采用 LZF 压缩，如果为了节省 CPU 时间，可以关闭该选项，但会导致数据库文件变的巨大
dir： 本地数据库存放路径，默认值为 ./
slaveof： 当本机为从服务时，设置主服务的IP及端口，如：192.168.0.100:6500
slave-read-only： 当本机为从服务时，是否设置为只读状态
masterauth： 当本机为从服务时，设置主服务的连接密码，如：123456
maxclients： 最大客户端连接数，默认10000，参照
/soft/redis-6.0.6/redis.conf中的说明。可在客户端中使用config get maxclients查看这个配置
appendonly：指定是否在每次更新操作后进行日志记录（注意是进行日志记录，而不是存储数据），默认为no。Redis 在默认情况下是异步的把数据写入磁盘，如果不开启，可能会在断电时导致一段时间内的数据丢失。因为 redis 本身同步数据文件是按上面 save 条件来同步的，所以有的数据会在一段时间内只存在于内存中
appendfilename：指定更新日志文件名，默认为 appendonly.aof，默认的就是appendonly.aof文件
###启动redis
```bash
/usr/local/redis/bin/redis.conf或/usr/local/redis/bin/redis-server
2
3
4
5
# 连接客户端
#-h：指定ip -p：指定端口 -a：若开启了验证，需指定验证密码登录
/usr/local/redis/bin/redis-cli -h 127.0.0.1 -p 6379 -a 123
2
# 设置redis开机启动
1：编辑 /etc下的 rc.local 文件,将redis的启动命令写入文件
vim /etc/rc.local
2：编辑脚本内容
/usr/local/redis/redis-server /usr/local/redis/redis.conf
tip：若开机脚本不起作用，可能因为 rc.local 的执行权限问题。
chmod 755 /etc/rc.local
2
3
4
5
6
7
8
9
# 用途举例
- 获取最新的N个数据：通过list实现按自然时间排序的数据
- 排行榜topN：利用zset(有序集合)实现
- 时效性数据，如手机、邮箱验证码：利用Expire过期
- 计数器，秒杀：原子性，利用自增方法INCR、DECR
- 去除大量数据中的重复数据：利用Set集合
- 构建队列：利用list
- 发布订阅消息系统：pub/sub模式
# Redis键(key)
- **keys ：查看当前库中所有key (匹配：keys 1)
- exists key：判断某个key是否存在
- type key：查看你的key是什么类型
- del key：删除指定的key的数据
- unlink key：非阻塞删除数据(异步删除,不是立刻删除)，仅将key从keyspace元数据中删除，真正的删除会在后续异步执行
- expire key 10(单位是seconds)：为给定的key设置过期时间
- ttl key：查看还有多少秒过期，-1表示永不过期，-2表示已经过期
- select (数字)：切换数据库(0-15)一共16个数据库
- dbsize：查看当前数据库的key的数量
- flushdb：清空当前库
- flushall：清空所有库
# 原子性
所谓原子操作是指不会被线程调度机制打断的操作;
这种操作一旦开始，就会一直运行到结束，中间不会有任何context switch(切换到另一线程)
(1)在单线程中，能够在单条指令中完成的操作都可以认为是"原子操作"，因为中断只能发生在指令之间
(2)在多线程中，不能被其他进程(线程)打断的操作就叫原子操作，redis单命令的原子性主要得益于redis是单线程
2
3
4
# String
String是Redis最基本的类型，你可以理解成与Memcached一模一样的类型，一个key对应一个value。v String类型是二进制安全的。意味着Redis的string 可以包含任何数据。比如jpg图片或者序列化的对象。· String类型是Redis最基本的数据类型，一个Redis 中字符串value最多可以是512M
# 基本命令
- set [key] [value] 存一个，重复set同一个key ，原来的值会被覆盖
- get [key] 获取对应key 的值
- append [key] [value] 在指定key 的值后面追加value字符串
- strlen [key] 获取对应key值的字符串长度
- setnx [key] [value]：只有在key不存在时设置key的值
- incr [key] 将key中储存的数字值加1，只能对数字值操作，如果为空，新增这个key，赋值为1
- decr [key] 将key中储存的数字值减1，只能对数字值操作，如果为空，新增这个key，赋值为-1
- incrby/decrby [key] [步长]：将key中储存的数字进行增减自定义步长
- mset [key] [value] [key] [value] [key] [value]....同时设置一个或多个key value键值对
- mget [key] [value] [key] [value] [key] [value]....同时获取一个或多个key的值
- msetnx [key] [value] [key] [value]：当其中有一个key是已存在的，整个msetnx操作会失败(原子性的)
- getrange [key] [index] [end] 获取值得范围，类似Java substring，左右均为闭区间
- setrange [key] [index] [value] 用value覆盖从index(包含index)开始后的字符串value长度个字符串(index从0开始)
- setex [key] [过期时间] [value]：在设置值的时候即可设置过期时间
- getset [key] [value]：以新换旧，设置新值得到旧值(取到的是旧值，然后存储的旧值被新值替换)
# String的数据结构
String 的string的底层数据结构为简单动态字符串(Simple Dynamic String,缩写SDS)。是可以修改的字符串，内部结构实现上类似于Java的ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配。内部为当前字符串实际分配的空间capacity一般要高于实际字符串长度len。当字符串长度小于1M时，扩容都是加倍现有的空间，如果超过1M，扩容时一次只会多扩1M的空间。需要注意的是字符串最大长度为512M。
SDS定义
struct sdshdr{
//记录buf数组中已使用字节的数量
//等于 SDS 保存字符串的长度
int len;
//记录 buf 数组中未使用字节的数量
int free;
//字节数组，用于保存字符串
char buf[];
}
2
3
4
5
6
7
8
9
# 相对于使用C语言原本的char[]的字符串有什么好处
常数复杂度获取字符串长度
由于 len 属性的存在，我们获取 SDS 字符串的长度只需要读取 len 属性，时间复杂度为 O(1)。而对于 C 语言，获取字符串的长度通常是经过遍历计数来实现的，时间复杂度为 O(n)。通过 strlen key 命令可以获取 key 的字符串长度
杜绝缓冲区溢出
我们知道在 C 语言中使用 strcat 函数来进行两个字符串的拼接，一旦没有分配足够长度的内存空间，就会造成缓冲区溢出。而对于 SDS 数据类型，在进行字符修改的时候，会首先根据记录的 len 属性检查内存空间是否满足需求，如果不满足，会进行相应的空间扩展，然后在进行修改操作，所以不会出现缓冲区溢出
减少修改字符串的内存重新分配次数
C语言由于不记录字符串的长度，所以如果要修改字符串，必须要重新分配内存（先释放再申请），因为如果没有重新分配，字符串长度增大时会造成内存缓冲区溢出，字符串长度减小时会造成内存泄露。
而对于SDS，由于len属性和free属性的存在，对于修改字符串SDS实现了空间预分配和惰性空间释放两种策略：
- 空间预分配：对字符串进行空间扩展的时候，扩展的内存比实际需要的多，这样可以减少连续执行字符串增长操作所需的内存重分配次数
- 惰性空间释放：对字符串进行缩短操作时，程序不立即使用内存重新分配来回收缩短后多余的字节，而是使用 free 属性将这些字节的数量记录下来，等待后续使用。（当然SDS也提供了相应的API，当我们有需要时，也可以手动释放这些未使用的空间。）
二进制安全
因为C字符串以空字符作为字符串结束的标识，而对于一些二进制文件（如图片等），内容可能包括空字符串，因此C字符串无法正确存取；而所有 SDS 的API 都是以处理二进制的方式来处理 buf 里面的元素，并且 SDS 不是以空字符串来判断是否结束，而是以 len 属性表示的长度来判断字符串是否结束
兼容部分 C 字符串函数
虽然 SDS 是二进制安全的，但是一样遵从每个字符串都是以空字符串结尾的惯例，这样可以重用 C 语言库<string.h> 中的一部分函数
总结
# List
# 概述
单键多值
Redis列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素到列表的头部(左边）或者尾部(右边)。 它的底层实际是个双向链表，对两端的操作性能很高，通过索引下标的操作中间的节点性能会较差。
# 常用命令
lrange/rrange [key] [start] [stop] 按照索引下标获取元素(从左到右或从右到左)
如果start是0，stop是-1，即可取到所有值，因为0表示左边第一个，-1表示右边第一个
lpush/rpush [key] [value] [key] [value] [key] [value]..... ：从左或右插入一个或多个值，插入方式为左插法或右插法，即每次插入到当前方向的最后一个的后面。
例如lpush k1 v1 k2 v2 k3 v3，然后使用lrange 0 2取到的值得顺序是v3 v2 v1，因为插入的方式是先v1
，然后v2 v1，然后v3 v2 v1。
lpop/rpop [key] 从左边/右边吐出一个值
rpoplpush [key1] [key2] 从[key1]列表右边吐出一个值，插到[key2]列表的左边
lindex[key] [index] 根据索引获取元素(从左到右)
llen[key] 获得链表长度
linsert[key] before/after [value] [newvalue] 在value的之前或之后插入newvalue
lset[key] [index] [value] 将列表key下标为index的值替换为value
lrem[key] [n] [value] 从左边开始删除n个value
# List的数据结构
List的数据结构为快速链表quickList。
首先在列表元素较少的情况下会使用一块连续的内存存储，这个结构是ziplist，也即是压缩列表。 它将所有的元素紧挨着一起存储，分配的是一块连续的内存。当数据量比较多的时候才会改成quicklist。
因为普通的链表需要的附加指针空间太大，会比较浪费空间。比如列表里存的只是int类型的数据，结构上还需要两个额外的指针prev和next。
Redis将链表和ziplist结合起来组成了quicklist。也就是将多个ziplist使用双向指针串起来使用。这样既满足了快速的插入删除性能，又不会出现太大的空间冗余。
链表结点定义：
typedef struct listNode{
//前置节点
struct listNode *prev;
//后置节点
struct listNode *next;
//节点的值
void *value;
}listNode
2
3
4
5
6
7
8
双向链表定义：
typedef struct list{
//表头节点
listNode *head;
//表尾节点
listNode *tail;
//链表所包含的节点数量
unsigned long len;
//节点值复制函数
void (*free) (void *ptr);
//节点值释放函数
void (*free) (void *ptr);
//节点值对比函数
int (*match) (void *ptr,void *key);
}list;
2
3
4
5
6
7
8
9
10
11
12
13
14
Redis链表特性：
- 双端：链表具有前置节点和后置节点的引用，获取这两个节点时间复杂度都为O(1)
- 无环：表头节点的 prev 指针和表尾节点的 next 指针都指向 NULL,对链表的访问都是以 NULL 结束
- 带链表长度计数器：通过 len 属性获取链表长度的时间复杂度为 O(1)
- 多态：链表节点使用 void* 指针来保存节点值，可以保存各种不同类型的值
# 作为List底层实现之一的压缩列表
压缩列表（ziplist）是Redis为了节省内存而开发的，是由一系列特殊编码的连续内存块组成的顺序型数据结构，一个压缩列表可以包含任意多个节点（entry），每个节点可以保存一个字节数组或者一个整数值
压缩列表的原理：压缩列表并不是对数据利用某种算法进行压缩，而是将数据按照一定规则编码在一块连续的内存区域，目的是节省内存
压缩列表的每个节点构成如下：
- previous_entry_ength：记录压缩列表前一个字节的长度。previous_entry_ength的长度可能是1个字节或者是5个字节，如果上一个节点的长度小于254，则该节点只需要一个字节就可以表示前一个节点的长度了，如果前一个节点的长度大于等于254，则previous length的第一个字节为254，后面用四个字节表示当前节点前一个节点的长度。利用此原理即当前节点位置减去上一个节点的长度即得到上一个节点的起始位置，压缩列表可以从尾部向头部遍历。这么做很有效地减少了内存的浪费
- encoding：节点的encoding保存的是节点的content的内容类型以及长度，encoding类型一共有两种，一种字节数组一种是整数，encoding区域长度为1字节、2字节或者5字节长
- content：content区域用于保存节点的内容，节点内容类型和长度由encoding决定
# Set
# 概述
Redis set对外提供的功能与list类似是一个列表的功能，特殊之处在于set是可以自动排重的，当你需要存储一个列表数据，又不希望出现重复数据时，set是一个很好的选择。
并且set提供了判断某个成员是否在一个set集合内的重要接口，这个也是list所不能提供的。
Redis的set是string类型的无序集合。它底层其实是一个value为null的hash表，所以添加，删除，查找的复杂度都是O(1)。
一个算法，随着数据的增加，执行时间的长短，如果是O(1)，数据增加，查找数据的时间不变
# 常用命令
sadd[key] [value1] [value2] 将一个或多个元素加入到集合key中
smembers[key] 取出该集合的所有值
sismember[key] [value] 判断集合key是含有该value，有为1，没有为0
scard[key] 返回该集合的元素个数
srem[key] [value1] [value2]..删除集合中的某(几)个元素
spop[key] [n]随机从该集合中吐出n个值，不加n默认吐出一个，即拿出，会从集合中删除
srandmember[key] [n]随机从该集合中取出n个值，不会从集合中删除
smove[source] [destination] value 把集合source中的值为value的元素移动到集合destination中
sinter[key1] [key2] 返回两个集合的交集元素
sunion[key1] [key2] 返回两个集合的并集元素
sdiff[key1] [key2] 返回两个集合的差集元素(key1中有的，key2中没有的)
# set的数据结构
set的数据结构是dict(dictionary)字典，字典是用哈希表实现的
Java中 HashSet的内部实现使用的是 HashMap，只不过所有的value都指向同一个对象。
Redis的set结构也是一样，它的内部也使用hash结构，所有的value都指向同一个内部值。
# 有序集合zset
# 概述
Redis有序集合zset与普通集合set非常相似，是一个没有重复元素的字符串集合。· 不同之处是有序集合的每个成员都关联了一个评分(score),这个评分(score)被用来按照从最低分到最高分的方式排序集合中的成员。集合的成员是唯一的，但是评分可以是重复了。 因为元素是有序的，所以你也可以很快的根据评分(score)或者次序(position)来获取一个范围的元素。 访问有序集合的中间元素也是非常快的,因此你能够使用有序集合作为一个没有重复成员的智能列表。
# 常用命令
zadd[key] [score1] [value1] [score2] [value2]...：将一个或多个member元素及其score值加入到有序集合key中
zrange[key] [start] [stop] [WAHTSCORES]：返回有序结合中，下标在start和top之间的元素,带WAHTSCORES可以让score和值一起返回。stop为-1时返回所有数据
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]统计该集合的元素score在区间[min,max]之间的元素个数
zrank[key] [value] 返回该值在集合中的排名，从0开始
案例：如何利用zset实现一个文章访问量的排行榜
# 数据结构
SortedSet(zset)是Redis提供的一个非常特别的数据结构，一方面它等价于Java的数据结构Map<String，Double>，可以给每一个元素value赋予一个权重score，另一方面它又类似于TreeSet，内部的元素会按照权重score进行排序，可以得到每个元素的名次，还可以通过score的范围来获取元素的列表。 zset底层使用了两个数据结构 (1) hash ,hash的作用就是关联元素value和权重score，保障元素value的唯一性，可以通过元素value找到相应的score值。 (2）跳跃表，跳跃表的目的在于给元素value排序，根据score的范围获取元素列表。
# 有序集合zset底层实现之跳跃表
概念
跳跃表（skiplist）是一种有序数据结构，它通过在每个节点中维持多个指向其它节点的指针，从而达到快速访问节点的目的。具有如下性质：
- 由很多层结构组成；
- 每一层都是一个有序的链表，排列顺序为由高层到底层，都至少包含两个链表节点，分别是前面的head节点和后面的nil节点；
- 最底层的链表包含了所有的元素；
- 如果一个元素出现在某一层的链表中，那么在该层之下的链表也全都会出现（上一层的元素是当前层的元素的子集）；
- 链表中的每个节点都包含两个指针，一个指向同一层的下一个链表节点，另一个指向下一层的同一个链表节点；
Redis中跳跃表结点的定义如下：
typedef struct zskiplistNode {
//层
struct zskiplistLevel{
//前进指针
struct zskiplistNode *forward;
//跨度
unsigned int span;
}level[];
//后退指针
struct zskiplistNode *backward;
//分值
double score;
//成员对象
robj *obj;
} zskiplistNode
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
多个跳跃表结点构成一个跳跃表：
typedef struct zskiplist{
//表头节点和表尾节点
structz skiplistNode *header, *tail;
//表中节点的数量
unsigned long length;
//表中层数最大的节点的层数
int level;
}zskiplist;
2
3
4
5
6
7
8
操作
- 搜索：从最高层的链表节点开始，如果比当前节点要大和比当前层的下一个节点要小，那么则往下找，也就是和当前层的下一层的节点的下一个节点进行比较，以此类推，一直找到最底层的最后一个节点，如果找到则返回，反之则返回空。
- 插入：首先确定插入的层数，有一种方法是假设抛一枚硬币，如果是正面就累加，直到遇见反面为止，最后记录正面的次数作为插入的层数。当确定插入的层数k后，则需要将新元素插入到从底层到k层
- 删除：在各个层中找到包含指定值的节点，然后将节点从链表中删除即可，如果删除以后只剩下头尾两个节点，则删除这一层
# hash
key item value
命令：
- hdel key field1 [field2] 删除一个或多个哈希表字段
- hexists key field 查看哈希表 key 中，指定的字段是否存在。
- hget key field 获取存储在哈希表中指定字段的值。
- hgetall key 获取在哈希表中指定 key 的所有字段和值
- hincrby key field increment 为哈希表key中的指定字段的整数值加上常量increment
- hincrbyfloat key field increment 为哈希表 key 中的指定字段的浮点数值加上增量 increment 。
- hkeys key 获取哈希表中所有字段
- hlen key 获取哈希表中字段的数量
- hmget key field1 [field2] 获取所有给定字段的值
- hmset key field1 value [field2 value2] 同时将多个 field-value (域-值)对设置到哈希表 key 中。
- hset key field value 将哈希表 key 中的字段 field 的值设为 value
- hsetnx key field value 只有在字段 field 不存在时，设置哈希表字段的值。
- hvals key 获取哈希表中所有值
- hscan key cursor [MATCH pattern] [COUNT count] 迭代哈希表中的键值对
# 底层结构
Redis 的字典使用哈希表作为底层实现哈希表结构定义：
typedef struct dictht{
//哈希表数组
dictEntry **table;
//哈希表大小
unsigned long size;
//哈希表大小掩码，用于计算索引值
//总是等于 size-1
unsigned long sizemask;
//该哈希表已有节点的数量
unsigned long used;
}dictht
2
3
4
5
6
7
8
9
10
11
哈希表是由数组 table 组成，table 中每个元素都是指向 dictEntry 结构，dictEntry 结构定义如下：
typedef struct dictEntry{
//键
void *key;
//值
union{
void *val;
uint64_tu64;
int64_ts64;
}v;
//指向下一个哈希表节点，形成链表
struct dictEntry *next;
}dictEntry
2
3
4
5
6
7
8
9
10
11
12
key 用来保存键，val 属性用来保存值，值可以是一个指针，也可以是uint64_t整数，也可以是int64_t整数
注意这里还有一个指向下一个哈希表节点的指针，我们知道哈希表最大的问题是存在哈希冲突，如何解决哈希冲突，有开放地址法和链地址法。这里采用的便是链地址法，通过next这个指针可以将多个哈希值相同的键值对连接在一起，用来解决哈希冲突
连在一起的键K1和键K0
**哈希算法：**Redis计算哈希值和索引值方法如下：
#1、使用字典设置的哈希函数，计算键 key 的哈希值 hash = dict->type->hashFunction(key); #2、使用哈希表的sizemask属性和第一步得到的哈希值，计算索引值 index = hash & dict->ht[x].sizemask;1
2
3
4**解决哈希冲突：**方法是链地址法。通过字典里面的 *next 指针指向下一个具有相同索引值的哈希表节点
**扩容和收缩：**当哈希表保存的键值对太多或者太少时，就要通过 rerehash(重新散列）来对哈希表进行相应的扩展或者收缩。具体步骤：
- 如果执行扩展操作，会基于原哈希表创建一个大小等于 ht[0].used*2n 的哈希表（也就是每次扩展都是根据原哈希表已使用的空间扩大一倍创建另一个哈希表）。相反如果执行的是收缩操作，每次收缩是根据已使用空间缩小一倍创建一个新的哈希表
- 重新利用上面的哈希算法，计算索引值，然后将键值对放到新的哈希表位置上
- 所有键值对都迁徙完毕后，释放原哈希表的内存空间
触发扩容的条件：
服务器目前没有执行 BGSAVE 命令或者 BGREWRITEAOF 命令，并且负载因子大于等于1
服务器目前正在执行 BGSAVE 命令或者 BGREWRITEAOF 命令，并且负载因子大于等于5
ps：负载因子 = 哈希表已保存节点数量 / 哈希表大小
渐近式 rehash
渐近式 rehash，扩容和收缩操作不是一次性、集中式完成的，而是分多次、渐进式完成的。如果保存在Redis中的键值对只有几个几十个，那么 rehash 操作可以瞬间完成，但是如果键值对有几百万，几千万甚至几亿，那么要一次性的进行 rehash，势必会造成Redis一段时间内不能进行别的操作。所以Redis采用渐进式 rehash,这样在进行渐进式rehash期间，字典的删除查找更新等操作可能会在两个哈希表上进行，第一个哈希表没有找到，就会去第二个哈希表上进行查找。但是进行 增加操作，一定是在新的哈希表上进行的
# Redis配置文件详解
port：端口号，默认6379
tcp-backlog
设置tcp的backlog,backlog其实是一个连接队列，backlog队列总和=未完成三次握手队列＋已经完成三次握手队列。 在高并发环境下你需要一个高backlog值来避免慢客户端连接问题。u 注意Linux内核会将这个值减小到/proc/sys/net/core/somaxconn的值(128），所以需要确认增大/proc/sys/net/core/somaxconn和/proc/sys/net/ipv4/tcp_max_syn_backlog(128)两个值来达到想要的效果
# Redis新数据类型
# Bitmaps
简介
常用命令
setbit[key] [offset] [value] 设bitmaps中某个偏移量的值为0或1.offset：偏移量从0开始
注意：很多应用的用户id以一个指定数字（例如10000 ）开头，直接将用户id和Bitmaps的偏移量对应势必会造成一定的浪费，通常的做法是每次做setbit 操作时将用户id减去这个指定数字。 在第一次初始化 Bitmaps时，假如偏移量非常大，那么整个初始化过程执行会比较慢，可能会造成 Redis的阻塞。1
2getbit[key] [offset] 获取bitmaps中第偏移量为offset的值(offset从0开始)
bitcount[key] [start end]：统计字符串被设置为1的bit数，一般情况下给定的整个字符串都会被进行计数，通过指定额外的start和end参数可以让计数只在特定的位之间进行。start和end参数的设置都可以使用负数值，比如-1表示最后一个位，-2表示倒数第二个位。start、end是指bit组的字节的下标数，[start,end]左右皆为闭区间。[start,end]表示有(end-start+1)8个字节空间。例如[0,2]，即下标在0~(2-0+1)*8-1=23之间的下标中，值为1的个数。
bitop and(or/not/xor) [destkey] [key]：bitop是一个复合操作，它可以做多个Bitmaps的and(交集）、or(并集） 、not(非) 、 xor（异或）操作并将结果保存在destkey中。
set与bitmap对比
# HyperLogLog
简介
常用命令
- pfadd[key] [element]...：添加一个或多个元素到HyperLogLog中。执行命令后如果HLL估计的近似基数发生变化，则返回1，否则返回0
- pfcount[key] [key]....：计算HLL的近似基数，可以计算多个
- pfmerge[destkey] [sourcekey] [sourcekey]....：将一个或多个HLL合并后的结果存在另一个HLL destkey中。比如每月活跃用户可以使用每天的活跃用户来合并计算出来
- pfadd[key] [element]...：添加一个或多个元素到HyperLogLog中。执行命令后如果HLL估计的近似基数发生变化，则返回1，否则返回0
# Geospatial
简介
基本命令
Geospatial底层是zset，所以可用zset的命令实现Geospatial自身命令不能实现的操作(如删除操作)
geoadd[key] [longitude] [latitude] [member] [longitude latitude member...]：添加地理位置(经度、纬度、名称)
两极无法直接添加，一般会下载城市数据，直接通过Java程序一次性导入 有效的经度从-180度到180度。有效的纬度从-85.05112878度到85.05112878度 当坐标位置超出指定范围时，该命令将会返回一个错误 已经添加的数据，是无法再次往里面添加的
geopos[key] [member...]：获取一个或多个指定地区的坐标值
geodis[key] [member1] [member2] [m|km|ft|mi]：获取两个位置之间的直线距离
m表示单位为米(默认值) km表示单位为千米 mi表示单位为英里 ft表示单位为英尺
georadius[key] [longitude] [latitude] radius [m|km|ft|mi]：找出某一半径内的元素(经度 纬度 距离 单位)
# jedis访问redis步骤
关闭Linux防火墙或者开放redis的端口
firewall-cmd --zone=public --add-port=6379/tcp --permanent开启6379防火墙 firewall-cmd --reload：重新加载防火墙1
2记得注释掉配置文件中的bind 127.0.0.1
redis.conf设置密码后，服务端启动时需带上redis.conf即指定配置文件启动
./redis-server redis.conf1客户端启动时需要使用-a带上requirpass的密码
./redis-cli -a 密码1使用jedis连接时，也需要加上密码
# redis事务
笔记
redis事务的本质就是一组命令的集合。一个事务中所有的命令都会被序列化(顺序)，按顺序执行 具有：一次性、顺序性、排他性 redis单条命令保证原子性(因为redis是单线程)，但事务不保证原子性
过程：
开启事务：multi
命令入队：输入一系列命令
中途也可取消事务：使用discard命令
执行事务：exec
# 事务不执行的情况
出现编译型异常(代码有问题即命令本身有错误)，那么事务中的所有命令都不会执行
因为队列中有一次命令出现了语法错误，所以所有命令均未执行
出现运行时异常(如1/0这种)，出现此种错误，其他命令仍可正常执行
这种错误其他命令仍可正常执行
# 乐观锁实现秒杀系统
注意
我们知道大多数是基于数据版本（ version )的记录机制实现的。即为数据增加一个版本标识，在基于数据库表的版本解决方案中，一般是通过为数据库表增加一个"version"字段来实现读取出数据时，将此版本号一同读出，之后更新时，对此版本号加1。此时，将提交数据的版本号与数据库表对应记录的当前版本号进行比对，如果提交的数据版本号大于数据库当前版本号，则予以更新，否则认为是过期数据。redis中可以使用watch命令会监视给定的key，当exec时候如果监视的key从调用watch后发生过变化，则整个事务会失败。也可以调用watch多次监视多个key。这样就可以对指定的key加乐观锁了。注意watch的key是对整个连接有效的，事务也一样。如果连接断开，监视和事务都会被自动清除。当然了exec , discard , unwatch命令都会清除连接中的所有监视
# redis实现乐观锁
当一个线程在要对数据进行操作，但它的操作还未完成时，数据便被另一线程改变，这一线程操作的数据就不是原来的数据了。为了防止这种情况，可以使用乐观锁。
使用watch[key]....：监控一个或多个key，当事务未执行时，如果监控的key的值发生了改变，事务便不会执行
先设定两个初值k1 k2，监控k1，然后进行事务
在事务未执行时，在另一终端标签页对k1 k2进行修改
再回到原来的页面执行事务，发现事务并没有执行，使用unwatch后继续监控key
然后再使用unwatch命令解除监控，即可再次监控key
# SpringBoot整合Redis
SpringBoot 2.x之后，原来使用的jedis被替换为了lettuce
- jedis:采用客户端直连，多个线程操作的话，是不安全的，如果想要避免不安全的，使用jedis pool连接池!BIO模式
- lettuce：采用netty，实例可以在多个线程中进程共享，不存在线程不安全的情况，可以减少线程数据。更像NIO模式
# 编写redis配置类,对key value进行序列化(实体类需实现Serializable接口)
/**
* @author zdk
* @date 2021/7/18 21:08
*/
@Configuration
public class RedisConfig {
@Bean("myRedisTemplate")
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<String, Object> template = new RedisTemplate<>();
template.setConnectionFactory(redisConnectionFactory);
Jackson2JsonRedisSerializer jackson2JsonRedisSerializer=new Jackson2JsonRedisSerializer(Object.class);
ObjectMapper objectMapper=new ObjectMapper();
objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
//key采用string的序列化方式
template.setKeySerializer(stringRedisSerializer);
//value采用Jackson序列化方式
template.setValueSerializer(jackson2JsonRedisSerializer);
//hash的key也采用string的序列化方式
template.setHashKeySerializer(stringRedisSerializer);
//value序列化方式采用Jackson
template.setHashValueSerializer(jackson2JsonRedisSerializer);
template.afterPropertiesSet();
return template;
}
@Bean("myStringRedisTemplate")
public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory) {
StringRedisTemplate template = new StringRedisTemplate();
template.setConnectionFactory(redisConnectionFactory);
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
//key采用string的序列化方式
template.setKeySerializer(stringRedisSerializer);
//value采用string的序列化方式
template.setValueSerializer(stringRedisSerializer);
template.afterPropertiesSet();
return template;
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# 编写redis常用操作工具类
/**
* @author zdk
* @date 2021/7/19 9:56
*/
@Component
public class RedisUtil {
@Autowired
@Qualifier("myRedisTemplate")
private RedisTemplate<String, Object> redisTemplate;
@Autowired
@Qualifier("myStringRedisTemplate")
private StringRedisTemplate stringRedisTemplate;
// =============================common============================
/**
* 指定缓存失效时间
* @param key 键
* @param time 时间(秒)
* @return
*/
public Boolean expire(String key, long time) {
try {
if (time > 0) {
redisTemplate.expire(key, time, TimeUnit.SECONDS);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据key 获取过期时间
* @param key 键 不能为null
* @return 时间(秒) 返回0代表为永久有效
*/
public Long getExpire(String key) {
return redisTemplate.getExpire(key, TimeUnit.SECONDS);
}
/**
* 判断key是否存在
* @param key 键
* @return true 存在 false不存在
*/
public Boolean hasKey(String key) {
try {
return redisTemplate.hasKey(key);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除缓存
* @param key 可以传一个值 或多个
*/
@SuppressWarnings("unchecked")
public void del(String... key) {
if (key != null && key.length > 0) {
if (key.length == 1) {
redisTemplate.delete(key[0]);
} else {
redisTemplate.delete((Collection<String>) CollectionUtils.arrayToList(key));
}
}
}
// ============================String=============================
/**
* 普通缓存获取
* @param key 键
* @return 值
*/
public String get(String key) {
return key == null ? null : stringRedisTemplate.opsForValue().get(key);
}
/**
* 普通缓存放入
* @param key 键
* @param value 值
* @return true成功 false失败
*/
public boolean set(String key, String value) {
try {
stringRedisTemplate.opsForValue().set(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 普通缓存放入并设置时间
* @param key 键
* @param value 值
* @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期
* @return true成功 false 失败
*/
public boolean set(String key, String value, long time) {
try {
if (time > 0) {
stringRedisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS);
} else {
set(key, value);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 递增
* @param key 键
* @param delta 要增加几(大于0)
* @return
*/
public Long incr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递增因子必须大于0");
}
return stringRedisTemplate.opsForValue().increment(key, delta);
}
/**
* 递减
* @param key 键
* @param delta 要减少几(小于0)
* @return
*/
public Long decr(String key, long delta) {
if (delta < 0) {
throw new RuntimeException("递减因子必须大于0");
}
return stringRedisTemplate.opsForValue().increment(key, -delta);
}
// ================================Map=================================
/**
* HashGet
* @param key 键 不能为null
* @param item 项 不能为null
* @return 值
*/
public Object hget(String key, String item) {
return redisTemplate.opsForHash().get(key, item);
}
/**
* 获取hashKey对应的所有键值
* @param key 键
* @return 对应的多个键值
*/
public Map<Object, Object> hmget(String key) {
return redisTemplate.opsForHash().entries(key);
}
/**
* HashSet
* @param key 键
* @param map 对应多个键值
* @return true 成功 false 失败
*/
public boolean hmset(String key, Map<String, Object> map) {
try {
redisTemplate.opsForHash().putAll(key, map);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* HashSet 并设置时间
* @param key 键
* @param map 对应多个键值
* @param time 时间(秒)
* @return true成功 false失败
*/
public boolean hmset(String key, Map<String, Object> map, long time) {
try {
redisTemplate.opsForHash().putAll(key, map);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
* @param key 键
* @param item 项
* @param value 值
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value) {
try {
redisTemplate.opsForHash().put(key, item, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 向一张hash表中放入数据,如果不存在将创建
* @param key 键
* @param item 项
* @param value 值
* @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间
* @return true 成功 false失败
*/
public boolean hset(String key, String item, Object value, long time) {
try {
redisTemplate.opsForHash().put(key, item, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 删除hash表中的值
* @param key 键 不能为null
* @param item 项 可以使多个 不能为null
*/
public void hdel(String key, Object... item) {
redisTemplate.opsForHash().delete(key, item);
}
/**
* 判断hash表中是否有该项的值
* @param key 键 不能为null
* @param item 项 不能为null
* @return true 存在 false不存在
*/
public boolean hHasKey(String key, String item) {
return redisTemplate.opsForHash().hasKey(key, item);
}
/**
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
* @param key 键
* @param item 项
* @param by 要增加几(大于0)
* @return
*/
public double hincr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, by);
}
/**
* hash递减
* @param key 键
* @param item 项
* @param by 要减少记(小于0)
* @return
*/
public double hdecr(String key, String item, double by) {
return redisTemplate.opsForHash().increment(key, item, -by);
}
// ============================set=============================
/**
* 根据key获取Set中的所有值
* @param key 键
* @return
*/
public Set<Object> sGet(String key) {
try {
return redisTemplate.opsForSet().members(key);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 根据value从一个set中查询,是否存在
* @param key 键
* @param value 值
* @return true 存在 false不存在
*/
public Boolean sHasKey(String key, Object value) {
try {
return redisTemplate.opsForSet().isMember(key, value);
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将数据放入set缓存
* @param key 键
* @param values 值 可以是多个
* @return 成功个数
*/
public Long sSet(String key, Object... values) {
try {
return redisTemplate.opsForSet().add(key, values);
} catch (Exception e) {
e.printStackTrace();
return (long) 0;
}
}
/**
* 将set数据放入缓存
* @param key 键
* @param time 时间(秒)
* @param values 值 可以是多个
* @return 成功个数
*/
public Long sSetAndTime(String key, long time, Object... values) {
try {
Long count = redisTemplate.opsForSet().add(key, values);
if (time > 0) {
expire(key, time);
}
return count;
} catch (Exception e) {
e.printStackTrace();
return (long)0;
}
}
/**
* 获取set缓存的长度
* @param key 键
* @return
*/
public Long sGetSetSize(String key) {
try {
return redisTemplate.opsForSet().size(key);
} catch (Exception e) {
e.printStackTrace();
return (long)0;
}
}
/**
* 移除值为value的
* @param key 键
* @param values 值 可以是多个
* @return 移除的个数
*/
public Long setRemove(String key, Object... values) {
try {
Long count = redisTemplate.opsForSet().remove(key, values);
return count;
} catch (Exception e) {
e.printStackTrace();
return (long)0;
}
}
// ===============================list=================================
/**
* 获取list缓存的内容
* @param key 键
* @param start 开始
* @param end 结束 0 到 -1代表所有值
* @return
*/
public List<Object> lGet(String key, long start, long end) {
try {
return redisTemplate.opsForList().range(key, start, end);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 获取list缓存的长度
* @param key 键
* @return
*/
public Long lGetListSize(String key) {
try {
return redisTemplate.opsForList().size(key);
} catch (Exception e) {
e.printStackTrace();
return (long)0;
}
}
/**
* 通过索引 获取list中的值
* @param key 键
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
* @return
*/
public Object lGetIndex(String key, long index) {
try {
return redisTemplate.opsForList().index(key, index);
} catch (Exception e) {
e.printStackTrace();
return null;
}
}
/**
* 将list放入缓存
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, Object value) {
try {
redisTemplate.opsForList().rightPush(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, Object value, long time) {
try {
redisTemplate.opsForList().rightPush(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
* @param key 键
* @param value 值
* @return
*/
public boolean lSet(String key, List<Object> value) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 将list放入缓存
*
* @param key 键
* @param value 值
* @param time 时间(秒)
* @return
*/
public boolean lSet(String key, List<Object> value, long time) {
try {
redisTemplate.opsForList().rightPushAll(key, value);
if (time > 0) {
expire(key, time);
}
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 根据索引修改list中的某条数据
* @param key 键
* @param index 索引
* @param value 值
* @return
*/
public boolean lUpdateIndex(String key, long index, Object value) {
try {
redisTemplate.opsForList().set(key, index, value);
return true;
} catch (Exception e) {
e.printStackTrace();
return false;
}
}
/**
* 移除N个值为value
* @param key 键
* @param count 移除多少个
* @param value 值
* @return 移除的个数
*/
public Long lRemove(String key, long count, Object value) {
try {
return redisTemplate.opsForList().remove(key, count, value);
} catch (Exception e) {
e.printStackTrace();
return (long)0;
}
}
// ===============================zset=================================
public Boolean zAdd(String key,Object value,double score){
try {
return redisTemplate.opsForZSet().add(key, value, score);
}catch (Exception e){
e.printStackTrace();
return false;
}
}
public Set<Object> zRange(String key,long start,long stop){
return redisTemplate.opsForZSet().range(key, start, stop);
}
public Set<Object> zRangeByScore(String key,double min,double max){
return redisTemplate.opsForZSet().rangeByScore(key, min, max);
}
public Set<Object> zRangeAll(String key){
return redisTemplate.opsForZSet().range(key, 0, -1);
}
}
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
# redis各种数据结构使用场景总结
- String：value除了可以是字符串还能是数字
- 计数器：incr
- 统计多单位的数量
- 粉丝数
- 对象缓存存储
- List：
- 消息排队
- 消息队列(Lpush,Rpop)
- 栈(Lpush,Lpop)
- Set：
- 将每个用户关注的人放入一个set，粉丝放入一个set，可以使用sinter获取他们的共同关注
- 共同爱好
- hash
- 更适合储存对象(易变更的数据)
- zset
- set排序
- 储存班级成绩表、工资表排序
- 消息排序，用权重判断是否重要
- 排行榜应用实现，取TopN测试
- Geospatial
- 地理位置
- 附近的人
- bitmap
- 网站、游戏签到
- 网站每天访问人数统计
- 指定用户是否在某一天访问了
# redis持久化
RDB(Redis DataBase)
RDB保存的文件是dump.rdb在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是Snapshot快照，它恢复时是将快照文件直接读到内存里。
Redis会单独创建 ( fork )一个子进程来进行持久化，会先将数据写入到一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。
整个过程中，主进程是不进行任何I0操作的。这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。
触发规则：
- save的规则满足的情况下，会自动触发rdb规则
- 执行flushall命令，也会触发rdb
- 退出redis服务，也会产生rdb文件
备份就自动生成一个dump.rdb
如何恢复数据：
只需将rdb文件放在redis启动目录下，redis会自动检查恢复dump.rdb中的数据
查看需要存在的位置
127.0.0.1:6379> config get dir 1) "dir" 2) "/usr/local/redis/bin" #如果在此目录下存在dump.rdb文件，redis会自动恢复其中的数据1
2
3
优点：
- 适合大规模的数据恢复
- 适用于对数据完整性要求不高的情况
缺点：
- 需要一定的时间间隔进行操作。如果redis意外宕机，最后一次修改的数据就没了
- fork进程的时候，会占用一定的内存
AOF(Append Only File)
将所有写入的命令都记录下来，储存在文件中，恢复的时候就把文件中的命令全部执行一次。 以日志的形式来记录每个写操作，将Redis执行过的所有指令记录下来（读操作不记录），只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作。
AOF保存的是appendonly.aof文件
AOF在redis.conf中的配置
appendonly no #默认是不开启的 # The name of the append only file (default: "appendonly.aof") appendfilename "appendonly.aof" #保存的文件名1
2
3#AOF持久化策略 appendfsync always #每次修改都会执行一次sync，会消耗更多性能 appendfsync everysec #每一秒执行一次sync，可能会丢失这一秒的数据 appendfsync no #不执行sync，这时操作系统自己同步数据，速度最快1
2
3
4
5#自动重写的规则 no-appendfsync-on-rewrite no #是否开启文件重写 auto-aof-rewrite-percentage 100 #重写的精确值 #重写的文件最小值，如果文件超过这个值，就会fork一个新的进程来将文件进行重写 auto-aof-rewrite-min-size 64mb1
2
3
4
5
6开启appendonly yes后重启redis即可生效，重启时如果没有appendonly.aof文件，会自动生成一个 如果原来的aof文件有错误，此时redis是不能启动的。需要运行 redis-check-aof --fix appendonly.aof1
2
3优点：
- 每次修改都会同步，数据的完整性更好
- 每秒同步一次，只可能会丢失一秒的数据
- 从不同步的话，效率最高
缺点：
- aof文件远大于rdb，修复速度更慢
- aof是io操作，运行效率也要比rdb慢，所以redis默认配置是rdb而不是aof
扩展：
- RDB持久化方式能够在指定的时间间隔内对你的数据进行快照存储
- AOF持久化方式记录每次对服务器写的操作，当服务器重启的时候会重新执行这些命令来恢复原始的数据，AOF命令以Redis 协议追加保存每次写的操作到文件末尾，Redis还能对AOF文件进行后台重写，使得AOF文件的体积不至于过大。
- 只做缓存，如果你只希望你的数据在服务器运行的时候存在，你也可以不使用任何持久化
- 同时开启两种持久化方式
- 在这种情况下，当redis重启的时候会优先载入AOF文件来恢复原始的数据，因为在通常情况下AOF文件保存的数据集要比RDB文件保存的数据集要完整。
- RDB的数据不实时，同时使用两者时服务器重启也只会找AOF文件，那要不要只使用AOF呢?作者建议不要，因为RDB更适合用于备份数据库(AOF在不断变化不好备份），快速重启，而且不会有AOF可能潜在的Bug，留着作为一个万一的手段。
- 性能建议
- 因为RDB文件只用作后备用途，建议只在Slave上持久化RDB文件，而且只要15分钟备份一次就够了，只保留save 900 1这条规则。
- 如果Enable AOF，好处是在最恶劣情况下也只会丢失不超过两秒数据，启动脚本较简单只load自己的AOF文件就可以了，代价一是带来了持续的lO，二是AOFrewrite的最后将rewrite过程中产生的新数据写到新文件造成的阻塞几乎是不可避免的。只要硬盘许可，应该尽量减少AOF rewrite的频率，AOF重写的基础大小默认值64M太小了，可以设到5G以上，默认超过原大小100%大小重写可以改到适当的数值。
- 如果不Enable AOF，仅靠Master-Slave Repllcation实现高可用性也可以，能省掉一大笔IO，也减少了rewrite时带来的系统波动。代价是如果Master/Slave同时倒掉，会丢失十几分钟的数据，启动脚本也要比较两个Master/Slave 中的 RDB文件，载入较新的那个，微博就是这种架构。
# Redis的发布和订阅
# 什么是发布和订阅
Redis发布订阅(pub/sub)是一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息
Redis客户端可以订阅任意数量的频道
2
# 发布订阅命令行实现(广发用于构建即时通信应用，如网络聊天室(chatroom)和实时广播、实时提醒等)
- PSUBSCRIBE pattern [pattern ...]：订阅一个或多个符合给定模式的频道
- PUBSUB subcommand [argument [argument ...]]：查看订阅和发布系统状态
- PUBLISH channel message：将消息发送到指定的频道
- PUNSUBSCRIBE [pattern [pattern ...]]：退订所有给定模式的频道
- SUBSCRIBE channel [channel...]：订阅给定的一个或多个频道的信息
- UNSUBSCRIBE [channel [channel...]]：退订给定的频道
打开一个客户端订阅一个频道为channel1
SUBSCRIBE channel11
打开另一个客户端，给channel1发布消息hello
publish channel1 hello1
发布成功返回的是订阅者的数量
订阅了channel1的订阅者收到消息
注：发布的消息没有持久化，没有订阅的客户端收不到hello，只能收到订阅后channel1发布的消息
发布订阅的原理：
笔记
Redis是使用C实现的，通过分析Redis 源码里的pubsub.c文件，了解发布和订阅机制的底层实现，籍此加深对Redis的理解。Redis 通过PUBLISH、SUBSCRIBE和PSUBSCRIBE等命令实现发布和订阅功能。 通过SUBSCRIBE命令订阅某频道后，redis-server里维护了一个字典，字典的键就是一个个channel，而字典的值则是一个链表，链表中保存了所有订阅这个channel的客户端。SUBSCRIBE命令的关键，就是将客户端添加到给定channel的订阅链表中。通过PUBLISH命令向订阅者发送消息，redis-server 会使用给定的频道作为键，在它所维护的channel字典中查找记录了订阅这个频道的所有客户端的链表，遍历这个链表，将消息发布给所有订阅者。 Pub/Sub从字面上理解就是发布( Publish )与订阅(Subscribe )，在Redis中，你可以设定对某一个key值进行消息发布及消息订阅，当一个key值上进行了消息发布后，所有订阅它的客户端都会收到相应的消息。这一功能最明显的用法就是用作实时消息系统，比如普通的即时聊天，群聊等功能。
使用场景：
- 实时消息系统(网站的消息推送)
- 实时聊天(频道当做聊天室，将信息回显给所有人)
- 订阅，关注系统
更复杂的场景就会使用消息中间件：MQ 等
