Redis - NoSQL 和 Jedis 入门

RendaZhang

发布于 2020-10-26 15:22:29

1.4K0

发布于 2020-10-26 15:22:29

文章被收录于专栏：RendaRenda

概述

互联网架构的演变历程

第 1 阶段：

数据访问量不大，简单的架构即可搞定。
适合小型项目。

app -> dao -> mysql

第 2 阶段：

数据访问量大，使用缓存技术来缓解数据库的压力。
不同的业务访问不同的数据库。
适合中型项目。

app -> dao -> cache -> [mysql1, mysql2, mysql3]

第 3 阶段：

主从读写分离。
之前的缓存确实能够缓解数据库的压力，但是写和读都集中在一个数据库上，压力又了。
一个数据库负责写，一个数据库负责读；分工合作。
让 Master（主数据库）来响应事务性（增删改）操作，让 Slave（从数据库）来响应非事务性（查询）操作，然后再采用主从复制来把 Master 上的事务性操作同步到 Slave 数据库中。
MySQL 的 Master / Slave 就是网站的标配。
适合大型项目。

app -> dao -> cache -> 主库 -> [从库1, 从库2]

第 4 阶段：

在 MySQL 的主从复制，读写分离的基础上，MySQL 的主库开始出现瓶颈。
由于 MyISAM 使用表锁，所以并发性能特别差。
分库分表开始流行，MySQL 也提出了表分区，虽然不稳定，但有了希望。
使用 MySQL 集群。
适合超大型项目。

app -> dao -> cache -> {[主库 -> (从库1, 从库2)], [主库 -> (从库1, 从库2)]}

Redis 入门介绍

互联网需求的 3 高：高并发，高可扩，高性能。

Redis 是一种运行速度很快，并发性能很强，并且运行在内存上的 NoSQL（Not only SQL）数据库。

NoSQL 非关系型数据库和传统 RDBMS 关系型数据库相比的优势：

NoSQL 数据库无需事先为要存储的数据建立字段，随时可以存储自定义的数据格式。
而在关系数据库里，增删字段是一件非常麻烦的事情；如果是非常大数据量的表，增加字段简直就是一个噩梦。

RDBMS

高度组织化结构化数据
结构化查询语言 SQL
数据和关系都存储在单独的表中
数据操纵语言，数据定义语言
严格的一致性
基础事务

NoSQL

代表着不仅仅是 SQL
没有声明性查询语言
没有预定义的模式
键值对存储，列存储，文档存储，图形数据库
最终一致性，而非 ACID 属性
非结构化和不可预知的数据
CAP 定理
高性能，高可用性和可伸缩性

Redis 的常用使用场景：

缓存，是 Redis 当今最为人熟知的使用场景；在提升服务器性能方面非常有效；一些频繁被访问的数据，经常被访问的数据如果放在关系型数据库，每次查询的开销都会很大，而放在 Redis 中，因为 Redis 是放在内存中，可以很高效的访问。
排行榜，在使用传统的关系型数据库（MySQL、Oracle 等）来做这个事儿，非常的麻烦，而利用 Redis 的 SortSet（有序集合）数据结构能够简单的搞定。
计算器 / 限速器，利用 Redis 中原子性的自增操作，可以统计类似用户点赞数、用户访问数等，这类操作如果用 MySQL，频繁的读写会带来相当大的压力；限速器比较典型的使用场景是限制某个用户访问某个 API 的频率，常用的有抢购时防止用户疯狂点击带来不必要的压力。
好友关系，利用集合的一些命令，比如求交集、并集、差集等；可以方便搞定一些共同好友、共同爱好之类的功能。
简单消息队列，除了 Redis 自身的发布 / 订阅模式，也可以利用 List 来实现一个队列机制，比如：到货通知、邮件发送之类的需求，不需要高可靠，但是会带来非常大的 DB 压力，完全可以用 List 来完成异步解耦。
Session 共享，以 JSP 为例，默认 Session 是保存在服务器的文件中，如果是集群服务，同一个用户过来可能落在不同机器上，这就会导致用户频繁登陆；采用 Redis 保存 Session 后，无论用户落在那台机器上都能够获取到对应的 Session 信息。

Redis / Memcache / MongoDB 对比

Redis / Memcache / MongoDB 都是 NoSQL 数据库。

Redis 和 Memcache

Redis 和 Memcache 都是内存数据库；不过 Memcache 还可用于缓存其他东西，例如图片、视频等等。
Memcache 数据结构单一 key / value；Redis 更丰富一些，还提供 list，set， hash 等数据结构的存储，有效的减少网络 IO 的次数。
虚拟内存 – Redis 当物理内存用完时，可以将一些很久没用到的 value 交换到磁盘。
存储数据安全 – Memcache 挂掉后，数据没了（没有持久化机制）；Redis 可以定期保存到磁盘（持久化）。
灾难恢复 – Memcache 挂掉后，数据不可恢复；Redis 数据丢失后可以通过 RBD（将 Redis 在内存中的数据库记录定时 dump 到磁盘上进行持久化）或 AOF（将 Redis 的操作日志以追加的方式写入文件）恢复。

Redis 和 MongoDB

Redis 和 MongoDB 并不是竞争关系，更多的是一种协作共存的关系。
MongoDB 是一个基于分布式文件存储的数据库，是一个介于关系数据库和非关系数据库之间的产品，是非关系数据库当中功能最丰富，最像关系数据库的。
MongoDB 本质上还是硬盘数据库，在复杂查询时仍然会有大量的资源消耗，而且在处理复杂逻辑时仍然要不可避免地进行多次查询。
MongoDB 需要 Redis 或 Memcache 这样的内存数据库来作为中间层进行缓存和加速。
比如在某些复杂页面的场景中，整个页面的内容如果都从 MongoDB 中查询，可能要几十个查询语句，耗时很长；如果需求允许，则可以把整个页面的对象缓存至 Redis 中，定期更新；这样 MongoDB 和 Redis 就能很好地协作起来。

分布式数据库 CAP 原理

CAP 简介

传统的关系型数据库事务具备 ACID：

Atomicity 原子性
Consistency 一致性
Isolation 独立性
Durability 持久性

分布式数据库的 CAP：

Consistency - 强一致性

All nodes see the same data at the same time，更新操作成功并返回客户端后，所有节点在同一时间的数据完全一致，这就是分布式的一致性；一致性的问题在并发系统中不可避免，对于客户端来说，一致性指的是并发访问时更新过的数据如何获取的问题；从服务端来看，则是更新如何复制分布到整个系统，以保证数据最终一致。

Availability - 高可用性

可用性指 Reads and writes always succeed，即服务一直可用，而且要是正常的响应时间；好的可用性主要是指系统能够很好的为用户服务，不出现用户操作失败或者访问超时等用户体验不好的情况。

Partition Tolerance - 分区容错性

即分布式系统在遇到某节点或网络分区故障时，仍然能够对外提供满足一致性或可用性的服务；分区容错性要求能够让应用，虽然是一个分布式系统，但看上去却是一个可以运转正常的整体；比如现在的分布式系统中有某一个或者几个机器宕掉了，其他剩下的机器还能够正常运转满足系统需求，对于用户而言并没有什么体验上的影响。

CAP 理论

CAP 理论提出就是针对分布式数据库环境的，所以，P 这个属性必须容忍它的存在，而且是必须具备的。

因为 P 是必须的，那么需要选择的就是 A 和 C。

在分布式环境下，为了保证系统可用性，通常都采取了复制的方式，避免一个节点损坏，导致系统不可用。那么就出现了每个节点上的数据出现了很多个副本的情况，而数据从一个节点复制到另外的节点时需要时间和要求网络畅通的，所以，当 P 发生时，也就是无法向某个节点复制数据时，这时候你有两个选择：

选择可用性 A，此时，那个失去联系的节点依然可以向系统提供服务，不过它的数据就不能保证是同步的了（失去了 C 属性）。
选择一致性 C，为了保证数据库的一致性，必须等待失去联系的节点恢复过来，在这个过程中，那个节点是不允许对外提供服务的，这时候系统处于不可用状态（失去了 A 属性）。

最常见的例子是读写分离，某个节点负责写入数据，然后将数据同步到其它节点，其它节点提供读取的服务，当两个节点出现通信问题时，就面临着选择 A - 继续提供服务，但是数据不保证准确，C - 用户处于等待状态，一直等到数据同步完成。

CAP 总结

分区是常态，不可避免，三者不可共存。

可用性和一致性：

一致性高，可用性低
一致性低，可用性高

因此，根据 CAP 原理将 NoSQL 数据库分成了满足 CA 原则、满足 CP 原则和满足 AP 原则三大类：

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

下载与安装

下载

Redis：http://www.redis.net.cn

图形工具：https://redisdesktop.com/download

安装

虽然可以在安装在 windows 操作系统，但是官方不推荐，所以安装在 Linux 系统中。

1）上传 tar.gz 包到 /opt 目录下，并解压：

tar -zxvf redis-5.0.4.tar.gz

2）安装 gcc（必须有网络）：

yum -y install gcc

忘记是否安装过，可以使用 gcc -v 命令查看 gcc 版本，如果没有安装过，会提示命令不存在。

3）进入 Redis 目录，进行编译：

make

4）编译之后，开始安装：

make install

安装后的操作

后台运行方式

Redis 默认不会使用后台运行，如果需要，修改配置文件 daemonize=yes，当后台服务启动的时候，会写成一个进程文件运行。

打开配置文件：

vim /opt/redis-5.0.4/redis.conf

注释掉 bind，关闭保护模式，并修改为后台启动：

...
# ~~~ WARNING ~~~ If the computer running Redis is directly exposed to the
# internet, binding to all the interfaces is dangerous and will expose the
# instance to everybody on the internet. So by default we uncomment the
# following bind directive, that will force Redis to listen only into
# the IPv4 loopback interface address (this means Redis will be able to
# accept connections only from clients running into the same computer it
# is running).
#
# IF YOU ARE SURE YOU WANT YOUR INSTANCE TO LISTEN TO ALL THE INTERFACES
# JUST COMMENT THE FOLLOWING LINE.
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
# bind 127.0.0.1

...

# By default protected mode is enabled. You should disable it only if
# you are sure you want clients from other hosts to connect to Redis
# even if no authentication is configured, nor a specific set of interfaces
# are explicitly listed using the "bind" directive.
protected-mode no

...

# By default Redis does not run as a daemon. Use 'yes' if you need it.
# Note that Redis will write a pid file in /var/run/redis.pid when daemonized.
daemonize yes

...

以配置文件的方式启动：

cd /usr/local/bin
redis-server /opt/redis-5.0.4/redis.conf

29674:C 02 Oct 2020 02:56:47.338 # oO0OoO0OoO0Oo Redis is starting oO0OoO0OoO0Oo
29674:C 02 Oct 2020 02:56:47.338 # Redis version=5.0.4, bits=64, commit=00000000, modified=0, pid=29674, just started
29674:C 02 Oct 2020 02:56:47.338 # Configuration loaded

防火墙开放 Redis 的端口号：

firewall-cmd --zone=public --add-port=6379/tcp --permanent
firewall-cmd --reload

关闭数据库

单实例关闭

redis-cli shutdown

多实例关闭

redis-cli -p 6379 shutdown

常用操作

检测 6379 端口是否在监听

netstat -lntp | grep 6379

检测后台进程是否存在

ps -ef | grep redis

连接 Redis 并测试

redis-cli
ping

Redis 在 linux 支持命令补全（tab）

HelloWorld

# 保存数据
set k1 china   
# 获取数据
get kl

测试性能

先 ctrl + c，退出 Redis 客户端

redis-benchmark

执行命令后，命令不会自动停止，需要手动 ctrl+c 停止测试

====== PING_INLINE ======
  100000 requests completed in 2.52 seconds
  50 parallel clients
  3 bytes payload
  keep alive: 1

62.39% <= 1 milliseconds
99.95% <= 2 milliseconds
99.97% <= 3 milliseconds
100.00% <= 3 milliseconds
39682.54 requests per second

...

默认 16 个数据库

vim /opt/redis-5.0.4/redis.conf

...

# Set the number of databases. The default database is DB 0, you can select
# a different one on a per-connection basis using SELECT <dbid> where
# dbid is a number between 0 and 'databases'-1
databases 16

...

127.0.0.1:6379> get k1                 # 查询 k1
"china"
127.0.0.1:6379> select 16              # 切换 16 号数据库
(error) ERR DB index is out of range   # 数据库的下标超出了范围
127.0.0.1:6379> select 15              # 切换 15 号数据库
OK
127.0.0.1:6379[15]> get k1             # 查询 k1
(nil)
127.0.0.1:6379[15]> select 0           # 切换 0 号数据库
OK
127.0.0.1:6379> get k1                 # 查询 k1
"china"

数据库键的数量

dbsize

清空数据库

清空当前库

flushdb

清空所有（16个）库，慎用

flushall

模糊查询（keys）

模糊查询 keys 命令，有三个通配符：*，?，[]

**`*` - 通配任意多个字符**

查询所有的键：

keys *

模糊查询 k 开头，后面随便多少个字符：

keys k*

模糊查询 e 为最后一位，前面随便多少个字符：

keys *e

双 * 模式，匹配任意多个字符 - 查询包含 k 的键：

keys *k*

`?` - 通配单个字符

模糊查询 k 字头，并且匹配一个字符：

keys k?

只记得第一个字母是 k，长度是 3

keys k??

[] - 通配括号内的某一个字符

记得其他字母，第二个字母可能是 a 或 e

keys r[ae]dis

键

exists key - 判断某个 key 是否存在

127.0.0.1:6379[15]> EXISTS k1
(integer) 1
127.0.0.1:6379[15]> EXISTS y1
(integer) 0

move key db - 移动（剪切，粘贴）键到几号库

127.0.0.1:6379[15]> EXISTS x1   # 将 x1 移动到 8 号库
(integer) 1                     # 移动成功
127.0.0.1:6379[15]> MOVE x1 8   # 查看当前库中是否存在 x1
(integer) 1                     # 不存在（因为已经移走了）
127.0.0.1:6379[15]> EXISTS x1   # 切换 8 号库
(integer) 0
127.0.0.1:6379[15]> SELECT 8    # 查看当前库中的所有键
OK
127.0.0.1:6379[8]> KEYS *
1) "x1"

ttl key - 查看键还有多久过期（-1 永不过期，-2 已过期）；Time To Live

127.0.0.1:6379[8]> TTL x1
(integer) -1                # 永不过期

expire key 秒 - 为键设置过期时间（生命倒计时）

127.0.0.1:6379[8]> set k1 v1       # 保存 k1
OK
127.0.0.1:6379[8]> TTL k1          # 查看 k1 的过期时间
(integer) -1                       # 永不过期
127.0.0.1:6379[8]> EXPIRE k1 10    # 设置 k1 的过期时间为 10 秒
(integer) 1                        # 设置成功
127.0.0.1:6379[8]> get k1          # 获取 k1
"v1"
127.0.0.1:6379[8]> TTL k1          # 查看 k1 的过期时间
(integer) 1                        # 还有 1 秒过期
127.0.0.1:6379[8]> get k1
(nil)                              # 从内存中销毁了

type key - 查看键的数据类型

127.0.0.1:6379[8]> type k1
string                             # k1 的数据类型是 string 字符串

使用 Redis

五大数据类型

操作文档：http://redisdoc.com/

字符串 String

set / get / del / append / strlen

127.0.0.1:6379[8]> set k1 v1     # 保存数据
OK
127.0.0.1:6379[8]> set k2 v2     # 保存数据
OK
127.0.0.1:6379[8]> keys *
1) "k2"
2) "k1"
127.0.0.1:6379[8]> del k2        # 删除数据 k2
(integer) 1
127.0.0.1:6379[8]> keys *
1) "k1"
127.0.0.1:6379[8]> get k1        # 获取数据 k1
"v1"
127.0.0.1:6379[8]> append k1 abc # 往 k1 的值追加数据 abc
(integer) 5                      # 返回值的长度（字符数量）
127.0.0.1:6379[8]> get k1
"v1abc"
127.0.0.1:6379[8]> strlen k1     # 返回 k1 值的长度（字符数量）
(integer) 5

incr / decr / incrby / decrby

加减操作，操作的必须是数字类型

incr - increment decr - decrement

127.0.0.1:6379[8]> set k1 1       # 初始化 k1 的值为 1
OK
127.0.0.1:6379[8]> incr k1        # k1 自增 1（相当于 ++）
(integer) 2
127.0.0.1:6379[8]> incr k1
(integer) 3
127.0.0.1:6379[8]> get k1
"3"
127.0.0.1:6379[8]> decr k1        # k1 自减 1（相当于 --）
(integer) 2
127.0.0.1:6379[8]> decr k1
(integer) 1
127.0.0.1:6379[8]> get k1
"1"
127.0.0.1:6379[8]> INCRBY k1 3    # k1 自增 3（相当于 +=3）
(integer) 4
127.0.0.1:6379[8]> GET k1
"4"
127.0.0.1:6379[8]> DECRBY k1 2    # k1 自减 2（相当于 -=2）
(integer) 2
127.0.0.1:6379[8]> get k1
"2"

getrange / setrange

类似 between … and …

range：范围

# 初始化 k1 的值为 abcdef
127.0.0.1:6379[8]> SET k1 abcdef
OK
127.0.0.1:6379[8]> get k1
"abcdef"
# 查询 k1 全部的值
127.0.0.1:6379[8]> getrange k1 0 -1
"abcdef"
# 查询 k1 的值，范围是下标 0 ~ 下标 3（包含 0 和 3，共返回 4 个字符）
127.0.0.1:6379[8]> getrange k1 0 3
"abcd"
# 替换 k1 的值，从下标 1 开始提供为 xxx
127.0.0.1:6379[8]> setrange k1 1 xxx
(integer) 6
127.0.0.1:6379[8]> get k1
"axxxef"

setex / setnx

set with expir：添加数据的同时设置生命周期

# 添加 k1 v1 数据的同时，设置 5 秒的声明周期
127.0.0.1:6379[8]> SETEX k1 5 v1
OK
127.0.0.1:6379[8]> get k1
"v1"
# 已过期，k1 的值 v1 自动销毁
127.0.0.1:6379[8]> get k1
(nil)

set if not exist：添加数据的时候判断是否已经存在，防止已存在的数据被覆盖掉

# k1 不存在，添加成功
127.0.0.1:6379[8]> setnx k1 v1
(integer) 1
# 添加失败，因为 k1 已经存在
127.0.0.1:6379[8]> setnx k1 renda
(integer) 0

mset / mget / msetnx

m：更多

# set 不支持一次添加多条数据
127.0.0.1:6379[8]> set k1 v1 k2 v2
(error) ERR syntax error
# mset 可以一次添加多条数据
127.0.0.1:6379[8]> mset k1 v1 k2 v2 k3 v3
OK
127.0.0.1:6379[8]> keys *
1) "k2"
2) "k1"
3) "k3"
# 一次获取多条数据
127.0.0.1:6379[8]> mget k2 k3
1) "v2"
2) "v3"
# 一次添加多条数据时，如果添加的数据中有已经存在的，则失败
127.0.0.1:6379[8]> msetnx k3 v3 k4 v4
(integer) 0
# 一次添加多条数据时，如果添加的数据中都不存在的，则成功
127.0.0.1:6379[8]> msetnx k4 v4 k5 v5
(integer) 1

getset：先 get 后 set

# 因为没有 k6，所以 get 为 null，然后将 k6 的值 v6 添加到数据库
127.0.0.1:6379[8]> getset k6 v6
(nil)
127.0.0.1:6379[8]> keys *
1) "k6"
2) "k2"
3) "k1"
4) "k4"
5) "k5"
6) "k3"
127.0.0.1:6379[8]> get k6
"v6"
# 先获取 k6 的值，然后修改 k6 的值为 vv6
127.0.0.1:6379[8]> getset k6 vv6
"v6"
127.0.0.1:6379[8]> get k6
"vv6"

列表 List

lpush / rpush / lrange

l：left 自左向右添加（从上往下添加）

r：right 自右向左添加（从下往上添加）

# 从上往下添加
127.0.0.1:6379[1]> lpush list01 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> keys *
1) "list01"
# 查询 list01 中的全部数据 0 表示开始，-1 表示结尾
127.0.0.1:6379[1]> lrange list01 0 -1
1) "5"
2) "4"
3) "3"
4) "2"
5) "1"
# 从下往上添加
127.0.0.1:6379[1]> rpush list02 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> lrange list02 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"

lpop / rpop

移除第一个元素（上左下右）

# 从左（上）边移除第一个元素
127.0.0.1:6379[1]> LPOP list02
"1"
# 从右（下）边移除第一个元素
127.0.0.1:6379[1]> RPOP list02
"5"

lindex

根据下标查询元素（从左向右，自上而下）

127.0.0.1:6379[1]> lrange list01 0 -1
1) "5"
2) "4"
3) "3"
4) "2"
5) "1"
# 从上到下数，下标为 2 的值
127.0.0.1:6379[1]> lindex list01 2
"3"
# 从上到下数，下标为 1 的值
127.0.0.1:6379[1]> lindex list01 1
"4"

llen

127.0.0.1:6379[1]> llen list01
(integer) 5

lrem

删除 n 个 value

127.0.0.1:6379[1]> lpush list01 1 2 2 3 3 3 4 4 4 4
(integer) 10
# 从 list01 中移除 2 个 3
127.0.0.1:6379[1]> lrem list01 2 3
(integer) 2
127.0.0.1:6379[1]> lrange list01 0 -1
 1) "4"
 2) "4"
 3) "4"
 4) "4"
 5) "3"
 6) "2"
 7) "2"
 8) "1"

ltrim

截取指定范围的值，别的全扔掉

ltrim key begindex endindex

127.0.0.1:6379[1]> lpush list01 1 2 3 4 5 6 7 8 9
(integer) 9
127.0.0.1:6379[1]> lrange list01 0 -1
1) "9"
2) "8"
3) "7"
4) "6"
5) "5"
6) "4"
7) "3"
8) "2"
9) "1"
# 截取下标 3 ~ 6 的值，别的全扔掉
127.0.0.1:6379[1]> ltrim list01 3 6
OK
127.0.0.1:6379[1]> lrange list01 0 -1
1) "6"
2) "5"
3) "4"
4) "3"

rpoplpush

从一个集合搞一个元素到另一个集合中（右出一个，左进一个）

127.0.0.1:6379[1]> rpush list01 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> lrange list01 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
127.0.0.1:6379[1]> rpush list02 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> lrange list02 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
# list01 右边出一个，从左进入到 list02 的第一个位置
127.0.0.1:6379[1]> RPOPLPUSH list01 list02
"5"
127.0.0.1:6379[1]> LRANGE list01 0 -1
1) "1"
2) "2"
3) "3"
4) "4"
127.0.0.1:6379[1]> LRANGE list02 0 -1
1) "5"
2) "1"
3) "2"
4) "3"
5) "4"
6) "5"

lset

改变某个下标的某个值

lset key index value

127.0.0.1:6379[1]> lrange list02 0 -1
1) "5"
2) "1"
3) "2"
4) "3"
5) "4"
6) "5"
# 将 list02 中下标为 0 的元素修改成 x
127.0.0.1:6379[1]> lset list02 0 x
OK
127.0.0.1:6379[1]> lrange list02 0 -1
1) "x"
2) "1"
3) "2"
4) "3"
5) "4"
6) "5"

linsert

插入元素（指定某个元素之前 / 之后）

linsert key before/after oldvalue newvalue

127.0.0.1:6379[1]> lrange list02 0 -1
1) "x"
2) "1"
3) "2"
4) "3"
5) "4"
6) "5"
# 从左边进入，在 list02 中的 2 元素之前插入 java
127.0.0.1:6379[1]> linsert list02 before 2 java
(integer) 7
127.0.0.1:6379[1]> lrange list02 0 -1
1) "x"
2) "1"
3) "java"
4) "2"
5) "3"
6) "4"
7) "5"
 # 从左边进入，在 list02 中的 2 元素之后插入 redis
127.0.0.1:6379[1]> linsert list02 after 2 redis
(integer) 8
127.0.0.1:6379[1]> lrange list02 0 -1
1) "x"
2) "1"
3) "java"
4) "2"
5) "redis"
6) "3"
7) "4"
8) "5"

列表性能总结：类似添加火车皮一样，头尾操作效率高，中间操作效率惨

集合 Set

与 Java 中的 set 特点类似，不允许重复

sadd / smembers / sismember

添加 / 查看 / 判断是否存在

# 添加元素（自动排除重复元素）
127.0.0.1:6379[1]> sadd set01 1 2 2 3 3 3
(integer) 3
# 查询 set01 集合
127.0.0.1:6379[1]> smembers set01
1) "1"
2) "2"
3) "3"
# 存在
127.0.0.1:6379[1]> sismember set01 2
(integer) 1
# 不存在
127.0.0.1:6379[1]> sismember set01 5
(integer) 0

注意：1 和 0 不是下标，而是布尔值；1 - true 存在，2 - false 不存在

scard

获得集合中的元素个数

# 集合中有 3 个元素
127.0.0.1:6379[1]> scard set01
(integer) 3

srem

删除集合中的元素

srem key value

# 移除 set01 中的元素 2；1 表示移除成功
127.0.0.1:6379[1]> srem set01 2
(integer) 1
127.0.0.1:6379[1]> smembers set01
1) "1"
2) "3"

srandmember

从集合中随机获取几个元素

srandmember 整数（个数）

127.0.0.1:6379[1]> sadd set01 1 2 3 4 5 6 7 8 9
(integer) 9
127.0.0.1:6379[1]> smembers set01
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
6) "6"
7) "7"
8) "8"
9) "9"
# 从 set01 中随机获取 3 个元素
127.0.0.1:6379[1]> srandmember set01 3
1) "2"
2) "3"
3) "9"
# 从 set01 中随机获取 5 个元素
127.0.0.1:6379[1]> srandmember set01 5
1) "8"
2) "9"
3) "3"
4) "1"
5) "6"

spop

随机出栈（移除）

127.0.0.1:6379[1]> smembers set01
1) "1"
2) "2"
3) "3"
4) "4"
5) "5"
6) "6"
7) "7"
8) "8"
9) "9"
# 随机移除一个元素
127.0.0.1:6379[1]> spop set01
"8"
# 随机移除一个元素
127.0.0.1:6379[1]> spop set01
"1"

smove

移动元素 - 将 key1 某个值赋值给 key2

127.0.0.1:6379[1]> SADD set01 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> SADD set02 x y z
(integer) 3
# 将 set01 中的元素 3 移动到 set02 中
127.0.0.1:6379[1]> smove set01 set02 3
(integer) 1
127.0.0.1:6379[1]> smove set01 set02 5
(integer) 1
127.0.0.1:6379[1]> SMEMBERS set01
1) "1"
2) "2"
3) "4"

数学集合类

交集 - sinter

并集 - sunion

差集 - sdiff

127.0.0.1:6379[1]> SADD set01 1 2 3 4 5
(integer) 5
127.0.0.1:6379[1]> SADD set02 2 a 1 b 3
(integer) 5
127.0.0.1:6379[1]> SINTER set01 set02
1) "1"
2) "2"
3) "3"
127.0.0.1:6379[1]> SUNION set01 set02
1) "b"
2) "1"
3) "a"
4) "5"
5) "2"
6) "3"
7) "4"
# 在 set01 中存在，在 set02 中不存在
127.0.0.1:6379[1]> sdiff set01 set02
1) "4"
2) "5"
# 在 set02 中存在，在 set01 中不存在
127.0.0.1:6379[1]> sdiff set02 set01
1) "b"
2) "a"

哈希 hash

类似 java 里面的 Map<String, Object>

Key / Value（键值对）模式不变，但 Value（值）又是一个键值对

hset / hget / hmset / hmget / hgetall / hdel

添加 / 得到 / 多添加 / 多得到 / 得到全部 / 删除属性

# 添加 user，值为 id=1001
127.0.0.1:6379[1]> HSET user id 1001
(integer) 1
127.0.0.1:6379[1]> HGET user
(error) ERR wrong number of arguments for 'hget' command
# 查询 user，必须指明具体的字段
127.0.0.1:6379[1]> HGET user id
"1001"
# 添加学生 student，属性一堆
127.0.0.1:6379[1]> HMSET student id 101 name tom age 22
OK
# 获取学生名字
127.0.0.1:6379[1]> HGET student name
"tom"
# 获取学生名字和年龄
127.0.0.1:6379[1]> HMGET student name age
1) "tom"
2) "22"
# 获取学生全部信息
127.0.0.1:6379[1]> HGETALL student
1) "id"
2) "101"
3) "name"
4) "tom"
5) "age"
6) "22"
# 删除学生年龄属性
127.0.0.1:6379[1]> HDEL student age
(integer) 1
127.0.0.1:6379[1]> HGETALL student
1) "id"
2) "101"
3) "name"
4) "tom"

hlen

返回元素的属性个数

127.0.0.1:6379[1]> HGETALL student
1) "id"
2) "101"
3) "name"
4) "tom"
# student 属性的数量，id 和 name 共两个属性
127.0.0.1:6379[1]> HLEN student
(integer) 2

hexists

判断元素是否存在某个属性

# student 中是否存在 name 属性
127.0.0.1:6379[1]> HEXISTS student name
(integer) 1
# student 中是否存在 age 属性
127.0.0.1:6379[1]> HEXISTS student age
(integer) 0

hkeys / hvals

获得属性的所有 key / 获得属性的所有 value

# 获取 student 所有的属性名
127.0.0.1:6379[1]> HKEYS student
1) "id"
2) "name"
# 获取 student 所有属性的值（内容）
127.0.0.1:6379[1]> HVALS student
1) "101"
2) "tom"

hincrby / hincrbyfloat

自增（整数）/ 自增（小数）

127.0.0.1:6379[1]> hmset student id 101 name tom age 22
OK
# 自增整数 2
127.0.0.1:6379[1]> HINCRBY student age 2
(integer) 24
127.0.0.1:6379[1]> HGET student age
"24"
127.0.0.1:6379[1]> HMSET user id 1001 money 1000
OK
# 自增小数 5.5
127.0.0.1:6379[1]> HINCRBYFLOAT user money 5.5
"1005.5"
127.0.0.1:6379[1]> HGET user money
"1005.5"

hsetnx

添加的时候，先判断是否存在

# 添加失败，因为 age 已存在
127.0.0.1:6379[1]> HSETNX student age 18
(integer) 0
# 添加成功，因为 sex 不存在
127.0.0.1:6379[1]> HSETNX student sex male
(integer) 1
127.0.0.1:6379[1]> hgetall student
1) "id"
2) "101"
3) "name"
4) "tom"
5) "age"
6) "24"
7) "sex"
8) "male"

有序集合 Zset

需求：

充 10 元可享 vip1；

充 20 元可享 vip2；

充 30 元可享 vip3；

…

zadd / zrange （withscores）

添加 / 查询

127.0.0.1:6379> ZADD zset01 10 vip1 20 vip2 30 vip3 40 vip4 50 vip5
(integer) 5
# 查询数据
127.0.0.1:6379> zrange zset01 0 -1
1) "vip1"
2) "vip2"
3) "vip3"
4) "vip4"
5) "vip5"
# 带着分数查询数据
127.0.0.1:6379> zrange zset01 0 -1 withscores
 1) "vip1"
 2) "10"
 3) "vip2"
 4) "20"
 5) "vip3"
 6) "30"
 7) "vip4"
 8) "40"
 9) "vip5"
10) "50"

zrangebyscore

模糊查询

(：不包含

limit：跳过几个截取几个

# 20 <= score <= 40
127.0.0.1:6379> ZRANGEBYSCORE zset01 20 40
1) "vip2"
2) "vip3"
3) "vip4"
# 20 <= score < 40
127.0.0.1:6379> ZRANGEBYSCORE zset01 20 (40
1) "vip2"
2) "vip3"
# 20 < score < 40
127.0.0.1:6379> ZRANGEBYSCORE zset01 (20 (40
1) "vip3"
# 10 <= score <= 40，共返回四个，跳过前 2 个，取 2 个
127.0.0.1:6379> ZRANGEBYSCORE zset01 10 40 limit 2 2
1) "vip3"
2) "vip4"
# 20 <= score <= 40，共返回四个，跳过前 2 个，取 1 个
127.0.0.1:6379> ZRANGEBYSCORE zset01 10 40 limit 2 1
1) "vip3"

zrem

删除元素

# 移除 vip5
127.0.0.1:6379> ZREM zset01 vip5
(integer) 1
127.0.0.1:6379> zrange zset01 0 -1
1) "vip1"
2) "vip2"
3) "vip3"
4) "vip4"

zcard / zcount / zrank / zscore

集合长度 / 范围内元素个数 / 得元素下标 / 通过值得分数

# 集合中元素的个数
127.0.0.1:6379> ZCARD zset01
(integer) 4
# 分数在 20 ~ 30之间，共有几个元素
127.0.0.1:6379> ZCOUNT zset01 20 30
(integer) 2
# vip3 在集合中的下标（从上向下）
127.0.0.1:6379> ZRANK zset01 vip3
(integer) 2
# 通过元素获得对应的分数
127.0.0.1:6379> ZSCORE zset01 vip2 
"20"

zrevrank

逆序找下标（从下向上）

127.0.0.1:6379> ZRANGE zset01 0 -1
1) "vip1"
2) "vip2"
3) "vip3"
4) "vip4"
127.0.0.1:6379> ZREVRANK zset01 vip3
(integer) 1

zrevrange

逆序查询

# 顺序查询
127.0.0.1:6379> ZRANGE zset01 0 -1
1) "vip1"
2) "vip2"
3) "vip3"
4) "vip4"
# 逆序查询
127.0.0.1:6379> ZREVRANGE zset01 0 -1
1) "vip4"
2) "vip3"
3) "vip2"
4) "vip1"

zrevrangebyscore

逆序范围查找

# 逆序查询分数在 30 ~ 20 之间的（注意，先写大值，再写小值）
127.0.0.1:6379> ZREVRANGEBYSCORE zset01 30 20
1) "vip3"
2) "vip2"
# 如果小值在前，则结果为 null（empty list or set）
127.0.0.1:6379> ZREVRANGEBYSCORE zset01 20 30
(empty list or set)

Redis 持久化

RDB

Redis DataBase

在指定的时间间隔内，将内存中的数据集的快照写入磁盘；

默认保存在 /usr/local/bin 中，文件名 dump.rdb。

自动备份

Redis 是内存数据库，当每次用完 Redis，关闭 Linux 时，Redis 会自动将数据备份到一个文件中：/usr/local/bin/dump.rdb

1) 默认的自动备份策略不利于测试，所以修改 redis.conf 文件中的自动备份策略

vim /opt/redis-5.0.4/redis.conf

进入 vim 编辑器的底行模式，输入 /SNAPSHOTTING 并点击回车键进行搜索，修改 SNAPSHOTTING 下的备份策略：

save 900 1      # 900 秒内，至少变更 1 次，才会自动备份
save 120 10     # 120 秒内，至少变更 10 次，才会自动备份
save 60 10000   # 60 秒内，至少变更 10000 次，才会自动备份

如果只是用 Redis 的缓存功能，不需要持久化，那么可以注释掉所有的 save 行来停用保存功能，可以直接一个空字符串来实现停用：save ""

重启 Redis-Server

/usr/local/bin/redis-server /opt/redis-5.0.4/redis.conf

2) 使用 shutdown 关闭 Redis 来模拟关机；关机之前和关机之后，对比 dump.rdb 文件的更新时间

[root@localhost bin]# redis-cli 
127.0.0.1:6379> ping
PONG
127.0.0.1:6379> keys *
1) "zset01"
127.0.0.1:6379> set k2 v2
OK
127.0.0.1:6379> SHUTDOWN
not connected> exit

注意：当使用 shutdown 命令，Redis 会自动将数据库备份，所以 dump.rdb 文件创建时间更新了

3) 启动 Redis，要在 120 秒内改变 10 条数据，再查看 dump.rdb 文件的更新时间（开两个终端窗口，方便查看）

cd /usr/local/bin
ll | grep dump

4) 120 秒内改变 10 条数据这一动作触发了备份指令，目前，/usr/local/bin/dump.rdb 文件中保存了 10 条数据，将 dump.rdb 拷贝一份 dump10.rdb，此时两个文件中都保存 10 条数据

5) 数据已经备份了，这时清空全部数据库 flushall，再次 shutdown 模拟关机

6) 再次启动 Redis，发现数据消失了，dump.rdb 文件中的内容并没有恢复到 Redis 中

因为，当保存 10 条以上的数据时，数据备份起来了；

然后删除数据库，备份文件中的数据，也没问题；

但是，这个 shutdown 命令一旦执行，就会立刻备份，将删除之后的空数据库生成备份文件，将之前 10 条数据的备份文件覆盖掉了；所以，自动恢复失败。

为了解决这个问题，就要将备份文件再备份。

7) 将 dump.rdb 文件删除，将 dump10.rdb 重命名为 dump.rdb

8) 重新启动 redis 服务，登录 redis，数据 10 条全部恢复

手动备份

之前自动备份，必须更改好多数据，例如上边，改变了十多条数据，才会自动备份；

如果只保存一条数据想立刻备份，就需要每次操作完成，执行命令 save 就会立刻备份。

127.0.0.1:6379> set k1 x1
OK
127.0.0.1:6379> SAVE
OK

与 RDB 相关的配置

查看配置文件：

vim /opt/redis-5.0.4/redis.conf

快照 SNAPSHOTTING 下的配置解释如下

stop-writes-on-bgsave-error：开启备份后如果发生故障是否继续接受写请求。yes - 当后台备份时候反生错误，前台停止写入；no - 当后台备份时候反生错误，前台继续写入。
rdbcompression：对于存储到磁盘中的快照，是否启动 LZF 压缩算法，一般都会启动；因为这点性能，多买一台电脑，完全搞定 N 个来回了。yes - 启动；no - 不启动（不想消耗 CPU 资源，可关闭）。
rdbchecksum：在存储快照后，是否启动 CRC64 算法进行数据校验。开启后，大约增加 10% 左右的 CPU 消耗；如果希望获得最大的性能提升，可以选择关闭。
dbfilename：快照备份文件名字，默认为 dump.rdb。
dir：快照备份文件保存的目录，默认为当前目录。

优：适合大规模数据恢复，对数据完整性和一致性要求不高。

劣：一定间隔备份一次，意外 down 掉，就失去最后一次快照的所有修改。

AOF

Append Only File

以日志的形式记录每个写操作；

将 Redis 执行过的写指令全部记录下来（读操作不记录）；

只允许追加文件，不可以改写文件；

Redis 在启动之初会读取该文件从头到尾执行一遍，这样来重新构建数据。

开启 AOF

1）为了避免失误，最好将 redis.conf 总配置文件备份一下，然后再修改内容如下：

appendonly yes

# The name of the append only file (default: "appendonly.aof")

appendfilename "appendonly.aof"

2）重新启动 Redis，以新配置文件启动

redis-server /opt/redis-5.0.4/redis.conf

3）连接 Redis，加数据，删库，退出

4）查看当前文件夹 /usr/local/bin 多一个 AOF 文件，看看文件中的内容，保存的都是写操作

文件中的删库语句要删除，否则数据恢复不了。

编辑这个文件，如果不是 root 权限则要底行模式输入 wq! 强制执行

5）最后只需要重新连接，数据恢复成功

AOF 和 RDB 的优先级

查看 redis.conf 文件，AOF 和 RDB 两种备份策略可以同时开启，那系统会怎样选择？

1）编辑 appendonly.aof，修改为乱码，保存退出

2）启动 Redis 失败

所以是 AOF 优先载入来恢复原始数据；因为 AOF 比 RDB 数据保存的完整性更高。

3）修复 AOF 文件，删除不符合 Redis 语法规范的代码

reids-check-aof  --fix appendonly.aof

与 AOF 相关的配置

查看配置文件：

vim /opt/redis-5.0.4/redis.conf

文件追写模式 APPEND ONLY MODE 下的配置解释如下：

appendonly：开启 AOF 模式。
appendfilename：AOF 的文件名字，建议不修改。
appendfsync：追写策略。always - 每次数据变更，就会立即记录到磁盘，性能较差，但数据完整性好；everysec - 默认设置，异步操作，每秒记录，如果一秒内宕机，会有数据丢失；no - 不追写。
no-appendfsync-on-rewrite：当后台某个子线程在重写或保存时，主程序不执行追写策略，防止出现响应延迟问题。但是，为保证数据的安全性，默认 no 即可。AOF 采用文件追加的方式，文件会越来越大，为了解决这个问题，增加了重写机制，Redis 会自动记录上一次 AOF 文件的大小，当 AOF 文件大小达到预先设定的大小时，Redis 就会启动 AOF 文件进行内容压缩，只保留可以恢复数据的最小指令集合。
auto-aof-rewrite-percentage：一般设置为 AOF 文件大小已经超过原来的 100%，也就是一倍，才重写压缩。
auto-aof-rewrite-min-size：一般设置为 AOF 文件已经超过了 64 MB，才重写压缩。这样可以防止出现文件比较小的时候出现无必要的重写。

总结

RDB：只用作后备用途，建议 15 分钟备份一次就好。

AOF：

在最恶劣的情况下，也只丢失不超过 2 秒的数据，数据完整性比较高，但代价太大，会带来持续的 IO；
对硬盘的大小要求也高，默认 64 MB 太小了，企业级最少都是 5 G 以上；
Master / Slave（主/从）方案是更优的选择

NoSQL 事务

可以一次执行多个命令，是一个命令组，一个事务中，所有命令都会序列化（排队），不会被插队；一个队列中，一次性，顺序性，排他性的执行一系列命令。

三特性：

隔离性 - 所有命令都会按照顺序执行，事务在执行的过程中，不会被其他客户端的命令打断。
没有隔离级别 - 队列中的命令没有提交之前都不会被实际的执行，不存在类似“事务中查询要看到事务里的更新，事务外查询不能看到”的这些问题。
不保证原子性 - 如果一个命令失败，但是别的命令可能会执行成功，没有回滚。

三步走：

开启 multi
入队 queued
执行 exec

与关系型数据库事务相比：

multi - 可以理解成关系型事务中的 Begin
exec - 可以理解成关系型事务中的 Commit
discard - 可以理解成关系型事务中的 Rollback

开启事务

开启事务，加入队列，一起执行，并成功

# 开启事务
127.0.0.1:6379> multi
OK
# 加入队列
127.0.0.1:6379> set k1 v1
QUEUED
# 加入队列
127.0.0.1:6379> set k2 v2
QUEUED
# 加入队列
127.0.0.1:6379> get k2
QUEUED
# 加入队列
127.0.0.1:6379> set k3 v3
QUEUED
# 执行，一起成功
127.0.0.1:6379> exec
1) OK
2) OK
3) "v2"
4) OK

回滚事务

放弃之前的操作，恢复到原来的值

# 开启事务
127.0.0.1:6379> MULTI
OK
127.0.0.1:6379> SET k1 v1111
QUEUED
127.0.0.1:6379> SET k2 v2222
QUEUED
# 放弃操作
127.0.0.1:6379> DISCARD
OK
# 还是原来的值
127.0.0.1:6379> get k1
"v1"

报错恢复

一句报错，全部取消，恢复到原来的值

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set k4 v4
QUEUED
# 一句报错
127.0.0.1:6379> setvcdnkaj
(error) ERR unknown command `setvcdnkaj`, with args beginning with: 
127.0.0.1:6379> set k5 v5
QUEUED
# 队列中命令全部取消
127.0.0.1:6379> exec
(error) EXECABORT Transaction discarded because of previous errors.
# 还是原来的值
127.0.0.1:6379> keys *
1) "k1"
2) "k3"
3) "k2"

延迟报错

追究责任，谁报的错，找谁去

127.0.0.1:6379> MULTI
OK
# 虽然 v1 不能 ++，但是加入队列并没有报错，类似 java 中的通过编译
127.0.0.1:6379> INCR k1
QUEUED
127.0.0.1:6379> set k4 v4
QUEUED
127.0.0.1:6379> set k5 v5
QUEUED
# 真正执行的时候，报错
127.0.0.1:6379> EXEC
1) (error) ERR value is not an integer or out of range
2) OK
3) OK
127.0.0.1:6379> KEYS *
1) "k2"
2) "k1"
3) "k3"
4) "k4"
5) "k5"

Watch 监控

Watch 命令用于监视一个 (或多个) key ，如果在事务执行之前这个 (或这些) key 被其他命令所改动，那么事务将被打断。

模拟收入与支出。

正常情况下

# 收入 100 元
127.0.0.1:6379> set in 100 
OK
# 支出 0 元
127.0.0.1:6379> set out 0
OK
127.0.0.1:6379> multi
OK
# 收入 -20
127.0.0.1:6379> decrby in 20
QUEUED
# 支出 +20
127.0.0.1:6379> incrby out 20
QUEUED
# 结果，没问题
127.0.0.1:6379> exec
1) (integer) 80
2) (integer) 20

特殊情况下

在开启事务之前，先执行 watch 监控收入 in：

127.0.0.1:6379> watch in
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> decrby in 20
QUEUED
127.0.0.1:6379> incrby out 20
QUEUED

在 exec 之前，开启了另一个窗口（线程），对监控的 in 做了修改：

127.0.0.1:6379> set in 1000
OK
127.0.0.1:6379> get in
"1000"

第一个窗口的事务将被打断（失效），类似于“乐观锁”：

127.0.0.1:6379> exec
(nil)

unwatch：取消 watch 命令对所有 key 的操作。

一旦执行了 exec 命令，那么之前加的所有监控自动失效。

Redis 的发布订阅

进程间的一种消息通信模式：发送者 publish 发送消息，订阅者 subscribe 接收消息。

例如：微信订阅号。

订阅一个或多个频道。

一个窗口作为订阅者 Subscriber

127.0.0.1:6379> clear
# 1.订阅三个频道
127.0.0.1:6379> SUBSCRIBE cctv1 cctv2 cctv3
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "cctv1"
3) (integer) 1
1) "subscribe"
2) "cctv2"
3) (integer) 2
1) "subscribe"
2) "cctv3"
3) (integer) 3

另一个窗口作为发送者 Publisher

# 2.发送消息
127.0.0.1:6379> PUBLISH cctv3 NBA
(integer) 1
127.0.0.1:6379> PUBLISH cctv1 Renda
(integer) 1

第一个窗口的订阅者收到消息：

# 3.接收到推送过来的信息
1) "message"
2) "cctv3"
3) "NBA"
1) "message"
2) "cctv1"
3) "Renda"

主从复制

就是 Redis 集群的策略。

配从（库）不配主（库）：从库可以选择谁是主库，但主库不能选择从库。

读写分离：主机写，从机读。

一主二从

1）准备三台服务器（192.168.186.128，192.168.186.129，192.168.186.130），并修改 redis.conf 配置文件。

2）启动三台 Redis，并查看每台机器的角色，都是 Master。

[root@localhost redis-5.0.4]# /usr/local/bin/redis-cli 
127.0.0.1:6379> INFO replication
# Replication
role:master
connected_slaves:0
master_replid:68eda0fe7309a2dca0a1d4f7ac8ca53682c42142
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

3）测试开始

首先，将三个机器的 Redis 数据库全都清空，第一台添加值：

127.0.0.1:6379> MSET k1 v1 k2 v2
OK

其余两台机器，复制（找主库）：

127.0.0.1:6379> SLAVEOF 192.168.186.128 6379
OK

第一台再添加值：

127.0.0.1:6379> set k3 v3
OK

可以获得 slave 之前的 k1 和 k2，只要跟了主库，主库之前的数据也会立刻同步。

可以获得 slave 之后的 k3，只要跟了大哥，数据会立刻同步。

主机（128 master）可以添加成功，从机（129 和 130 是 slave）失败，从机只负责读取数据，无权写入数据，这就是“读写分离”。

主机显示两个从机：

127.0.0.1:6379> INFO replication
# Replication
role:master
connected_slaves:2
slave0:ip=192.168.186.130,port=6379,state=online,offset=70,lag=1
slave1:ip=192.168.186.129,port=6379,state=online,offset=70,lag=1
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:70
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:70

主机 128 SHUTDOWN，从机 129 和 130 仍然是 slave，并显示他们的 master 已离线：

127.0.0.1:6379> INFO replication
# Replication
role:slave
master_host:192.168.186.128
master_port:6379
master_link_status:down
master_last_io_seconds_ago:-1
master_sync_in_progress:0
slave_repl_offset:808
master_link_down_since_seconds:21
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:f1f8b79327c718104a335e6a259511266d687e7c
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:808
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:808

主机 128 重启，从机 129 和 130 仍然是 slave，并显示他们的 master 已上线：

127.0.0.1:6379> INFO replication
# Replication
role:slave
master_host:192.168.186.128
master_port:6379
master_link_status:up
master_last_io_seconds_ago:2
master_sync_in_progress:0
slave_repl_offset:0
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:0
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:0

关闭一个从机 129，主机 128 没有变化，只是显示少了一个 slave：

127.0.0.1:6379> INFO replication
# Replication
role:master
connected_slaves:1
slave0:ip=192.168.186.130,port=6379,state=online,offset=266,lag=1
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:266
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:266

而重启归来的从机自立门户成为了 master，不和原来的集群在一起了：

127.0.0.1:6379> INFO replication
# Replication
role:master
connected_slaves:0
master_replid:db4bc824a9b2da127ba6108ce8ff513179c1b609
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

主从继承

一个主机理论上可以多个从机，但是这样的话，这个主机会很累。

可以使用 java 面向对象继承中的传递性来解决这个问题，减轻主机的负担。

129 跟随 128：

127.0.0.1:6379> SLAVEOF 192.168.186.128 6379
OK

130 跟随 129：

127.0.0.1:6379> SLAVEOF 192.168.186.129 6379
OK

128：

127.0.0.1:6379> INFO replication
# Replication
role:master
connected_slaves:1
slave0:ip=192.168.186.129,port=6379,state=online,offset=1120,lag=0
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:1120
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:1120

129：

127.0.0.1:6379> INFO replication
# Replication
role:slave
master_host:192.168.186.128
master_port:6379
master_link_status:up
master_last_io_seconds_ago:7
master_sync_in_progress:0
slave_repl_offset:1106
slave_priority:100
slave_read_only:1
connected_slaves:1
slave0:ip=192.168.186.130,port=6379,state=online,offset=1106,lag=0
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:1106
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:953
repl_backlog_histlen:154

130：

127.0.0.1:6379> INFO replication
# Replication
role:slave
master_host:192.168.186.129
master_port:6379
master_link_status:up
master_last_io_seconds_ago:8
master_sync_in_progress:0
slave_repl_offset:1092
slave_priority:100
slave_read_only:1
connected_slaves:0
master_replid:dc6fc5fcaf449d33dad91ecabe3ba28d07dd1926
master_replid2:0000000000000000000000000000000000000000
master_repl_offset:1092
second_repl_offset:-1
repl_backlog_active:1
repl_backlog_size:1048576
repl_backlog_first_byte_offset:1
repl_backlog_histlen:1092

手选主机

1 个主机，2 个从机，当 1 个主机挂掉了，只能从 2 个从机中再次选 1 个主机。

手动选主机。

模拟测试：1 为 master，2 和 3 为 slave，当 1 挂掉后，2 选为 master，3 跟 2。

关闭主机 128，从机 129 执行命令成为主机：

slaveof no one

从机 130 执行命令跟随新主机 129：

SLAVEOF 192.168.186.129 6379

当 128 再次回归，128 和 129 已经形成新的集群，和 128 没有任何的关系了，所以 128 成为了光杆司令。

复制原理

1）Slave 服务器链接 Master 服务器，并发送同步请求。

2）Master 服务器启动后台的存盘进程，同时会收集所有写（Write）的命令集。

3）Master 服务器发送快照到 Slave 服务器。

4）Slave 服务器载入快照。

5）Master 服务器发送缓存写（Write）命令。

6）Slave 服务器执行命令。

完成上面几个步骤后就完成了 Slave 服务器数据初始化的所有操作，Slave 服务器此时可以接收来自用户的读请求。

全量复制：Slave 初始化阶段，这时 Slave 需要将 Master 上的所有数据都复制一份，Slave 接收到数据文件后，存盘，并加载到内存中（步骤 1、2、3、4）。
增量复制：Slave 初始化后，开始正常工作时主服务器发生的写操作同步到从服务器的过程（步骤 5、6）。

但只要是重新连接 Master，一次性（全量复制）同步将自动执行。

Redis 主从同步策略：主从刚刚连接的时候，进行全量同步；全量同步结束后，进行增量同步。

如果有需要，Slave 在任何时候都可以发起全量同步。

Redis 的策略是，无论如何，首先会尝试进行增量同步；如不成功，则要求 Slave 进行全量同步。

哨兵模式

哨兵 Sentinel 是 Redis 的高可用性解决方案：由一个或多个 Sentinel 实例组成的 Sentinel 系统可以监视任意多个主服务器，以及所有从服务器，并在被监视的主服务器进入下线状态时，自动将下线主服务器属下的某个从服务器升级为新的主服务器，然后由新的主服务器代替已下线的主服务器继续处理命令请求。

模拟测试：

1）128 为主服务器，129 和 130 为从服务器。

2）每一台服务器中在 /usr/local/bin 目录下创建一个配置文件 sentinel.conf，名字要正确，并编辑 sentinel.conf。

sentinel monitor 被监控主机名（自定义） ip port 票数

128 服务器 sentinel.conf：

sentinel monitor redis128 192.168.186.128 6379 1

129 服务器 sentinel.conf：

sentinel monitor redis129 192.168.186.129 6379 1

130 服务器 sentinel.conf：

sentinel monitor redis130 192.168.186.130 6379 1

3）启动服务的顺序：启动 128 服务器为 Master -> 启动 129 和 130 服务器为 Master -> 打开新的三个窗口分别启动 Redis 的哨兵 Sentinel 128、129、130。

/usr/local/bin/redis-sentinel sentinel.conf

4）将 128 挂掉，后台自动发起投票，选出新的主服务器。

127.0.0.1:6379> shutdown
not connected> exit

5）查看最后的分配：129 成为了新的主服务器，130 还是从服务器。

6）如果之前的主服务器再次归来：128 再次归来，自己成为了 master，和 129 平等；过了几秒之后，被哨兵检测到了 128 号机的归来，128 将变为 Slave。

42486:X 04 Oct 2020 10:56:08.098 # +sdown slave 192.168.186.128:6379 192.168.186.128 6379 @ redis128 192.168.186.129 6379
42486:X 04 Oct 2020 10:57:41.459 # -sdown slave 192.168.186.128:6379 192.168.186.128 6379 @ redis128 192.168.186.129 6379
42486:X 04 Oct 2020 10:57:51.434 * +convert-to-slave slave 192.168.186.128:6379 192.168.186.128 6379 @ redis128 192.168.186.129 6379

缺点

所有的写操作都是在 master 上完成的，然后再同步到 slave 上，所以两台机器之间通信会有延迟。
当系统很繁忙的时候，延迟问题会加重。
Slave 机器数量增加，问题也会加重。

总配置 redis.conf 详解

执行 vim /opt/redis-5.0.4/redis.conf 直接查看所有配置信息的详解

Jedis

Java 和 Redis 打交道的 API 客户端。

<dependencies>
    <dependency>
        <groupId>redis.clients</groupId>
        <artifactId>jedis</artifactId>
        <version>3.1.0</version>
    </dependency>
    <dependency>
        <groupId>commons-pool</groupId>
        <artifactId>commons-pool</artifactId>
        <version>1.6</version>
    </dependency>
</dependencies>

连接 Redis

public class Test1 {
    public static void main(String[] args) {
        Jedis jedis = new Jedis("192.168.186.128",6379);
        String pong = jedis.ping();
        System.out.println("pong = " + pong);
    }
}

常用 API

package com.renda;
import redis.clients.jedis.Jedis;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

public class testAPI {

    private void testString() {
        Jedis jedis = new Jedis("192.168.186.128", 6379);

        jedis.set("k1", "v1");
        jedis.set("k2", "v2");
        jedis.set("k3", "v3");

        Set<String> set = jedis.keys("*");
        Iterator<String> iterator = set.iterator();
        for (set.iterator(); iterator.hasNext(); ) {
            String k = iterator.next();
            System.out.println(k + " -> " + jedis.get(k));
        }

        // 查看 k2 是否存在
        Boolean k2Exists = jedis.exists("k2");
        System.out.println("k2Exists = " + k2Exists);
        // 查看 k1 的过期时间
        System.out.println(jedis.ttl("k1"));

        jedis.mset("k4", "v4", "k5", "v5");
        System.out.println(jedis.mget("k1", "k2", "k3", "k4", "k5"));
    }

    private void testList() {
        Jedis jedis = new Jedis("192.168.186.128", 6379);

        jedis.lpush("list01", "l1", "l2", "l3", "l4", "l5");
        List<String> list01 = jedis.lrange("list01", 0, -1);
        for (String s : list01) {
            System.out.println(s);
        }
    }

    private void testSet() {
        Jedis jedis = new Jedis("192.168.186.128", 6379);

        jedis.sadd("order", "jd001");
        jedis.sadd("order", "jd002");
        jedis.sadd("order", "jd003");
        Set<String> order = jedis.smembers("order");
        for (String s : order) {
            System.out.println(s);
        }

        jedis.srem("order", "jd002");

        System.out.println(jedis.smembers("order").size());
    }

    private void testHash() {
        Jedis jedis = new Jedis("192.168.186.128", 6379);

        jedis.hset("user1", "username", "renda");
        System.out.println(jedis.hget("user1", "username"));

        HashMap<String, String> map = new HashMap<String, String>();
        map.put("username", "Blair");
        map.put("gender", "female");
        map.put("address", "wuxi");
        map.put("phone", "1523641256");

        jedis.hmset("user2", map);

        List<String> list = jedis.hmget("user2", "username", "phone");
        for (String s : list) {
            System.out.println(s);
        }
    }

    private void testZset() {
        Jedis jedis = new Jedis("192.168.186.128", 6379);

        jedis.zadd("zset01", 60d, "zs1");
        jedis.zadd("zset01", 70d, "zs2");
        jedis.zadd("zset01", 80d, "zs3");
        jedis.zadd("zset01", 90d, "zs4");

        Set<String> zset01 = jedis.zrange("zset01", 0, -1);
        for (String s : zset01) {
            System.out.println(s);
        }
    }

    public static void main(String[] args) {
        testAPI testApi = new testAPI();
        test2Api.testString();
        test2Api.testList();
        test2Api.testSet();
        test2Api.testHash();
        test2Api.testZset();
    }
}

事务

初始化余额和支出

set balance 100
set expense 0

public class TestTransaction {

    public static void main(String[] args) throws InterruptedException {
        Jedis jedis = new Jedis("192.168.186.128",6379);

        int balance = Integer.parseInt(jedis.get("balance"));
        int expense = 10;

        // 监控余额
        jedis.watch("balance");
        // 模拟网络延迟
        Thread.sleep(10000);

        if (balance < expense) {
            // 解除监控
            jedis.unwatch();
            System.out.println("余额不足");
        } else {
            // 开启事务
            Transaction transaction = jedis.multi();
            // 余额减少
            transaction.decrBy("balance", expense);
            // 累计消费增加
            transaction.incrBy("expense", expense);
            // 执行事务
            transaction.exec();
            System.out.println("余额：" + jedis.get("balance"));
            System.out.println("累计支出：" + jedis.get("expense"));
        }
    }

}

模拟网络延迟：10 秒内，使用 linux 窗口修改 balance 为 5 模拟另一个线程的操作，此时因为 balance 被监控到改动，事务将被打断不会提交执行；输出的余额和累计支出将没有变化。

JedisPool

Redis 的连接池技术详情：https://help.aliyun.com/document_detail/98726.html

<dependency>
    <groupId>commons-pool</groupId>
    <artifactId>commons-pool</artifactId>
    <version>1.6</version>
</dependency>

使用单例模式进行优化：

public class JedisPoolUtil {

    private JedisPoolUtil () {
    }

    private volatile static JedisPool jedisPool = null;
    private volatile static Jedis jedis = null;

    /**
     * 返回一个连接池
     */
    private static JedisPool getInstance() {
        // 双层检测锁（企业中用的非常频繁）
        if (jedisPool == null) {
            synchronized (JedisPoolUtil.class) {
                if (jedisPool == null) {
                    JedisPoolConfig config = new JedisPoolConfig();
                    config.setMaxTotal(1000);
                    config.setMaxIdle(30);
                    config.setMaxWaitMillis(60*1000);
                    config.setTestOnBorrow(true);
                    jedisPool = new JedisPool(config, "192.168.186.128", 6379);
                }
            }
        }
        return jedisPool;
    }

    /**
     * 返回 jedis 对象
     */
    public static Jedis getJedis() {
        if (jedis == null) {
            jedis = getInstance().getResource();
        }
        return jedis;
    }

}

测试类：

public class TestJedisPool {

    public static void main(String[] args) {
        Jedis jedis1 = JedisPoolUtil.getJedis();
        Jedis jedis2 = JedisPoolUtil.getJedis();

        System.out.println(jedis1 == jedis2);
    }

}

高并发下的分布式锁

经典案例：秒杀，抢购优惠券等。

使用 Linux 窗口的 Redis Client 执行 set phone 10 设置测试案例的商品。

搭建工程并测试单线程

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.renda</groupId>
    <artifactId>high-concurrency-redis</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>war</packaging>

    <!-- 指定编码及版本 -->
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.encoding>UTF-8</maven.compiler.encoding>
        <java.version>1.11</java.version>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.springframework</groupId>
            <artifactId>spring-webmvc</artifactId>
            <version>5.2.7.RELEASE</version>
        </dependency>
        <!-- 实现分布式锁的工具类 -->
        <dependency>
            <groupId>org.redisson</groupId>
            <artifactId>redisson</artifactId>
            <version>3.6.1</version>
        </dependency>
        <!-- spring 操作 redis 的工具类 -->
        <dependency>
            <groupId>org.springframework.data</groupId>
            <artifactId>spring-data-redis</artifactId>
            <version>2.3.2.RELEASE</version>
        </dependency>
        <!-- redis 客户端 -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.1.0</version>
        </dependency>
        <!-- json 解析工具 -->
        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
            <version>2.9.8</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.tomcat.maven</groupId>
                <artifactId>tomcat7-maven-plugin</artifactId>
                <configuration>
                    <port>8001</port>
                    <path>/</path>
                </configuration>
                <executions>
                    <execution>
                        <!-- 打包完成后，运行服务 -->
                        <phase>package</phase>
                        <goals>
                            <goal>run</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

src\main\webapp\WEB-INF\web.xml

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xmlns="http://xmlns.jcp.org/xml/ns/javaee"
         xsi:schemaLocation="http://xmlns.jcp.org/xml/ns/javaee http://xmlns.jcp.org/xml/ns/javaee/web-app_3_1.xsd"
         id="WebApp_ID" version="3.1">

    <servlet>
        <servlet-name>springmvc</servlet-name>
        <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
        <init-param>
            <param-name>contextConfigLocation</param-name>
            <param-value>classpath:spring/spring.xml</param-value>
        </init-param>
    </servlet>
    <servlet-mapping>
        <servlet-name>springmvc</servlet-name>
        <url-pattern>/</url-pattern>
    </servlet-mapping>
</web-app>

src\main\resources\spring\spring.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="
       http://www.springframework.org/schema/beans
       http://www.springframework.org/schema/beans/spring-beans.xsd
       http://www.springframework.org/schema/context
       http://www.springframework.org/schema/context/spring-context.xsd">

    <context:component-scan base-package="com.renda.controller"/>

    <bean id="connectionFactory" class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory">
        <property name="hostName" value="192.168.186.128"/>
        <property name="port" value="6379"/>
    </bean>
    <!-- spring 为连接 redis，提供的一个模版工具类 -->
    <bean id="stringRedisTemplate" class="org.springframework.data.redis.core.StringRedisTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
    </bean>

</beans>

com.renda.controller.TestConcurrency

@Controller
public class TestConcurrency {
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    /**
     * 只能解决一个 tomcat 的并发问题：
     * synchronized 锁只解决了一个进程下的线程并发；
     * 如果分布式环境，多个进程并发，这种方案就失效了。
     */
    @RequestMapping("purchase")
    @ResponseBody
    public synchronized String purchase() {
        // 1.从 redis 中获取手机的库存数量
        int phoneCount = Integer.parseInt(stringRedisTemplate.opsForValue().get("phone"));
        // 2.判断手机的数量是否够秒杀
        if (phoneCount > 0) {
            phoneCount--;
            // 库存减少后，再将库存的值保存回 redis
            stringRedisTemplate.opsForValue().set("phone", phoneCount + "");
            System.out.println("库存减一，剩余：" + phoneCount);
        } else {
            System.out.println("库存不足");
        }
        return "over";
    }
}

高并发测试

1) 启动两次工程，端口号分别 8001 和 8002。

2) 使用 nginx 做负载均衡：

# 配置 Redis 多进程测试
upstream renda {
    server 192.168.1.116:8001;
    server 192.168.1.116:8002;
}

server {
    listen       80;
    server_name  www.redistest.com;
    location / {
        proxy_pass http://renda;
        index index.html index.htm;
    }
}

重新启动 Nginx：

/usr/local/nginx/sbin/nginx -s stop
/usr/local/nginx/sbin/nginx -c /usr/local/nginx/conf/nginx.conf

使用 SwitchHosts 编辑本地 host 地址：

# Redis
192.168.186.128 www.redistest.com

使用 Linux 窗口的 Redis Client 执行 set phone 20 设置测试案例的商品为 20 个。

3) 使用 JMeter 模拟 1 秒内发出 100 个 http 请求，会发现同一个商品会被两台服务器同时抢购。

实现 Redis 的分布式锁的思路

1）因为 redis 是单线程的，所以命令也就具备原子性，使用 setnx （判断如果不存在才执行 set）命令实现锁，保存 key / value。如果 key 不存在，则执行 set key value 给当前线程加锁，执行完成后，删除 key 表示释放锁；如果 key 已存在，阻塞线程执行，表示有锁。

2）如果加锁成功，在执行业务代码的过程中出现异常，导致没有删除 key（释放锁失败），那么就会造成死锁（后面的所有线程都无法执行）。为了解决这个问题，可以设置过期时间，例如 10 秒后，Redis 自动删除。

3）高并发下，由于时间段等因素导致服务器压力过大或过小，每个线程执行的时间不同：第一个线程，执行需要 13 秒，执行到第 10 秒时，redis 的 key 自动过期了（释放锁）；第二个线程，执行需要 7 秒，加锁，执行第 3 秒（锁被释放了，为什么，是因为被第一个线程的 finally 主动 deleteKey 释放掉了）。。。。连锁反应，当前线程刚加的锁，就被其他线程释放掉了，周而复始，导致锁会永久失效。

4）给每个线程加上唯一的标识 UUID 随机生成，释放的时候判断是否是当前的标识即可。

5）另外，还需要考虑过期时间如果设定。如果 10 秒太短不够用怎么办？设置 60 秒，太长又浪费时间。可以开启一个定时器线程，当过期时间小于总过期时间的 1/3 时，增长总过期时间。

Redisson

Redis 是最流行的 NoSQL 数据库解决方案之一，而 Java 是最流行的编程语言之一。

虽然两者看起来很自然地在一起，但是 Redis 其实并没有对 Java 提供原生支持。

相反，作为 Java 开发人员，想在程序中集成 Redis，必须使用 Redis 的第三方库。

而 Redisson 就是用于在 Java 程序中操作 Redis 的库，可以在程序中轻松地使用 Redis。

Redisson 在 java.util 中常用接口的基础上，提供了一系列具有分布式特性的工具类。

@Controller
public class TestConcurrency {

    @Autowired
    private Redisson redisson;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Bean
    public Redisson redisson() {
        Config config = new Config();
        // 使用单个 redis 服务器
        config.useSingleServer().setAddress("redis://192.168.186.128:6379").setDatabase(0);
        // 如果使用集群 redis：
        // config.useClusterServers().setScanInterval(2000).addNodeAddress("redis://192.168.186.128:6379","redis://192.168.186.129:6379","redis://192.168.186.130:6379");
        return (Redisson) Redisson.create(config);
    }

    @RequestMapping("purchase")
    @ResponseBody
    public synchronized String purchase() {
        // 定义商品 id，写死
        String productKey = "HUAWEI-P40";
        // 通过 redisson 获取锁（底层源码就是集成了 setnx，过期时间等操作）
        RLock rLock = redisson.getLock(productKey);
        // 上锁（过期时间为 30 秒）
        rLock.lock(30, TimeUnit.SECONDS);

        try {
            // 1.从 redis 中获取手机的库存数量
            int phoneCount = Integer.parseInt(stringRedisTemplate.opsForValue().get("phone"));
            // 2.判断手机的数量是否够秒杀
            if (phoneCount > 0) {
                phoneCount--;
                // 库存减少后，再将库存的值保存回 redis
                stringRedisTemplate.opsForValue().set("phone", phoneCount + "");
                System.out.println("库存减一，剩余：" + phoneCount);
            } else {
                System.out.println("库存不足");
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 释放锁
            rLock.unlock();
        }
        return "over";
    }

}

实现分布式锁的方案有很多，比如 ZooKeeper 的分布式锁特点就是高可靠性，Redis 的分布式锁的特点就是高性能。

目前分布式锁应用最多的仍然是 Redis。

本文参与腾讯云自媒体分享计划，分享自微信公众号。

原始发表：2020-10-11，如有侵权请联系 cloudcommunity@tencent.com 删除

云数据库 Redis