Redis

该文档是：Redis学习...带*号的是非常重要的...

博客连接：https://www.loveuluo.cn

日期：2021-01-26

1. NoSQL数据模型简介

以一个电商客户、订单、订购、地址模型来对比关系型数据库和非关系型数据库

• 传统关系型数据库如何设计

  • ER图（1：1、1：N、N：1）主外键等

  • NOSQL如何设计

    • BSON ()是一种类json的一种二进制形式的存储格式，简称Binary JSON，它和JSON一样，支持内嵌的文档对象和数组对象

  • 两者对比，问题和难点

    • 问题和难点

      • 为什么用聚合模型来处理

        • 高并发的操作是不太建议用关联查询的，互联网公司用冗余数据来避免关联查询
        • 分布式事务是支持不了太多的并发的

聚合模型

• KV
• BSON

• 列族

  • 顾名思义，是按列存储数据的。最大的特点是方便存储结构化和半结构化数据，方便做数据压缩，对针对某一 列或者某几列的查询有非常大的IO优势。

• 图形

2. NoSQL数据库的四大分类

KV

• 新浪：BerkeleyDB + Redis
• 美团：Redis + tair
• 阿里、百度：memcache + Redis

文档型数据库（bson格式比较多）

• CouchDB

• MongoDB

  • MongoDB是一个基于分布式文件存储的数据库。由C++语言编写。旨在为WEB应用提供可扩展的高性能数据存储解决方案。
  • MongoDB是一个介于关系数据库和非关系数据库之间的产品，是非关系数据库当中功能最丰富，最像关系数据库的。

列存储数据库

• Cassandra、HBase
• 分布式文件系统

图关系数据库

• 它不是放图形的、放的是关系比如：朋友圈社交网络、广告推荐系统
• 社交网络、推荐系统。专注于构建关系图谱
• Neo4j、InfoGrid

四者对比

3. 分布式数据库CAP+BASE原理*

3.1 传统的ACID分别是什么

• A (Atomicity) 原子性
• C (Consistency) 一致性
• I (Isolation) 独立性
• D (Durability) 持久性

关系型数据库遵循ACID规则，事务在英文中是transaction，和现实世界中的交易很类似，它有如下四个特性：

1、A (Atomicity) 原子性 原子性很容易理解，也就是说事务里的所有操作要么全部做完，要么都不做，事务成功的条件是事务里的所有操作都成功，只要有一个操作失败，整个事务就失败，需要回滚。比如银行转账，从A账户转100元至B账户，分为两个步骤：1）从A账户取100元；2）存入100元至B账户。这两步要么一起完成，要么一起不完成，如果只完成第一步，第二步失败，钱会莫名其妙少了100元。

2、C (Consistency) 一致性 一致性也比较容易理解，也就是说数据库要一直处于一致的状态，事务的运行不会改变数据库原本的一致性约束。

3、I (Isolation) 独立性 所谓的独立性是指并发的事务之间不会互相影响，如果一个事务要访问的数据正在被另外一个事务修改，只要另外一个事务未提交，它所访问的数据就不受未提交事务的影响。比如现有有个交易是从A账户转100元至B账户，在这个交易还未完成的情况下，如果此时B查询自己的账户，是看不到新增加的100元的

4、D (Durability) 持久性 持久性是指一旦事务提交后，它所做的修改将会永久的保存在数据库上，即使出现宕机也不会丢失。

3.2 CAP

• C:Consistency（强一致性）
• A:Availability（可用性）
• P:Partition tolerance（分区容错性）

CAP理论就是说在分布式存储系统中，最多只能实现上面的两点。

而由于当前的网络硬件肯定会出现延迟丢包等问题，所以分区容忍性是我们必须需要实现的。所以我们只能在一致性和可用性之间进行权衡，没有NoSQL系统能同时保证这三点。

• CA 传统Oracle数据库
• AP 大多数网站架构的选择
• CP Redis、Mongodb

注意：分布式架构的时候必须做出取舍。

一致性和可用性之间取一个平衡。多余大多数web应用，其实并不需要强一致性。因此牺牲C换取P，这是目前分布式数据库产品的方向。

一致性与可用性的决择

对于web2.0网站来说，关系数据库的很多主要特性却往往无用武之地

数据库事务一致性需求

很多web实时系统并不要求严格的数据库事务，对读一致性的要求很低， 有些场合对写一致性要求并不高。允许实现最终一致性。

数据库的写实时性和读实时性需求

对关系数据库来说，插入一条数据之后立刻查询，是肯定可以读出来这条数据的，但是对于很多web应用来说，并不要求这么高的实时性，比方说在微博发一条消息之后，过几秒乃至十几秒之后，我的订阅者才看到这条动态是完全可以接受的。

对复杂的SQL查询，特别是多表关联查询的需求

任何大数据量的web系统，都非常忌讳多个大表的关联查询，以及复杂的数据分析类型的报表查询，特别是SNS类型的网站，从需求以及产品设计角 度，就避免了这种情况的产生。往往更多的只是单表的主键查询，以及单表的简单条件分页查询，SQL的功能被极大的弱化了。

3.3 经典CAP图*

• CA - 单点集群，满足一致性，可用性的系统，通常在可扩展性上不太强大。
• CP - 满足一致性，分区容忍必的系统，通常性能不是特别高。
• AP - 满足可用性，分区容忍性的系统，通常可能对一致性要求低一些（大多数网站的选择）。

3.4 BASE

BASE就是为了解决关系数据库强一致性引起的问题而引起的可用性降低而提出的解决方案。

BASE其实是下面三个术语的缩写：

• 基本可用（Basically Available）
• 软状态（Soft state）
• 最终一致（Eventually consistent）

它的思想是通过让系统放松对某一时刻数据一致性的要求来换取系统整体伸缩性和性能上改观。为什么这么说呢，缘由就在于大型系统往往由于地域分布和极高性能的要求，不可能采用分布式事务来完成这些指标，要想获得这些指标，我们必须采用另外一种方式来完成，这里BASE就是解决这个问题的办法

3.5 分布式+集群简介

分布式系统（distributed system）

由多台计算机和通信的软件组件通过计算机网络连接（本地网络或广域网）组成。分布式系统是建立在网络之上的软件系统。正是因为软件的特性，所以分布式系统具有高度的内聚性和透明性。因此，网络和分布式系统之间的区别更多的在于高层软件（特别是操作系统），而不是硬件。分布式系统可以应用在在不同的平台上如：PC、工作站、局域网和广域网上等。

简单来讲：

1. 分布式：不同的多台服务器上面部署不同的服务模块（工程），他们之间通过Rpc/Rmi之间通信和调用，对外提供服务和组内协作。
2. 集群：不同的多台服务器上面部署相同的服务模块，通过分布式调度软件进行统一的调度，对外提供服务和访问。

4. Redis的安装

4.1 入门概述

是什么：

Redis:REmote DIctionary Server(远程字典服务器)是完全开源免费的，用C语言编写的，遵守BSD协议，是一个高性能的(key/value)分布式内存数据库，基于内存运行 并支持持久化的NoSQL数据库，是当前最热门的NoSql数据库之一，也被人们称为数据结构服务器。

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

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

能干嘛：

• 内存存储和持久化：redis支持异步将内存中的数据写到硬盘上，同时不影响继续服务
• 取最新N个数据的操作，如：可以将最新的10条评论的ID放在Redis的List集合里面
• 模拟类似于HttpSession这种需要设定过期时间的功能
• 发布、订阅消息系统
• 定时器、计数器

去哪下：

• Redis官网
• Redis中文网

怎么玩：

• 数据类型、基本操作和配置
• 持久化和复制，RDB/AOF
• 事务的控制
• 复制（主从关系）

4.2 Redis的安装

Linux版安装：

下载安装包放入opt文件夹下。

/opt目录下，解压命令tar -zxvf redis-3.0.4.tar.gz。

进入redis目录cd redis-3.0.4。

在redis-3.0.4目录下执行make命令进行安装。

运行make命令时如果出现错误。

编译成功后，进入src文件夹，执行make install进行redis安装

Windows版安装：

Redis X.X.X for Windows

大多数企业是使用Linux的Redis，Windows仅供学习。

5. HelloWorld

先对redis的配置文件进行备份。

用vim redis.conf命令修改配置文件。

用ps -ef|grep redis命令看看后台redis进程有没有启动。

用cd /usr/local/bin进入目录。

用redis-server命令指定配置文件运行redis在后台启动，然后redis-cli -p代表指定端口为6379连接到redis。

再用ps -ef|grep redis命令看看后台redis进程有没有启动。

HelloWord测试并关闭redis后台服务，并退出redis客户端。

6. 启动后杂项基础知识

• redis-benchmark可以测试redis运行性能

• 单进程

  • 单进程模型来处理客户端的请求。对读写等事件的响应 是通过对epoll函数的包装来做到的。Redis的实际处理速度完全依靠主进程的执行效率
  • Epoll是Linux内核为处理大批量文件描述符而作了改进的epoll，是Linux下多路复用IO接口select/poll的增强版本， 它能显著提高程序在大量并发连接中只有少量活跃的情况下的系统CPU利用率。
• 默认16个数据库，类似数组下表从0开始，初始默认使用1号库，可在配置文件配置

• select命令切换数据库

• dbsize查看当前数据库的key的数量
• keys *查看所有的key

• flushdb：清空当前库
• flushall；通杀全部库
• 统一密码管理，16个库都是同样密码，要么都OK要么一个也连接不上
• Redis索引都是从零开始
• 为什么默认端口是6379

7. 常用五大数据类型简介

• String（字符串）

  • string是redis最基本的类型，你可以理解成与Memcached一模一样的类型，一个key对应一个value。
  • string类型是二进制安全的。意思是redis的string可以包含任何数据。比如jpg图片或者序列化的对象 。
  • string类型是Redis最基本的数据类型，一个redis中字符串value最多可以是512M

• Hash（哈希，类似java里的Map）

  • Redis hash 是一个键值对集合。
  • Redis hash是一个string类型的field和value的映射表，hash特别适合用于存储对象。
  • 类似Java里面的Map<String,Object>

• List（列表）

  • Redis 列表是简单的字符串列表，按照插入顺序排序。你可以添加一个元素导列表的头部（左边）或者尾部（右边）。
  • 它的底层实际是个链表

• Set（集合）

  • Redis的Set是string类型的无序集合。它是通过HashTable实现实现的

• Zset(sorted set：有序集合)

  • Redis zset 和 set 一样也是string类型元素的集合，且不允许重复的成员。
  • 不同的是每个元素都会关联一个double类型的分数。
  • redis正是通过分数来为集合中的成员进行从小到大的排序。zset的成员是唯一的，但分数(score)却可以重复。

• 哪里去获得redis常见数据类型操作命令

  • Redis 命令参考
  • Redis 官网命令参考

8. Key

8.1 常用的

8.2 案例

• keys * 查询所有的key
• exists key，判断某个key是否存在
• move key db —>从当前库移到另一个库
• expire key 秒钟：为给定的key设置过期时间
• ttl key 查看还有多少秒过期，-1表示永不过期，-2表示已过期
• type key 查看你的key是什么类型

9. String*

单值单value。

9.1 常用

9.2 案例

• set/get/del
• append： 追加到后边

• strlen： 获取值的长度

• Incr(递增)/decr(递减)/incrby 数字(递增多少)/decrby： 数字(递减多少),一定要是数字才能进行加减

• getrange/setrange

• setex(set with expire)键秒值

• setnx(如果不存在)

• mset(批量插入)/mget(批量获取)/msetnx(如果有一个存在,全部都不插入)

• getset(先get再set)

10. List*

单值多value。

10.1 常用

10.2 案例

当成一个双向链表来看待。

• lpush(往左边添加元素)/rpush(往右边添加元素)/lrange

• lpop/rpop :移除一个元素

• lindex：按照索引下标获得元素(从左到右)

• llen：获取长度
• lrem key 删N个value

• ltrim key 开始index 结束index，截取指定范围的值后再赋值给key

• rpoplpush 源列表 目的列表 ：把源列表最右边的元素 放到目的列表的最左边

• lset key index value

• linsert key before/after 值1 值2

性能总结：

• 它是一个字符串链表，left、right都可以插入添加；
• 如果键不存在，创建新的链表；
• 如果键已存在，新增内容；
• 如果值全移除，对应的键也就消失了。
• 链表的操作无论是头和尾效率都极高，但假如是对中间元素进行操作，效率就很惨淡了。

11. Set

单值多value

11.1 常用

11.2 案例

• sadd(添加)/smembers(查询)/sismember(判断里边有没有某值)

• scard：获取集合里面的元素个数
• srem key value：删除集合中元素

• srandmember key：某个整数(随机出几个数)

• spop key：随机删除一个

• smove key1 key2：在key1里某个值 作用是将key1里的某个值赋给key2

• 数学集合类

  • 差集：sdiff

• 交集：sinter

• 并集：sunion

12. Hash*

KV模式不变，但V是一个键值对

12.1 常用

12.2 案例

• hset/hget

• hmset/hmget/hgetall

• hdel

• hlen

• hexists key ：在key里面的某个值的key存不存在

• hkeys/hvals

• hincrby/hincrbyfloat

• hsetnx：不存在就放进去

13. ZSet

在set基础上，加一个score值。 之前set是k1 v1 v2 v3， 现在zset是k1 score1 v1 score2 v2

13.1 常用

13.2 案例

• zadd/zrange

  • Withscores

• zrangebyscore key 开始score 结束score

  • withscores
  • ( 不包含

  

  • Limit 作用是返回限制

    • limit 开始下标步 多少步
• zrem key 某score下对应的value值，作用是删除元素

• zcard
• 
• zcount key score区间/zrank key values值，作用是获得下标值/zscore key 对应值,获得分数

• zrevrank key values值，作用是逆序获得下标值

• zrevrange

• zrevrangebyscore key 结束分数到开始分数

14. 配置文件介绍

Redis 的配置文件位于 Redis 安装目录下，文件名为 redis.conf(Windows 名为 redis.windows.conf)。

你可以通过 CONFIG 命令查看或设置配置项，也可以直接使用vim 编辑器进行编辑。

语法

Redis CONFIG 命令格式如下：

redis 127.0.0.1:6379> CONFIG GET CONFIG_SETTING_NAME

实例1

redis 127.0.0.1:6379> CONFIG GET loglevel

1) "loglevel"
2) "notice"

实例2：config命令设置密码

14.1 参数说明

redis.conf 配置项说明如下：

注意，下面配置项说明并不全，了解更多请检阅redis.conf。

15. 持久化之RDB*

RDB（Redis DataBase）

15.1 是什么

• 在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里
• Redis会单独创建（fork）一个子进程来进行持久化，会先将数据写入到 一个临时文件中，待持久化过程都结束了，再用这个临时文件替换上次持久化好的文件。 整个过程中，主进程是不进行任何IO操作的，这就确保了极高的性能。如果需要进行大规模数据的恢复，且对于数据恢复的完整性不是非常敏感，那RDB方式要比AOF方式更加的高效。RDB的缺点是最后一次持久化后的数据可能丢失。

Fork

Fork的作用是复制一个与当前进程一样的进程。新进程的所有数据（变量、环境变量、程序计数器等） 数值都和原进程一致，但是是一个全新的进程，并作为原进程的子进程

• rdb 保存的是dump.rdb文件
• 相关配置在配置文件的位置 - 在redis.conf搜寻### SNAPSHOTTING ###

15.2 如何触发RDB快照

• RDB是整个内存的压缩过的Snapshot(快照)，RDB的数据结构，可以配置复合的快照触发条件，默认为：

• 执行flushall（清空key）命令，也会产生dump.rdb文件，但里面是空的，无意义。
• 使用shutdown命令的话也会产生一个dump.rdb文件。

• 直接执行命令save或者是bgsave

  • Save：save时只管保存，其它不管，全部阻塞（马上备份，别的活都不能干直到保存完毕）
  • BGSAVE：Redis会在后台异步进行快照操作， 快照同时还可以响应客户端请求。可以通过lastsave 命令获取最后一次成功执行快照的时间

15.3 如何恢复

企业中一般干活的机器和备份数据的机器是两台机器。

• copy一份dump.rdb文件模拟放到了备份的机器上

• 配置文件中默认的快照配置(默认生成的和读回内存的文件名称为dump.rdb)。此时如果要恢复之前的数据，只需要把dump_bk.rdb拿过来(放到如上图redis安装目录)改名为dump.rdb覆盖掉现有的，然后启动服务器即可。

• 将备份文件 dump_bk.rdb改名为dump.rdb) 移动到 redis 安装目录并启动服务即可。
• CONFIG GET dir可以获取目录

15.4 其余的配置

• stop-writes-on-bgsave-error

• rdbcompression

• rdbchecksum

15.5 优势与劣势

• 优势

  • 适合大规模的数据恢复
  • 对数据完整性和一致性要求不高

• 劣势

  • 在一定间隔时间做一次备份，所以如果redis意外down掉的话，就 会丢失最后一次快照后的所有修改
  • Fork的时候，内存中的数据被克隆了一份，大致2倍的膨胀性需要考虑

15.6 如何停止

动态所有停止RDB保存规则的方法：redis-cli config set save ""

15.7 小结

• RDB是一个非常紧凑的文件。
• RDB在保存RDB文件时父进程唯一需要做的就是fork出一个子进程，接下来的工作全部由子进程来做，父进程不需要再做其他I0操作，所以RDB持久化方式可以最大化redis的性能。
• 与AOF相比，在恢复大的数据集的时候，RDB方式会更快一一些。
• 数据丢失风险大。
• RDB需要经常fork子进程来保存数据集到硬盘上，当数据集比较大的时候fork的过程是非常耗时的吗，可能会导致Redis在一些毫秒级不能回应客户端请求。

16. 持久化之AOF*

AOF（Append Only File）

16.1 是什么

以日志的形式来记录每个写操作，将Redis执行过的所有写指令记录下来(读操作不记录)， 只许追加文件但不可以改写文件，redis启动之初会读取该文件重新构建数据，换言之，redis 重启的话就根据日志文件的内容将写指令从前到后执行一次以完成数据的恢复工作

16.2 AOF配置

• 相关配置在配置文件的位置 - 在redis.conf搜寻### APPEND ONLY MODE ###

• aof保存的是appendonly.aof文件（在配置文件可修改文件名）

• appendfsync

16.3 AOF启动/修复/恢复

• 正常恢复

  • 启动：设置Yes

    • 修改默认的appendonly no，改为yes
  • 将有数据的aof文件复制一份保存到对应目录(config get dir)
  • 恢复：重启redis然后重新加载（跟RDB的一样，自动加载安装目录下的appendonly.aof）

• 异常恢复

  • 启动：设置Yes

    • 修改默认的appendonly no，改为yes
  • 备份被写坏的AOF文件

  • 修复：

    • Redis-check-aof --fix进行修复（也可以用redis-check-dump --fix修复dump）
  • 恢复：重启redis然后重新加载（跟RDB的一样，自动加载安装目录下的appendonly.aof）

16.4 rewrite

• 看看appendonly.aof文件里边的内容长啥样：

• rewrite是什么：

  • AOF采用文件追加方式，文件会越来越大。为避免出现此种情况，新增了重写机制， 当AOF文件的大小超过所设定的阈值时，Redis就会启动AOF文件的内容压缩， 只保留可以恢复数据的最小指令集。可以使用命令bgrewriteaof

• 重写原理

  • AOF文件持续增长而过大时，会fork出一条新进程来将文件重写(也是先写临时文件最后再rename)， 遍历新进程的内存中数据，每条记录有一条的Set语句。重写aof文件的操作，并没有读取旧的aof文件， 而是将整个内存中的数据库内容用命令的方式重写了一个新的aof文件，这点和快照有点类似

• 触发机制

  • Redis会记录上次重写时的AOF大小，默认配置是当AOF文件大小是上次rewrite后大小的一倍(100%)且文件大于64M时触发

16.5 优势与劣势

• 优势

  • 每修改同步：appendfsync always 同步持久化 每次发生数据变更会被立即记录到磁盘 性能较差但数据完整性比较好
  • 每秒同步：appendfsync everysec 异步操作，每秒记录 如果一秒内宕机，有数据丢失
  • 不同步：appendfsync no 从不同步

• 劣势

  • 相同数据集的数据而言aof文件要远大于rdb文件，恢复速度慢于rdb
  • Aof运行效率要慢于rdb,每秒同步策略效率较好，不同步效率和rdb相同

16.6 小结

• AOF文件时一个只进行追加的日志文件
• Redis可以在AOF文件体积变得过大时，自动地在后台对AOF进行重写
• AOF文件有序地保存了对数据库执行的所有写入操作，这些写入操作以Redis协议的格式保存，因此AOF文件的内容非常容易被人读懂，对文件进行分析也很轻松
• 对于相同的数据集来说，AOF文件的体积通常要大于RDB文件的体积
• 根据所使用的fsync 策略，AOF的速度可能会慢于RDB

16.7 AOF和RDB总结

性能建议：

17 .事务*

17.1 是什么

可以一次执行多个命令，本质是一组命令的集合。一个事务中的所有命令都会序列化，按顺序地串行化执行而不会被其它命令插入，不许加塞。

17.2 能干嘛

一个队列中，一次性、顺序性、排他性的执行一系列命令。

17.3 怎么玩

(1)常用命令

(2)案例

正常执行

放弃事务

全体连坐

类似Java的编译异常(语法就错误了)

冤头债主

类似Java的运行异常(运行时错误)

17.4 watch监控*

悲观锁/乐观锁/CAS(Check And Set)：

• 悲观锁

  • 悲观锁(Pessimistic Lock), 顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。

• 乐观锁

  • 乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量。
  • 乐观锁策略:提交版本必须大于记录当前版本才能执行更新
• CAS

信用卡可用余额和欠额：

• 初始化信用卡可用余额和欠额

• 无加塞篡改，先监控再开启multi， 保证两笔金额变动在同一个事务内

• 有加塞篡改

  • 监控了key，如果key的值被修改了，后面一个事务的执行失效

• unwatch

• 一旦执行了exec之前加的监控锁都会被取消掉了（一次性）

小结：

• Watch指令，类似乐观锁，事务提交时，如果Key的值已被别的客户端改变， 比如某个list已被别的客户端push/pop过了，整个事务队列都不会被执行
• 通过WATCH命令在事务执行之前监控了多个Keys，倘若在WATCH之后有任何Key的值发生了变化， EXEC命令执行的事务都将被放弃，同时返回Nullmulti-bulk应答以通知调用者事务执行失败

17.5 3阶段

• 开启：以MULTI开始一个事务
• 入队：将多个命令入队到事务中，接到这些命令并不会立即执行，而是放到等待执行的事务队列里面
• 执行：由EXEC命令触发事务

17.6 3特性

• 单独的隔离操作：事务中的所有命令都会序列化、按顺序地执行。事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
• 没有隔离级别的概念：队列中的命令没有提交之前都不会实际的被执行，因为事务提交前任何指令都不会被实际执行， 也就不存在”事务内的查询要看到事务里的更新，在事务外查询不能看到”这个让人万分头痛的问题
• 不保证原子性：redis同一个事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚

不遵循传统的ACID中的AI

18. 消息订阅发布简介(了解即可)

用观察者模式理解学习

进程间的一种消息通信模式：发送者(pub)发送消息，订阅者(sub)接收消息。

下图展示了频道 channel1 ， 以及订阅这个频道的三个客户端 —— client2 、 client5 和 client1 之间的关系：

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

18.1 常用命令

18.2 实例

以下实例演示了发布订阅是如何工作的。在我们实例中我们创建了订阅频道名为 redisChat:

redis 127.0.0.1:6379> SUBSCRIBE redisChat

Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redisChat"
3) (integer) 1

现在，我们先重新开启个 redis 客户端，然后在同一个频道 redisChat 发布两次消息，订阅者就能接收到消息。

redis 127.0.0.1:6379> PUBLISH redisChat "Redis is a great caching technique"

(integer) 1

redis 127.0.0.1:6379> PUBLISH redisChat "Learn redis by runoob.com"

(integer) 1

# 订阅者的客户端会显示如下消息
1) "message"
2) "redisChat"
3) "Redis is a great caching technique"
1) "message"
2) "redisChat"
3) "Learn redis by runoob.com"

订阅多个通配符 *

PSUBSCRIBE new*

收取消息，

PUBLISH new1 redis2015

19. 主从复制*

19.1 是什么

行话：也就是我们所说的主从复制，主机数据更新后根据配置和策略， 自动同步到备机的master/slaver机制，Master以写为主，Slave以读为主

19.2 能干嘛

• 读写分离
• 容灾恢复

19.3 怎么玩*

(1)准备工作

• 配从(库)不配主(库)

• 从库配置命令：

  slaveof 主库IP 主库端口

  • 每次与master断开之后，都需要重新连接，除非你配置进redis.conf文件（具体位置：redis.conf搜寻#### REPLICATION ####）
  • info replication

• 修改配置文件细节操作(这里以6379端口的为例)

  • 拷贝多个redis.conf文件，按’redis[port].conf’重命名

  

  • 开启daemonize yes
  • pid文件名字
  • 指定端口

  

  • log文件名字

  

  • dump.rdb名字

  

replication
英 [ˌreplɪ’keɪʃ(ə)n] 美 [ˌreplɪ’keɪʃ(ə)n]
n.
(绘画等的)复制;拷贝;重复(实验);(尤指对答辩的)回答

(2)常用3招

一主二仆(一台主机两台备机)

• 初始化

• 一个Master(主机)两个Slave(从机)

• 存在的问题

  • 问题一：主机写入数据后从机能修改吗？

• 问题二：主机挂了之后，两个从机是会上位成为主机还是原地待命？

  

• 问题三：主机又回来了后，主机新增记录，从机还能否顺利复制？

  

• 问题四：其中一台从机down后情况如何？依照原有它能跟上大部队吗？

• 日志查看

  • 主机日志
  • 备机日志

• 总结主从问题演示

  1. 切入点问题？slave1、slave2是从头开始复制还是从切入点开始复制?比如从k4进来，那之前的123是否也可以复制？

     • 答：从头开始复制；123也可以复制

  2. 从机是否可以写？set可否？

     • 答：从机不可写，不可set，主机可写

  3. 主机shutdown后情况如何？从机是上位还是原地待命

     • 答：从机还是原地待命（咸鱼翻身，还是咸鱼）

  4. 主机又回来了后，主机新增记录，从机还能否顺利复制？

     • 答：能

  5. 其中一台从机down后情况如何？依照原有它能跟上大部队吗？

     • 答：不能跟上，每次与master断开之后，都需要重新连接，除非你配置进redis.conf文件（具体位置：redis.conf搜寻#### REPLICATION ####）

薪火相传

• slaveof 新主库IP 新主库端口

• 上一个Slave可以是下一个slave的Master，Slave同样可以接收其他 slaves的连接和同步请求，那么该slave作为了链条中下一个的master, 可以有效减轻master的写压力（奴隶的奴隶还是奴隶）

• 中途变更转向：会清除之前的数据，重新建立拷贝最新的

反客为主

  SLAVEOF no one

• 使当前数据库停止与其他数据库的同步，转成主数据库

19.4 复制原理

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

19.5 哨兵模式(sentinel)*

一组sentinel能同时监控多个master

(1)是什么

反客为主的自动版，能够后台监控主机是否故障，如果故障了根据投票数自动将从库转换为主库

(2)怎么玩(使用步骤)

1. 调整结构，6379带着6380、6381(先重新变成一主两仆模式)
2. 新建sentinel.conf文件，名字绝不能错

1. 配置哨兵,填写内容

   • sentinel monitor 被监控数据库名字(自己起名字) 127.0.0.1 6379 1
   • 上面最后一个数字1，表示主机挂掉后salve投票看让谁接替成为主机，得票数多少后成为主机

   

2. 启动哨兵

   • redis-sentinel /sentinel.conf（目录依照各自的实际情况配置，可能目录不同）

   

   

3. 正常主从演示
4. 原有的master挂了
5. 投票新选
6. 重新主从继续开工，info replication查查看

问题：如果之前挂了的master重启回来，会不会双master冲突？
答： 不会，原master，变成slave

19.6 复制的缺点

复制延时

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

20. Jedis

20.1 测试连通

• 在idea创建普通Maven工程
• pom.xml引入jedis相关依赖
• 创建新类，测试连通，源码如下：

public class TestPing {
    public static void main(String[] args) {
        //如果要连接 阿里云上的redis记得安全组要放开6379端口
        Jedis jedis=new Jedis("47.99.xxx.xxx",6379);
        String ping = jedis.ping();
        //输出PONG，代表redis连通成功
        System.out.println(ping);
    }
}

20.2 常用API

public class TestAPI {
    public static void main(String[] args) {

        Jedis jedis = new Jedis("127.0.0.1", 6379);
        // key
        Set<String> keys = jedis.keys("*");
        for (Iterator iterator = keys.iterator(); iterator.hasNext();) {
            String key = (String) iterator.next();
            System.out.println(key);
        }
        System.out.println("jedis.exists====>" + jedis.exists("k2"));
        System.out.println(jedis.ttl("k1"));
        
        // String
        // jedis.append("k1","myreids");
        System.out.println(jedis.get("k1"));
        jedis.set("k4", "k4_redis");
        System.out.println("----------------------------------------");
        jedis.mset("str1", "v1", "str2", "v2", "str3", "v3");
        System.out.println(jedis.mget("str1", "str2", "str3"));
        
        
        // list
        System.out.println("----------------------------------------");
        // jedis.lpush("mylist","v1","v2","v3","v4","v5");
        List<String> list = jedis.lrange("mylist", 0, -1);
        for (String element : list) {
            System.out.println(element);
        }
        
        // set
        jedis.sadd("orders", "jd001");
        jedis.sadd("orders", "jd002");
        jedis.sadd("orders", "jd003");
        Set<String> set1 = jedis.smembers("orders");
        for (Iterator iterator = set1.iterator(); iterator.hasNext();) {
            String string = (String) iterator.next();
            System.out.println(string);
        }
        jedis.srem("orders", "jd002");
        System.out.println(jedis.smembers("orders").size());
        
        // hash
        jedis.hset("hash1", "userName", "lisi");
        System.out.println(jedis.hget("hash1", "userName"));
        Map<String, String> map = new HashMap<String, String>();
        map.put("telphone", "13811814763");
        map.put("address", "atguigu");
        map.put("email", "abc@163.com");
        jedis.hmset("hash2", map);
        List<String> result = jedis.hmget("hash2", "telphone", "email");
        for (String element : result) {
            System.out.println(element);
        }
        
        // zset
        jedis.zadd("zset01", 60d, "v1");
        jedis.zadd("zset01", 70d, "v2");
        jedis.zadd("zset01", 80d, "v3");
        jedis.zadd("zset01", 90d, "v4");

        Set<String> s1 = jedis.zrange("zset01", 0, -1);
        for (Iterator iterator = s1.iterator(); iterator.hasNext();) {
            String string = (String) iterator.next();
            System.out.println(string);
        }

    }
}

20.3 事务

(1)日常

public class TestRiChang {
    public static void main(String[] args) {
        Jedis jedis=new Jedis("47.99.xxx.xxx",6379);
        //multi：标记一个事务的开始，在这里multi会返回一个对象，用这个返回的对象来操作
        Transaction transaction = jedis.multi();
        transaction.set("k20","v20");
        transaction.set("k30","v30");
        //提交事务
        transaction.exec();
        //取消事务
//      transaction.discard();
    }
}

(2)加锁

import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

public class TestTX {
    public boolean transMethod() throws InterruptedException {
        Jedis jedis = new Jedis("127.0.0.1", 6379);
        int balance;// 可用余额
        int debt;// 欠额
        int amtToSubtract = 10;// 实刷额度

        jedis.watch("balance");
        
        //模拟线程睡眠，此时用redis客户直接修改金额如果改的 < 实刷额度直接unwatch了，甚至都不用开启事务啥啥啥的。如果>实刷金额，也就代表余额够会进入else中，但是因为被watch监控着。就算exec提交也不会生效。
        Thread.sleep(7000);
        
        balance = Integer.parseInt(jedis.get("balance"));
        if (balance < amtToSubtract) {
            jedis.unwatch();
            System.out.println("余额不足");
            return false;
        } else {
            System.out.println("***********transaction");
            Transaction transaction = jedis.multi();
            transaction.decrBy("balance", amtToSubtract);
            transaction.incrBy("debt", amtToSubtract);
            transaction.exec();
            balance = Integer.parseInt(jedis.get("balance"));
            debt = Integer.parseInt(jedis.get("debt"));

            System.out.println("*******" + balance);
            System.out.println("*******" + debt);
            return true;
        }
    }

    /**
     * 通俗点讲，watch命令就是标记一个键，如果标记了一个键， 在提交事务前如果该键被别人修改过，那事务就会失败，这种情况通常可以在程序中 重新再尝试一次。
     * 首先标记了键balance，然后检查余额是否足够，不足就取消标记，并不做扣减； 足够的话，就启动事务进行更新操作，
     * 如果在此期间键balance被其它人修改， 那在提交事务（执行exec）时就会报错， 程序中通常可以捕获这类错误再重新执行一次，直到成功。
     * 
     * @throws InterruptedException
     */
    public static void main(String[] args) throws InterruptedException {
        TestTX test = new TestTX();
        boolean retValue = test.transMethod();
        System.out.println("main retValue-------: " + retValue);
    }
}

20.4 主从复制

1. 6379,6380启动，先各自先独立
2. 主写，从读

import redis.clients.jedis.Jedis;

public class TestMS {
    public static void main(String[] args) {
        Jedis jedis_M = new Jedis("127.0.0.1", 6379);
        Jedis jedis_S = new Jedis("127.0.0.1", 6380);
        //从机连接主机
        jedis_S.slaveof("127.0.0.1", 6379);
        //主机设置值
        jedis_M.set("class", "1122V2");
        //从机获取值
        String result = jedis_S.get("class");//可能有延迟，需再次启动才能使用
        System.out.println(result);
    }
}

20.5 JedisPool

(1)JedisPoolUtil

1. 获取Jedis实例需要从JedisPool中获取
2. 用完Jedis实例需要返还给JedisPool
3. 如果Jedis在使用过程中出错，则也需要还给JedisPool

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class JedisPoolUtil {
    private static volatile JedisPool jedisPool = null;

    private JedisPoolUtil() {
    }

    public static JedisPool getJedisPoolInstance() {
        if (null == jedisPool) {
            synchronized (JedisPoolUtil.class) {
                if (null == jedisPool) {
                    JedisPoolConfig poolConfig = new JedisPoolConfig();
                    poolConfig.setMaxActive(1000);
                    poolConfig.setMaxIdle(32);
                    poolConfig.setMaxWait(100 * 1000);
                    poolConfig.setTestOnBorrow(true);

                    jedisPool = new JedisPool(poolConfig, "127.0.0.1", 6379);
                }
            }
        }
        return jedisPool;
    }

    public static void release(JedisPool jedisPool, Jedis jedis) {
        if (null != jedis) {
            jedisPool.returnResourceObject(jedis);
        }
    }

}

运行测试

TestPool.java

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

public class TestPool {

    public static void main(String[] args) {
        JedisPool jedisPool = JedisPoolUtil.getJedisPoolInstance();
        JedisPool jedisPool2 = JedisPoolUtil.getJedisPoolInstance();

        System.out.println(jedisPool == jedisPool2);

        Jedis jedis = null;
        try {
            jedis = jedisPool.getResource();
            jedis.set("aa", "bb");
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            JedisPoolUtil.release(jedisPool, jedis);
        }
    }
}

(2)配置总结

JedisPool的配置参数大部分是由JedisPoolConfig的对应项来赋值的。

• maxActive：控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；如果赋值为-1，则表示不限制；如果pool已经分配了maxActive个jedis实例，则此时pool的状态为exhausted。
• maxIdle：控制一个pool最多有多少个状态为idle(空闲)的jedis实例；
• whenExhaustedAction

：表示当pool中的jedis实例都被allocated完时，pool要采取的操作；默认有三种。

• maxWait：表示当borrow一个jedis实例时，最大的等待时间，如果超过等待时间，则直接抛JedisConnectionException；
• testOnBorrow：获得一个jedis实例的时候是否检查连接可用性（ping()）；如果为true，则得到的jedis实例均是可用的；
• testOnReturn：return 一个jedis实例给pool时，是否检查连接可用性（ping()）；
• testWhileIdle：如果为true，表示有一个idle object evitor线程对idle object进行扫描，如果validate失败，此object会被从pool中drop掉；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
• timeBetweenEvictionRunsMillis：表示idle object evitor两次扫描之间要sleep的毫秒数；
• numTestsPerEvictionRun：表示idle object evitor每次扫描的最多的对象数；minEvictableIdleTimeMillis
• minEvictableIdleTimeMillis：表示一个对象至少停留在idle状态的最短时间，然后才能被idle object evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义；
• softMinEvictableIdleTimeMillis：在minEvictableIdleTimeMillis基础上，加入了至少minIdle个对象已经在pool里面了。如果为-1，evicted不会根据idle time驱逐任何对象。如果minEvictableIdleTimeMillis>0，则此项设置无意义，且只有在timeBetweenEvictionRunsMillis大于0时才有意义；
• lifo：borrowObject返回对象时，是采用DEFAULT_LIFO（last in first out，即类似cache的最频繁使用队列），如果为False，则表示FIFO队列；

其中JedisPoolConfig对一些参数的默认设置如下：

• testWhileIdle=true
• minEvictableIdleTimeMills=60000
• timeBetweenEvictionRunsMillis=30000
• numTestsPerEvictionRun=-1

21 .Jedis - Spring Boot缓存注解

21.1 配置

1、在common模块添加依赖

由于redis缓存是公共应用，所以我们把依赖与配置添加到了common模块下面，在common模块pom.xml下添加以下依赖

<!-- redis -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<!-- spring2.X集成redis所需common-pool2-->
<dependency>
    <groupId>org.apache.commons</groupId>
    <artifactId>commons-pool2</artifactId>
    <version>2.6.0</version>
</dependency>

2、在service-base模块添加redis配置类

配置类：RedisConfig.java

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.cache.RedisCacheConfiguration;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.RedisSerializationContext;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import java.time.Duration;
@Configuration
@EnableCaching //开启缓存
public class RedisConfig {
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
        //key序列化方式
        template.setKeySerializer(redisSerializer);
        //value序列化
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //value hashmap序列化
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory) {
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        //解决查询缓存转换异常的问题
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        // 配置序列化（解决乱码的问题）,过期时间600秒
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
                .entryTtl(Duration.ofSeconds(600))
                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
                .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
                .cacheDefaults(config)
                .build();
        return cacheManager;
    }
    
}

21.2 在接口中添加redis缓存

由于首页数据变化不是很频繁，而且首页访问量相对较大，所以我们有必要把首页接口数据缓存到redis缓存中，减少数据库压力和提高访问速度。

改造service-cms模块首页banner接口，首页课程与讲师接口类似

(1) Spring Boot缓存注解

1、缓存@Cacheable

根据方法对其返回结果进行缓存，下次请求时，如果缓存存在，则直接读取缓存数据返回；如果缓存不存在，则执行方法，并把返回的结果存入缓存中。一般用在查询方法上。

查看源码，属性值如下：

2、缓存@CachePut

使用该注解标志的方法，每次都会执行，并将结果存入指定的缓存中。其他方法可以直接从响应的缓存中读取缓存数据，而不需要再去查询数据库。一般用在新增方法上。

查看源码，属性值如下：

3、缓存@CacheEvict

使用该注解标志的方法，会清空指定的缓存。一般用在更新或者删除方法上

查看源码，属性值如下：

(2)banner接口改造

1、在service-cms模块配置文件添加redis配置

spring.redis.host=192.168.44.132
spring.redis.port=6379
spring.redis.database= 0
spring.redis.timeout=1800000
spring.redis.lettuce.pool.max-active=20
spring.redis.lettuce.pool.max-wait=-1
#最大阻塞等待时间(负数表示没限制)
spring.redis.lettuce.pool.max-idle=5
spring.redis.lettuce.pool.min-idle=0

2、修改CrmBannerServiceImpl，添加redis缓存注解

@Service
public class CrmBannerServiceImpl extends ServiceImpl<CrmBannerMapper, CrmBanner> implements CrmBannerService {
    @Override
    /*根据方法对其返回结果进行缓存，下次请求时，如果缓存存在，则直接读取缓存数据返回；
       如果缓存不存在，则执行方法，并把返回的结果存入缓存中。一般用在查询方法上。
    //@Cacheable这个注解帮我封装好了会自动把数据存入redis：*/
    //banner表示存入redis的value(必填)，key表示存入的key会以::拼接到value中 里边要再加一层引号(选填)
    @Cacheable(value = "banner",key = "'selectAllBanner'")
    public List<CrmBanner> selectAllBanner() {
        QueryWrapper<CrmBanner> queryWrapper=new QueryWrapper<>();
        queryWrapper.orderByAsc("sort");
        //last方法 拼接sql语句
        queryWrapper.last("limit 8");
        List<CrmBanner> crmBanners = baseMapper.selectList(queryWrapper);
        return crmBanners;
    }
    @Override
    public void pageBanner(Page<CrmBanner> pageParam, Object o) {
        baseMapper.selectPage(pageParam,null);
    }
    @Override
    public CrmBanner getBannerById(String id) {
        return baseMapper.selectById(id);
    }
    @CacheEvict(value = "banner", allEntries=true)
    @Override
    public void saveBanner(CrmBanner banner) {
        baseMapper.insert(banner);
    }
    @CacheEvict(value = "banner", allEntries=true)
    @Override
    public void updateBannerById(CrmBanner banner) {
        baseMapper.updateById(banner);
    }
    @CacheEvict(value = "banner", allEntries=true)
    @Override
    public void removeBannerById(String id) {
        baseMapper.deleteById(id);
    }
}

3、在redis添加了key

4、通过源码查看到key生成的规则