最具有代表性的 KV 数据库，常用于分布式缓存

一、NoSQL 概述

1、为什么要使用 NoSQL

1、单机 MySQL

瓶颈：
- 数据量太大，一个机器放不下
- 数据索引，机器内存不够
- 访问量（混合读写），一个机器承受不了

2、MemCached + MySQL+ 垂直拆分(读写分离)

网站大部分都是读操作，每次都查询数据库就很麻烦，使用缓存保持效率
发展过程：优化数据结构和索引 –> 文件缓存(IO) –> Memcached（当时最流行）

3、分库分表 + 水平拆分(MySQL 集群)

技术和业务发展的同时，对人的要求越来越高

本质：数据库（读&写）

MylSAM：表锁，十分影响效率，高并发下出现严重问题
InnoDB：行锁
慢慢的用分库分表减轻写的压力，但 mySQL 推出的表分区没有被大规模使用

4、为什么要 NoSQL

用户个人信息，社交网络，地理位置，用户日志
数据多种多样：定位，音乐，热榜，图片
NoSQL 可以很好的处理以上情况：
不需要固定格式去存储这些，不需要太多操作去横向扩展

2、什么是 NoSQL

NoSQL = Not Only SQL，泛指非关系型数据库
非关系型数据库：

NoSQL 特点：

解耦

方便扩展，数据之间没有关系，很好扩展
大数据量高性能（redis 一秒写 8w 次，读 11w 次，NoSQL 的缓存记录级，是一种细粒度的缓存，性能比较高）
数据类型多样，不需要实现设计数据库，随取随用
传统 RDBMS 和 NoSQL 区别

传统RDBMS
- 结构化组织
- SQL
- 数据和关系存储在单独的表中
- 操作，数据定义语言

NoSql
- 不仅仅是数据
- 没有固定的查询语言
- 简直对存储，列存储，文档存储，徒刑数据库
- 最终一致性
- CAP定力和BASE理论（异地多活），初级架构师
- 高性能，高可用，高可扩展性

了解：3V + 3 高
3V：海量 Volume，多样 Variety，实时 Velocity
3 高：高并发，高可扩，高性能

真正在公司中的实践：NoSQL 与 RDBMS 结合使用最强大

3、阿里巴巴演进分析

敏捷开发，极限编程

# 1、商品基本信息
名称、价格、商家信息
关系型数据库解决这些问题：Mysql/Oracle
# 2、商品的描述，评论（文字较多）
文档类型数据库，MangoDB
# 3、图片
分布式文件系统 FastDFS
- 淘宝自己的TFS
- Google的GFS
- Hadoop的HDFS
- 阿里云的OSS
# 4、商品的关键字(搜索)
- 搜索引擎 solr  elasticsearch
- 淘宝 ISearch：多隆
# 5、商品的热门波段信息
- 内存数据库
- Redis、Tair、Memcache
# 6、商品的交易，外部支付接口
- 三方应用

大型互联网公司应用问题
- 数据问题太多
- 数据源繁多，经常重构
- 数据改造，大面积改造
解决方案：统一的数据服务层 UDSL

4、NoSQL 四大分类

1、KV 键值对

2、文档类型数据库

MongoDB
- 一个基于分布式文件存储的数据库，C++编写，用来处理大量的文档
- 介于关系型和非关系型数据库之间，非关系型数据库功能最丰富，最像关系型数据库的

3、列存储数据库

HBase
分布式文件系统

4、图关系数据库

二、Redis 入门

Redis 是什么？

Redis (Remote Dictionary Server)，远程字典服务

Redis 是一个开源（BSD 许可）的，内存中的数据结构存储系统，它可以用作数据库、缓存和消息中间件。它支持多种类型的数据结构，如字符串（strings），散列（hashes），列表（lists），集合（sets），有序集合（sorted sets）与范围查询， bitmaps， hyperloglogs 和地理空间（geospatial）索引半径查询。 Redis 内置了复制（replication），LUA 脚本（Lua scripting）， LRU 驱动事件（LRU eviction），事务（transactions）和不同级别的磁盘持久化（persistence），并通过 Redis 哨兵（Sentinel）和自动分区（Cluster）提供高可用性（high availability）。

查看 Redis 命令大全 →

Redis 能做什么？

内存存储，持久化，内存断电即失，持久化很重要（rdb，aof）
效率高，可用于高速缓存
发布订阅系统
地图信息分布
计时器，计数器

Redis 特性

多样的数据类型
持久化
集群
事务

Redis 安装步骤

安装步骤

1、基本使用

redis-server # 启动服务
redis-benchmark # 测试速度
redis-cli -p portName # 指定端口启动客户端
ps -ef|grep redis #查看redis进程状态

# 之后在客户端内操作
auth xxx #输入设置的密码
ping # 测试连接，成功结果为PONG
shutdown # 退出连接
exit # 退出

set name lqs #设置值
get name # 获取值
keys * # 获取所有键

压力测试参数

2、基础知识

默认有 16 个数据库，默认使用第 0 个数据库

127.0.0.1:6379> select 2 #切换数据库
127.0.0.1:6379[2]> DBSIZE #查看数据库大小
(integer) 0
127.0.0.1:6379> flushall # 清空全部
127.0.0.1:6379> flushdb # 清空数据库

常见 Bug 及解决
1、MISCONF Redis is configured to save RDB snapshots, but it is currently not a
config stop-writes-on-bgsave-error no （临时解决）
最好的办法

三、五大数据类型

1、Redis-Key

exists name # 查看是否存在
move name 1 # 从数据库1移除name

127.0.0.1:6379> EXPIRE name 10 #设置过期时间，单位为s
(integer) 1
127.0.0.1:6379> keys *
1) "name"
127.0.0.1:6379> ttl name #查看剩余时间
(integer) 3
127.0.0.1:6379> ttl name
(integer) -2 #表示已过期
127.0.0.1:6379> get name
(nil)

127.0.0.1:6379> set age 1
OK
127.0.0.1:6379> type age #查看key类型
string

2、String

# 字符串
127.0.0.1:6379> set age 100
OK
127.0.0.1:6379> APPEND age sui #追加字符串，不存在则创建
(integer) 6
127.0.0.1:6379> STRLEN age
(integer) 6

# 自增自减
127.0.0.1:6379> set num 0
OK
127.0.0.1:6379> INCR num
#相当于num++
#如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作。
#如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误。
#本操作的值限制在 64 位(bit)有符号数字表示之内。
(integer) 1
127.0.0.1:6379> get num
"1"
127.0.0.1:6379> DECR num #相当于num--
(integer) 0
127.0.0.1:6379> INCRBY num 10 #设置步长，DECRBY同理
(integer) 10

# 字符串截取
127.0.0.1:6379> set name 12345678
OK
127.0.0.1:6379> GETRANGE name 1 4 #字符串截取
"2345"
127.0.0.1:6379> SETRANGE name 1 lqs #替换指定位置字符串
(integer) 8
127.0.0.1:6379> get name
"1lqs5678" #原来为12345678，234共3个长度替换为lqs

127.0.0.1:6379> SETEX k2 30 lqs #给k2设置为lqs，时间为30s

127.0.0.1:6379> SETNX mykey 100 #不存在就设置值，返回1
#在分布式锁中经常使用
(integer) 1
127.0.0.1:6379> get mykey
"100"
127.0.0.1:6379> SETNX mykey 101 #存在就不设置，返回0
(integer) 0
127.0.0.1:6379> get mykey
"100"

127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3 #批量创建，原子性操作
OK
127.0.0.1:6379> mget k1 k2 k3 #批量获取，原子性操作
1) "v1"
2) "v2"
3) "v3"

# 对象
#设置一个user：1对象值为json字符串来保存一个对象
127.0.0.1:6379> set user:1 {name:lqs,age:21}
OK
127.0.0.1:6379> get user:1 #获取user：1
"{name:lqs,age:21}"
127.0.0.1:6379> mset user:1 name:zhangsan,age=12 #修改对象
OK

127.0.0.1:6379> getset k3 v33 #先获取原来的值返回再设置新值，如果不存在返回nil
"v3"
127.0.0.1:6379> get k3
"v33"

3、List

所有的 list 命令以 l 开头

127.0.0.1:6379> lpush list one #向list插入值，单个或多个均可，插入到列表头部，left
(integer) 1
127.0.0.1:6379> lpush list two
(integer) 2
127.0.0.1:6379> lpush list three
(integer) 3
127.0.0.1:6379> lrange list 0 -1 #获取所哟元素
1) "three"
2) "two"
3) "one"
127.0.0.1:6379> lrange list 0 1 #从左边获取0到1两个位置的元素，不能从右边获取
1) "three"
2) "two"
127.0.0.1:6379> rpush list r1 r2 r3 r4 #向list插入值，插入到列表尾部。right
(integer) 10
127.0.0.1:6379> lpop list #从头部弹出
"3"
127.0.0.1:6379> rpop list #从尾部弹出
"r4"
127.0.0.1:6379> lindex list 3 #通过下标获取值，只能用l不能用r
"two"
127.0.0.1:6379> llen list # 获取长度
(integer) 8
127.0.0.1:6379> lrem list a b #从列表左边按顺序删除a个b元素
127.0.0.1:6379> ltrim l 2 6 #从l左边只保留2到6之内的元素
127.0.0.1:6379> rpoplpush l list # 将l弹出一个push到list中
127.0.0.1:6379> rset l 0 r #将l左边0号索引赋值为r

127.0.0.1:6379> rpush mylist hello
(integer) 1
127.0.0.1:6379> rpush mylist world
(integer) 2
#在mylist world后(before/after)插入other
127.0.0.1:6379> linsert mylist before world other
(integer) 3
127.0.0.1:6379> lrange mylist 0 -1
1) "hello"
2) "other"
3) "world"

实际为一个链表，每个 node 的 right/left 都要可以插入值
在左右两边插入或改动值效率最高，中间元素效率较低

4、Set

set 中的值不能重复

127.0.0.1:6379> sadd ms 1 #set添加元素
(integer) 1
127.0.0.1:6379> sadd ms 2
(integer) 1
127.0.0.1:6379> sadd ms 1
(integer) 0
127.0.0.1:6379> SMEMBERS ms # 查看ms中所有元素 / set中的值不能重复
1) "1"
2) "2"

127.0.0.1:6379> SISMEMBER ms 1 # 判断某元素是否存在，返回0/1
127.0.0.1:6379> SCARD ms #获取ms中元素的个数
127.0.0.1:6379> SREM ms 2 #移除元素
127.0.0.1:6379> SRANDMEMBER ms n #从ms中随机抽取n个元素，不写默认为1
127.0.0.1:6379> spop ms n #随机从ms中删除n个元素
127.0.0.1:6379> sdiff k1 k2 # k1和k2的不同的元素，可扩展至n个
127.0.0.1:6379> sinter k1 k2 #求k1和k2的交集
127.0.0.1:6379> sunion k1 k2 #k1和k2的并集集

5、Hash

Hash == KEY– <key,value>

127.0.0.1:6379> hset myhash field1 lqs #存hash
(integer) 1
127.0.0.1:6379> hmset myhash field1 lqs f2 l2 f3 l3 # 同时设置多个
OK
127.0.0.1:6379> hget myhash field1 #取hash值
"lqs"
127.0.0.1:6379> hmget myhash field1 f2 f3 #获取多个hash值
1) "lqs"
2) "l2"
3) "l3"
127.0.0.1:6379> hgetall myhash # 获取所有值
127.0.0.1:6379> hdel myhash f2 #删除hash中某个key
127.0.0.1:6379> hlen myhash #获取hash的长度
127.0.0.1:6379> hexists myhash field1 #判断hash中key是否存在
127.0.0.1:6379> hkeys myhash # 获取所有的key
127.0.0.1:6379> hvals myhash #获取所有的value
127.0.0.1:6379> hset myhash field 1 # 设置值
127.0.0.1:6379> HINCRBY myhash field1 1 # 自增
127.0.0.1:6379> HDECRBY myhash field1 1 # 自减
127.0.0.1:6379> HSETNX myhash field1 hello # 不存在则设置，存在就不设置
# hash的value可以存储对象，适合存储用户信息这些经常变动的

6、Zset

127.0.0.1:6379> zadd myset 1 one 2 two 3 three # 添加多个值
(integer) 3
127.0.0.1:6379> ZRANGE myset 0 -1
1) "one"
2) "two"
3) "three"

四、三种特殊数据类型

1、Geospatial 地理位置

可以实现两地位置信息，两地之间的距离
文档：http://www.redis.net.cn/order/3685.html

geoadd

添加地理坐标值
规则

- 有效的经度从-180 度到 180 度。
- 有效的纬度从-85.05112878 度到 85.05112878 度。
- 超过这个范围会报错
- 两级无法直接添加，一般直接下载城市数据，通过 Java 程序一次性导入
- 参数：key value(纬度 经度 城市名称)

例子

lqs@lqsdeiMac  ~  redis-cli  ✔  15:55:34
127.0.0.1:6379> geoadd china:city 116.40 39.90 beijin
(integer) 1
127.0.0.1:6379> geoadd china:city 127.47 31.23 shanghai
(integer) 1

geopos

获得当前位置坐标

127.0.0.1:6379> geopos china:city beijin

1. 1. "116.39999896287918091" 2) "39.90000009167092543"
      127.0.0.1:6379> geopos china:city shanghai
1. 1. "127.47000128030776978" 2) "31.22999903975783553"
      127.0.0.1:6379>

geodist

两地之间的距离

如果两个位置之间的其中一个不存在，那么命令返回空值
GEODIST 命令在计算距离时会假设地球为完美的球形，在极限情况下，这一假设最大会造成 0.5% 的误差。
单位
- m：米（默认值）
- km：千米
- mi：英里
- ft：英尺

127.0.0.1:6379> geodist china:city beijin shanghai km
"1388.2364"
127.0.0.1:6379> geodist china:city beijin shanghai
"1388236.3971"
redis> GEODIST Sicily Foo Bar
(nil)

georadius

以给定的经纬度为中心，找出某一半径内的元素

127.0.0.1:6379> GEOADD Sicily 13.361389 38.115556 "Palermo" 15.087269 37.502669 "Catania"
127.0.0.1:6379> GEORADIUS Sicily 15 37 200 km

1. "Palermo"
2. "Catania"

几个可选参数

WITHDIST: 在返回位置元素的同时，将位置元素与中心之间的距离也一并返回。距离的单位和用户给定的范围单位保持一致。
WITHCOORD: 将位置元素的经度和维度也一并返回。
WITHHASH: 以 52 位有符号整数的形式，返回位置元素经过原始 geohash 编码的有序集合分值。这个选项主要用于底层应用或者调试，实际中的作用并不大。

redis> GEORADIUS Sicily 15 37 200 km WITHDIST

1. 1. "Palermo"
   2. "190.4424"
2. 1. "Catania" 2) "56.4413"
      redis> GEORADIUS Sicily 15 37 200 km WITHCOORD
3. 1. "Palermo"
   2. 1. "13.361389338970184"
      2. "38.115556395496299"
4. 1. "Catania" 2) 1) "15.087267458438873" 2) "37.50266842333162"
      redis> GEORADIUS Sicily 15 37 200 km WITHDIST WITHCOORD
5. 1. "Palermo"
   2. "190.4424"
   3. 1. "13.361389338970184"
      2. "38.115556395496299"
6. 1. "Catania" 2) "56.4413" 3) 1) "15.087267458438873" 2) "37.50266842333162"
      redis>

排序

ASC: 根据中心的位置，按照从近到远的方式返回位置元素。
DESC: 根据中心的位置，按照从远到近的方式返回位置元素。

127.0.0.1:6379> GEORADIUS Sicily 15 37 200 km WITHDIST WITHCOORD DESC COUNT 1

1. 1. "Palermo" 2) "190.4424" 3) 1) "13.36138933897018433" 2) "38.11555639549629859"
      127.0.0.1:6379>

截取

COUNT：根据参数 n 返回 n 个数据
在默认情况下， GEORADIUS 命令会返回所有匹配的位置元素。虽然用户可以使用**COUNT ****<count>** 选项去获取前 N 个匹配元素，但是因为命令在内部可能会需要对所有被匹配的元素进行处理，所以在对一个非常大的区域进行搜索时，即使只使用COUNT 选项去获取少量元素，命令的执行速度也可能会非常慢。但是从另一方面来说，使用COUNT 选项去减少需要返回的元素数量，对于减少带宽来说仍然是非常有用的。

返回值

在没有给定任何WITH 选项的情况下，命令只会返回一个像 [“New York”,”Milan”,”Paris”] 这样的线性（linear）列表。
在指定了WITHCOORD 、WITHDIST 、WITHHASH 等选项的情况下，命令返回一个二层嵌套数组，内层的每个子数组就表示一个元素。

在返回嵌套数组时，子数组的第一个元素总是位置元素的名字。至于额外的信息，则会作为子数组的后续元素，按照以下顺序被返回：

以浮点数格式返回的中心与位置元素之间的距离，单位与用户指定范围时的单位一致。
geohash 整数。
由两个元素组成的坐标，分别为经度和纬度。

2、HyperLogLog

基数统计,有误差，可忽略
非常节省内存，2^64 个数据也只需要 12KB 内存
适合统计各种计数，比如注册 IP 数、每日访问 IP 数、页面实时UV、在线用户数等
只能统计数量，而没办法去知道具体的内容是什么

pfadd

新建元素

1 2	`127.0.0.1:6379> pfadd lqs 1 2 3 4 5 (integer) 1`

pfcount

统计数量

127.0.0.1:6379> pfadd lqs 1 2 3 4 5
(integer) 1
127.0.0.1:6379> pfadd lxl 2 4 5 6 5
(integer) 1
127.0.0.1:6379> pfcount lqs
(integer) 5
127.0.0.1:6379> pfcount lxl
(integer) 4
127.0.0.1:6379> pfmerge lqs lxl
OK
127.0.0.1:6379> pfcount lqs
(integer) 6
127.0.0.1:6379> pfcount lxl
(integer) 4
127.0.0.1:6

pfmerge

合并
合并后的数据

127.0.0.1:6379> pfmerge lqs lxl
OK
127.0.0.1:6379> pfcount lqs
(integer) 6
127.0.0.1:6379> pfcount lxl
(integer) 4
127.0.0.1:6

3、Bitmaps

位图：适合只有两个状态的需求

0 和 1 相对：活跃/不活跃，登陆/未登陆，打卡/未打卡

setbit

用 Bitmaps 记录一周的打卡

127.0.0.1:6379> setbit sign 0 1
(integer) 0
127.0.0.1:6379> setbit sign 2 0
(integer) 0
127.0.0.1:6379> setbit sign 1 1
(integer) 0
127.0.0.1:6379> setbit sign 3 0
(integer) 0
127.0.0.1:6379> setbit sign 4 1
(integer) 0
127.0.0.1:6379> setbit sign 5 0
(integer) 0
127.0.0.1:6379> setbit sign 6 0
(integer) 0
127.0.0.1:6379> setbit sign 7 0
(integer) 0
127.0.0.1:6379>

getbit

获取某一天的状态

1 2	`127.0.0.1:6379> getbit sign 1 (integer) 1`

bitcout

统计结果

1 2	`127.0.0.1:6379> bitcount sign (integer) 3`

事务

Redis 事务本质：一组命令的集合。一个事务中所有命令都会被序列化，执行过程中按照顺序执行

事务概念

Redis 没有隔离级别的概念
Redis 单条命令保证原子性，事务不保证原子性
Redis 事务过程
- 开启事务：multi
- 命令入队：……
  - 事务中所有事务没用被直接执行，发起执行命令时才会执行
- 执行事务：exec

# 正常开启事务

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k1 v1
QUEUED
127.0.0.1:6379> set k2 v1
QUEUED
127.0.0.1:6379> set k3 v2
QUEUED
127.0.0.1:6379> set k4 v4
QUEUED
127.0.0.1:6379> exec

1. OK
2. OK
3. OK
4. OK

# 不正常的结束事务

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set a 1
QUEUED
127.0.0.1:6379> set b 2
QUEUED

- 手动结束事务
  127.0.0.1:6379> discard
  OK
  127.0.0.1:6379> get b
  (nil)

异常

编译期异常

代码有问题，事务中所有命令都不执行

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set a 1
QUEUED
127.0.0.1:6379> setget b
(error) ERR unknown command `setget`, with args beginning with: `b`,
127.0.0.1:6379> set b 1
QUEUED
127.0.0.1:6379> exec
(error) EXECABORT Transaction discarded because of previous errors.

运行时异常

类似 1/0，如果语法没问题，其他命令可以执行

127.0.0.1:6379> set k1 "v1"
OK
127.0.0.1:6379> multi
OK

# 操作不对，但是语法没有错误

127.0.0.1:6379> incr k1
QUEUED
127.0.0.1:6379> set k2 v2
QUEUED
127.0.0.1:6379> set k3 v3
QUEUED
127.0.0.1:6379> get k3
QUEUED
127.0.0.1:6379> exec

1. (error) ERR value is not an integer or out of range

# 第一条不对，但其他的仍被执行

2. OK
3. OK
4. "v3"
   127.0.0.1:6379>

监控

悲观锁：认为做什么都会出问题，做什么都加锁
乐观锁：认为什么时候都不出问题，加一个 version！更新数据的时候判断一下

测试多线程监控

1、先单独执行左边事务，成功
2、中间线程，先开始监视，不执行事务
3、获取 money 并设置值，此时 exec 中间线程的事务，失败

测试多线程修改值，使用 watch 可以当作 Redis 的乐观锁操作

Jedis

Redis 官方推荐的的 Java 连接开发工具，使用 Java 操作 Redis 中间件

SpringBoot 整合 Redis

SpringData 是和 SpringBoot 齐名的项目
在 SpringBoot2.x 之后，SpringData 使用 Lettuce 替代率 Jedis
Jedis:直连技术，多线程操作不安全；想要避免不安全可以使用 pool 技术，更像 BIO
Lettuce:采用 netty，实例可以在多个线程在进行共享，多线程安全，更像 NIO

1、源码分析

//默认为true。false：不使用代理，每次获取新的，性能更高
@Configuration(proxyBeanMethods = false)
@ConditionalOnClass(RedisOperations.class)
@EnableConfigurationProperties(RedisProperties.class)
@Import({ LettuceConnectionConfiguration.class, JedisConnectionConfiguration.class })
public class RedisAutoConfiguration {

	@Bean
	//不存在redistemplate配置类时，默认配置类生效
	@ConditionalOnMissingBean(name = "redisTemplate")
	//默认两个Object，需要强制转换
	public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory)
			throws UnknownHostException {
		RedisTemplate<Object, Object> template = new RedisTemplate<>();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

	@Bean
	@ConditionalOnMissingBean
	//String常用，单独提出来一个bean
	public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory redisConnectionFactory)
			throws UnknownHostException {
		StringRedisTemplate template = new StringRedisTemplate();
		template.setConnectionFactory(redisConnectionFactory);
		return template;
	}

}

2、自定义 redisTemplate

public class RedisConfig {

    //自定义一个redisTemplate
    @Bean
    @SuppressWarnings("all")
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory){
        //一般直接使用<String, Object>
        RedisTemplate<String, Object> template = new RedisTemplate<>();
        template.setConnectionFactory(factory);
        return template;

        //json序列化设置
        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);

        //String序列化
        StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();

        //key采用String方式序列化
        template.setKeySerializer(stringRedisSerializer);
        //hash也采用String序列化
        template.setHashKeySerializer(stringRedisSerializer);
        //value序列化采用jsckson
        template.setValueSerializer(jackson2JsonRedisSerializer);
        //hash的value采用jsckson
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        template.afterPropertiesSet();

        return template;
    }
}

3、自定义 Redis 工具类

@Component
public final class RedisUtil {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    // =============================common============================
    /**
     * 指定缓存失效时间
     * @param key  键
     * @param time 时间(秒)
     */
    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(CollectionUtils.arrayToList(key));
            }
        }
    }


    // ============================String=============================

    /**
     * 普通缓存获取
     * @param key 键
     * @return 值
     */
    public Object get(String key) {
        return key == null ? null : redisTemplate.opsForValue().get(key);
    }

    /**
     * 普通缓存放入
     * @param key   键
     * @param value 值
     * @return true成功 false失败
     */
    public boolean set(String key, Object value) {
        try {
            redisTemplate.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, Object value, long time) {
        try {
            if (time > 0) {
                redisTemplate.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)
     */
    public long incr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递增因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, delta);
    }


    /**
     * 递减
     * @param key   键
     * @param delta 要减少几(小于0)
     */
    public long decr(String key, long delta) {
        if (delta < 0) {
            throw new RuntimeException("递减因子必须大于0");
        }
        return redisTemplate.opsForValue().increment(key, -delta);
    }


    // ================================Map=================================

    /**
     * HashGet
     * @param key  键 不能为null
     * @param item 项 不能为null
     */
    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 对应多个键值
     */
    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)
     */
    public double hincr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, by);
    }


    /**
     * hash递减
     *
     * @param key  键
     * @param item 项
     * @param by   要减少记(小于0)
     */
    public double hdecr(String key, String item, double by) {
        return redisTemplate.opsForHash().increment(key, item, -by);
    }


    // ============================set=============================

    /**
     * 根据key获取Set中的所有值
     * @param key 键
     */
    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 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 0;
        }
    }


    /**
     * 获取set缓存的长度
     *
     * @param key 键
     */
    public long sGetSetSize(String key) {
        try {
            return redisTemplate.opsForSet().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 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 0;
        }
    }

    // ===============================list=================================

    /**
     * 获取list缓存的内容
     *
     * @param key   键
     * @param start 开始
     * @param end   结束 0 到 -1代表所有值
     */
    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 键
     */
    public long lGetListSize(String key) {
        try {
            return redisTemplate.opsForList().size(key);
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }
    }


    /**
     * 通过索引 获取list中的值
     *
     * @param key   键
     * @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
     */
    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 值
     */
    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  时间(秒)
     */
    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 {
            Long remove = redisTemplate.opsForList().remove(key, count, value);
            return remove;
        } catch (Exception e) {
            e.printStackTrace();
            return 0;
        }

    }

}

Redis.conf 详解

通过配置文件启动
Mac 使用 HomeBrew 安装的 Redis 配置文件位置为:/usr/local/etc/redis.conf
配置文件对大小写不敏感

1、网络

# 绑定的 IP

bind 127.0.0.1 ::1

# 保护模式

protected-mode yes

# 端口

port 6379

2、通用

# 以守护进程运行，默认为 no，需要手动开启为 yes

daemonize yes

# 如果后台方式运行，需要指定 pid 文件

pidfile /var/run/redis_6379.pid

# 日志信息

# Specify the server verbosity level.

# This can be one of:

# debug (a lot of information, useful for development/testing)

# verbose (many rarely useful info, but not a mess like the debug level)

# notice (moderately verbose, what you want in production probably)

# warning (only very important / critical messages are logged)

loglevel notice

# 日志文件位置名

logfile ""

# 数据库数量,默认 16 个

database 16

# 是否显示 logo

always-show-logo yes

3、快照

持久化，在规定的时间内，执行了多少操作，会持久化到.rdb.aof 文件中

# 900s 内，如果至少有一个 key 进行修改，进行持久化操作

save 900 1

# 300s 内，至少 10 个 key 修改，进行持久化操作

save 300 10

# 60s 内 1000 个 key 修改，进行持久化操作

save 60 10000

# 持久化出错，是否还进行工作

stop-writes-on-bgsave-error yes

# 是否压缩 rdb 文件，需要消耗一定 cpu 资源

rdbcompression yes

# 保存 rdb 文件的时候，进行错误的检查校验

rdbchecksum yes

# 文件保存目录

dir ./

4、主从复制

5、安全

默认没用密码

127.0.0.1:6379> ping
PONG
127.0.0.1:6379> config get requirepass

1. "requirepass"
2. ""

可以加一个密码
设置完之后重启 redis

127.0.0.1:6379> config set requirepass "123456"
OK
127.0.0.1:6379> config get requirepass

1. "requirepass"
2. "123456"

# 重启之后登陆

127.0.0.1:6379> auth 123456

6、客户端

# 客户端最多数量

maxclients 10000

# 最大内存容量（单位：字节）

maxmemory <bytes>

# 内存满了处理策略策略

maxmemory-policy noeviction

- volatile-lru -> 根据 LRU 算法生成的过期时间来删除。
- allkeys-lru -> 根据 LRU 算法删除任何 key。
- volatile-random -> 根据过期设置来随机删除 key。
- allkeys->random -> 无差别随机删。
- volatile-ttl -> 根据最近过期时间来删除（辅以 TTL）
- noeviction -> 谁也不删，直接在写操作时返回错误。

7、APPEND ONLY 模式

# 默认不开启

appendonly no

# 持久化文件的名字(default: "appendonly.aof")

appendfilename "appendonly.aof"

# always：每次修改都执行 sync，消耗性能

# everysec：默认值，每秒执行一次 sync，可能丢失这一秒的数据

# no：不执行 sync，操作系统自己同步数据，速度最快

appendfsync everysec/always/no

Redis 持久化

Redis 是内存数据库，关机数据丢失，所以需要持久化操作

1、RDB

Redis DataBase
默认的持久化规则
在指定时间间隔内将内存中的数据集快照写入磁盘，即 SnapShot 快照，恢复时将快照文件存进内存

介绍

会单独创建一个子进程进行持久化，先将数据写入到一个临时文件中。
待持久化结束之后，在用临时文件替换上次持久化的文件
整个过程中，主进程不进行任何 IO 操作，确保极高性能

优点：性能比 AOF 更高效，需要大规模数据且数据恢复完整性不太敏感可以使用 RDB
缺点：最后一次持久化的数据可能丢失

触发规则

save 规则满足的情况
执行 flushall 命令
退出 Redis

恢复文件

文件名：dump.rdb
将 rdb 文件放入 redis 启动目录即可

# 查看目录位置

127.0.0.1:6379> config get dir

1. "dir"
2. "/usr/local/var/db/redis"
   127.0.0.1:6379>

# 生产环境可能需要备份 dump.rdb

2、AOF

Append Only File
以日志的形式记录每个写操作，将 Redis 执行过的所有指令记录下来

只许追加文件但不可以改写文件，Redis 启动之后会读取该文件重新构建数据
如果 aof 文件有错误，Redis 会启动失败，利用自动提供的工具redis-check-aof进行修复

优点
- 每次修改都同步，文件完整性更好
- 每次同步一次，可能会丢失一秒的数据
- 从不同步，效率最高
缺点
- 相对于数据文件来说，aof 远大于 rdb，修复文件慢
- aof 运行效率要比 rdb 慢，默认使用 rdb

扩展

只做缓存，就不需要持久化
同时开启两种持久化

Redis 重启的时候首先载入 AOF 文件恢复原始数据，通常情况下 AOF 比 RDB 数据更完整
RDB 数据不实时，同时使用两者也只会找 AOF 文件。建议不使用 AOF，只用 RDB 备份数据库，方便快速重启，不会有 AOF 潜在的 bug

性能建议

RDB 文件只用作后背用途，建议只在 Slave 上持久化 RDB 文件，15 分钟备份一次即可，即 save 900 1
使用 AOF 的好处：情况最恶劣也只丢失不超过 2s 的数据，编写脚本 load AOF 文件即可。代价：带来持续 IO。AOF rewrite 过程中产生的新数据到文件的阻塞基本不可避免；磁盘许可时，应该尽量见效 AOF rewrite 的频率，将默认值从 64M 改到 5G 以上
不使用 AOF，仅靠 Master-Slave Replcation 也可实现高可用，节省 IO，减少 rewrite 带来的系统波动。代价：如果 master/Slave 通水倒掉，会丢失十几分钟的数据，启动脚本也要比较 master/Slave 的 RDB 文件，载入较新的

Redis 发布订阅

Redis 发布订阅（pub/sub）：一种消息通信模式，发送者发送信息，订阅者接收消息。微信/微博关注系统

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

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

命令

基本命令

两个订阅者都订阅一个发布者

发布者发布再发布消息
订阅端会接收到消息

原理

使用场景

实时消息系统
群聊天室

更复杂的会使用 MQ

Redis 主从复制

将一台 Redis 服务器的数据复制到其他 redis 服务器。
前者为主节点（Master），后者为从节点（Slave）。
数据复制为单向的，只能从主节点到从节点。
Master 以写为主，Slave 以读为主。

作用

数据冗余：实现数据热备份，是持久化之外的数据冗余方式
服务冗余：当主节点出问题时，可由从节点提供服务，实现快速故障恢复。
负载均衡：在主从复制的基础上，配合读写分离，可以由主节点提供写服务，从节点提供读服务，分担负载。在写少读多读情况下可大大提高 redis 服务器读并发量
高可用基石：哨兵和集群实施的基础

工程中，只用一台服务器是不行的，原因：

结构上，单个 redis 会发生单点故障，一台服务器需要处理所有的请求负载，压力太大
容量上，单台 Redis 的最大使用内存不应该超过 20g
很多数据，如电商网站的商品数据，一般都是写少读多，适合使用主从复制

环境配置

# 查看当前库的信息

127.0.0.1:6379> info replication

# Replication

role:master
connected_slaves:0
master_replid:a30fd1701c63d8b83fd8fe449d30c92c63be3156
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:0
repl_backlog_size:1048576
repl_backlog_first_byte_offset:0
repl_backlog_histlen:0
127.0.0.1:6379>

注意：使用 Docker 可以简单的部署集群，此处只用实现伪集群，了解即可

原理：将 redis.conf 复制为多份，启动时使用不同配置文件启动即可

复制多份配置文件

修改复制配置文件的几项关键内容

端口
pid 名字
log 文件名字
dump.rdb 名字

以不同配置文件分别启动 redis-server

配置一主二从

# 6377 和 6378 认 6379 为主

127.0.0.1:6378> SLAVEOF 127.0.0.1 6379
127.0.0.1:6377> SLAVEOF 127.0.0.1 6379

# 使用 info replication 可以看到主从关系

细节

确定主从之后，主机可以写，从机不可再进行写操作
主机断开连接之后，从机连接到主机的，但是没有写操作，这个时候如果重启了，就会变回主机。只要变回从机，立马就会从从机中获取值

主从复制原理

总结的不太好，不过发现一篇不错的博客，redis 主从复制原理,有时间了好好总结一下。

Slave 启动成功连接到 Matser 后发送一个 sync 同步命令
Master 接到命令，启动后台的存盘进程，同时手机所有接收到的用于修改数据集命令。在后台执行完毕之后，Master 将传送整个数据文件到 Slave，并完成一次完全同步

全量复制：
增量复制：

………………

哨兵模式

自动选举主机的模式

主服务器宕机之后，手动将其中一台服务器升级为主服务器，需要人工干预，费时费力，导致一段时间内的服务不可用，一般不推荐使用。现在一般使用哨兵模式。Redis 自 2.8 版本开始正式提供了 Sentinel 架构来解决此问题

优点：

哨兵集群，基于主从复制模式，所有主从配置的优点，它都有
主从可以切换，故障可以转移，系统可用性会更好
哨兵模式就是主从模式的升级，手动–>自动,健壮性更好

缺点：

Redis 不方便在线扩容，集群容量一旦到达上限，在线扩容就比较麻烦

哨兵模式的全部配置

Redis 缓存穿透和雪崩（面试&常用）

关系到服务的高可用

缓存穿透

用户查询一个数据，Redis 中没有查询到，缓存未命中，于是向持久层数据库进行查询，也没有查询到结果，于是本次查询失败。当很多种这个情况发生时，会给持久层数据库带来很大压力，即缓存穿透

解决方案

布隆过滤器

一种数据结构，对所有可能的查询的参数以 hash 形式存储，在控制层先进性校验，不符合则丢弃，避免了对底层存储系统的查询压力

缓存空对象

当存储层不命中后，即使返回的空对象也将其缓存起来，同时设置一个过期时间。之后再访问这个数据将会从缓存中获取，保护后端数据源

带来两个问题

如果空值被缓存起来，意味着缓存需要更多的空间存储更多的键，因为其中可能有很多空值的键
基石对空值设置了过期时间，还是会存在缓存层和存储层会有一段时间窗口的不一致，这对于需要一致性的业务需求有一定影响

缓存击穿

指一个 key 非常热点，不停扛着大并发，集中对一个点进行访问，当着个 key 失效多瞬间，持续多大并发就会击破缓存，直接请求数据库

解决方案

设置热点数据永不过期

从缓存层面看，没有设置过期时间，所以不会出现热点 key 过气候产生的问题

加互斥锁

分布式锁：使用分布式锁，保证每个 key 同时只有一个线程去查询后端服务，其他线程没有获得分布式锁多权限，只需要等待即可。这种方式将高并发多压力转移到了分布式锁，因此对分布式锁的考验很大

缓存雪崩

在某一个时间段，缓存集中过期失效，Redis 宕机

eg：双十二之前一个小时设置了时间为一小时的一批缓存，当双十二零点时，缓存集体失效，此时有大量访问，查询全部落到数据库。数据库产生周期性多压力波峰，所有请求同时到持久层，导致持久层挂掉。

解决方案

Redis 高可用

增加几台 Redis 服务器，一台挂掉其他的还能继续工作。异地多活

限流降级

缓存失效后，通过加锁或队列来控制读数据库写缓存多线程数量。比如对某个 key 只允许一个线程查询数据和写缓存，其他线程等待

数据预热

正式部署前，把可能多数据预先访问一遍，这样部分可能大量访问多数据就会加载到缓存中。载即将发生大并发访问前手动触发加载缓存不同的 key，设置不同多过期时间，让缓存失效的时间尽量均匀

小结

资料获取：公众号回复 Redis 即可

遇见狂神说

NoSQL

#Redis

狂神Redis教程

https://polarisink.github.io/20220813/yuque/狂神Redis教程/

作者

Areis

发布于

2022年8月13日

许可协议

Docker-compose搭建Typecho博客上一篇