一、了解NoSql

1.1 什么是Nosql

1.2 为什么要使用NoSql

1.3 NoSql数据库的优势

1.4 常见的NoSql产品

1.5 各产品的区别

二、Redis介绍

2.1什么是Redis

2.2 Redis优势

2.3 Redis应用场景

2.4 Redis下载

三、Linux下安装Redis

3.1 环境准备

3.2 Redis的安装

3.2.1 Redis的编译环境

3.2.2 Redis的安装

3.3 Redis的启动

3.3.1 Redis的前端模式启动（了解）

3.3.2 Redis的后端启动

3.3.3 客户端访问redis

3.3.4 向Redis服务器发送命令

3.3.5 退出客户端

3.3.6 Redis的停止

3.3.7 第三方工具(redis-desktop-manager)操作redis

四、Redis数据结构

五、Redis常用指令

5.1 string类型

5.1.1 常用指令

5.1.2 应用场景之自增主键

5.2 hash类型

5.2.1 常用指令

5.2.2 string类型和hash类型的区别

5.2.3 应用之存储商品信息

5.3 list类型

5.3.1 ArrayList与LinkedList的区别

5.3.2 常用命令

5.3.3 应用之商品评论列表

5.4 set类型

5.4.1 set类型介绍

5.4.2 常用命令

5.4.3 集合运算命令

5.4.4 其他命令

5.5 zset类型（sortedset）

5.5.1 zset介绍

5.5.2 常用命令

5.5.3 其它命令

5..5.4 应用之商品销售排行榜

5.6 HyperLogLog命令

5.6.1 HyperLogLog命令介绍

5.6.2 HyperLogLog的优点

5.6.3 HyperLogLog 相关的一些基本命令。

六、Redis的通用命令

七、Redis的事务

7.1 Redis事务介绍

7.2 Redis事务命令

7.3 事务演示

7.4 事务失败处理

八、Redis发布订阅模式

九、Jedis连接Redis

9.1 创建项目，导入依赖

9.2 链接服务器

9.2.1 方案一 :单实例链接

9.2.2 方案二：连接池

十、Redis持久化方式

10.1 什么是Redis持久化

10.2 Redis 持久化存储方式

10.2.1 RDB持久化

10.2.2 AOF持久化

10.2.3 AOF与RDB区别

十一、Redis主从复制

11.1 主从搭建步骤:

十二、Redis哨兵模式

12.1 第一步：配置哨兵：

12.2 第二步：启动哨兵：

12.3 第三步：主机宕机

十三、Redis集群方案

13.1 redis-cluster架构图

13.2 redis-cluster投票:容错

13.3 集群搭建步骤

13.4 连接集群

13.5 查看集群信息

13.6 查看集群中节点信息

13.7 Jedis连接集群

13.7.1 关闭防火墙

13.7.2 代码实现

十四、Redis高端面试-缓存穿透，缓存击穿，缓存雪崩问题

14.1 缓存的概念

14.2 缓存雪崩

14.3 缓存穿透

14.4 缓存击穿

十五、Redis高端面试-分布式锁

15.1 使用分布式锁要满足的几个条件：

15.2 什么是分布式锁？

15.3 应用的场景

15.4 使用redis的setNX命令实现分布式锁

15.4.1 实现的原理：

15.4.2 基本命令解析

一、了解NoSql

1.1 什么是Nosql

NoSQL，泛指非关系型的数据库。随着互联网web2.0网站的兴起，传统的关系数据库在处理web2.0网站，特别是超大规模和高并发的SNS类型的web2.0纯动态网站已经显得力不从心，出现了很多难以克服的问题，而非关系型的数据库则由于其本身的特点得到了非常迅速的发展。NoSQL数据库的产生就是为了解决大规模数据集合，多重数据种类带来的挑战，尤其是大数据应用难题。

NoSQL最常见的解释是“non-relational”， “Not Only SQL”也被很多人接受。NoSQL仅仅是一个概念，泛指非关系型的数据库，区别于关系数据库，它们不保证关系数据的ACID特性。

1.2 为什么要使用NoSql

传统的数据库遇到的瓶颈：

传统的关系数据库具有不错的性能，高稳定型，久经历史考验，而且使用简单，功能强大，同时也积累了大量的成功案例。在互联网领域，MySQL成为了绝对靠前的王者，毫不夸张的说，MySQL为互联网的发展做出了卓越的贡献。

在90年代，一个网站的访问量一般都不大，用单个数据库完全可以轻松应付。在那个时候，更多的都是静态网页，动态交互类型的网站不多。
到了最近10年，网站开始快速发展。火爆的论坛、博客、sns、微博逐渐引领web领域的潮流。在初期，论坛的流量其实也不大，如果你接触网络比较早，你可能还记得那个时候还有文本型存储的论坛程序，可以想象一般的论坛的流量有多大。

现在网站的特点:
(1) 高并发读写
        Web2.0网站，数据库并发负载非常高，往往达到每秒上万次的读写请求
(2) 高容量存储和高效存储
        Web2.0网站通常需要在后台数据库中存储海量数据，如何存储海量数据并进行高效的查询往往是一个挑战
(3) 高扩展性和高可用性
        随着系统的用户量和访问量与日俱增，需要数据库能够很方便的进行扩展、维护

1.3 NoSql数据库的优势

(1) 易扩展

NoSQL数据库种类繁多，但是一个共同的特点都是去掉关系数据库的关系型特性。数据之间无关系，这样就非常容易扩展。也无形之间，在架构的层面上带来了可扩展的能力。

(2)大数据量，高性能

NoSQL数据库都具有非常高的读写性能，尤其在大数据量下，同样表现优秀。这得益于它的无关系性，数据库的结构简单。一般MySQL使用Query Cache，每次表的更新Cache就失效，是一种大粒度的 Cache，在针对web2.0的交互频繁的应用，Cache性能不高。而NoSQL的Cache是记录级的，是一种细粒度的Cache，所以NoSQL在这个层面上来说就要性能高很多了。

(3)灵活的数据模型

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

(4) 高可用

NoSQL在不太影响性能的情况，就可以方便的实现高可用的架构。比如Cassandra，HBase模型，通过复制模型也能实现高可用。

1.4 常见的NoSql产品

1.5 各产品的区别

二、Redis介绍

2.1什么是Redis

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

Redis 与其他 key - value 缓存产品有以下三个特点：
        (1) Redis支持数据的持久化，可以将内存中的数据保持在磁盘中，重启的时候可以再次加载进行使用
        (2) Redis不仅仅支持简单的key-value类型的数据，同时还提供list，set，zset，hash等数据结构的存储
        (3) Redis支持数据的备份，即master-slave(主从)模式的数据备份

2.2 Redis优势

(1) 性能极高 – Redis能读的速度是110000次/s,写的速度是81000次/s 。
(2) 丰富的数据类型 – Redis支持二进制案例的 Strings, Lists, Hashes, Sets 及 Ordered Sets 数据类型操作。
(3) 原子 – Redis的所有操作都是原子性的，同时Redis还支持对几个操作全并后的原子性执行。
(4) 丰富的特性 – Redis还支持 publish/subscribe, 通知, key 过期等等特性
(5) 采用单线程，避免了不必要的上下文切换和竞争条件，也不存在多进程或者多线程导致的切换而消耗 CPU，不用去考虑各种锁的问题，不存在加锁释放锁操作，没有因为可能出现死锁而导致的性能消耗；
(6) 使用多路I/O复用模型，非阻塞IO；

2.3 Redis应用场景

(1) 缓存(数据查询，短连接，新闻内容，商品内容等)，使用最多
(2) 聊天室在线好友列表
(3) 任务队列(秒杀，抢购，12306等)
(4) 应用排行榜
(5) 网站访问统计
(6) 数据过期处理(可以精确到毫秒)
(7) 分布式集群架构中的session问题

2.4 Redis下载

（1）Http://redis.io/ 英文地址
（2）Http://www.redis.cn/ 中文地址

三、Linux下安装Redis

3.1 环境准备

(1)虚拟机版本:VMware® Workstation 12 Pro
(2) Linux系统:Centos Release 6.5
(3) 远程命令端:xshell
(4)文件传输工具:SecureFXPortable

3.2 Redis的安装

3.2.1 Redis的编译环境

Redis是C语言开发的，安装redis需要先去官网下载源码进行编译，编译需要依赖于GCC编译环境，如果CentOS上没有安装gcc编译环境，需要提前安装，安装命令如下:（这里我们使用root用户处理这些操作）

[root@localhost ~]# yum install gcc-c++

如果提示是否下载，选择: y

如果提示是否安装,选择: y

3.2.2 Redis的安装

(1) 使用SecureFXPortable上传Redis安装文件到Linux目录

(2)上传Redis安装文件,这里我上传自建文件夹: /home/yhp/local

(3)解压redis文件

[root@localhost local]# tar -zxvf redis-5.0.5.tar.gz

(4)编译Redis(编译,将.c文件编译为.o文件)

进入解压文件夹,执行 make

[root@localhost local]# cd redis-5.0.5
[root@localhost redis-5.0.5]# make

编译成功！如果编译过程中出错，先删除安装文件目录，后解压重新编译。

(5) 安装

[root@localhost redis-5.0.5]# make PREFIX=/home/admin/myapps/redis install

说明:这里的/home/myapps/redis 是自定义的redis安装路径

(6)安装之后的bin目录

bin文件夹下的命令:

(7) Copy文件

Redis启动需要一个配置文件，可以修改端口号信息。将redis解压的文件夹中的redis.conf文件复制到安装目录

[root@localhost redis-5.0.5]# cp redis.conf /home/admin/myapps/redis

3.3 Redis的启动

3.3.1 Redis的前端模式启动（了解）

直接运行bin/redis-server将使永前端模式启动，前端模式启动的缺点是启动完成后，不能再进行其他操作，如果要操作必须使用ctrl+c，同时redis-server程序结束，不推荐此方法。

[root@localhost bin]# ./redis-server

下面是启动界面(这个界面只能启动，启动后不能进行其他操作)

使用ctrl+c退出前端启动。

3.3.2 Redis的后端启动

修改redis.conf配置文件，设置:daemonize yes,然后可以使用后端模式启动。

[root@localhost redis]# vi redis.conf

启动时，指定配置文件(这里所在文件夹是redis)

[root@localhost redis]# ./bin/redis-server ./redis.conf

Redis默认端口:6379,通过当前服务进行查看

[root@localhost redis]# ps -ef | grep -i redis

3.3.3 客户端访问redis

如果想要通过指令来操作redis，可以使用redis的客户端进行操作,在bin文件夹下运行redis-cli

该指令默认连接的127.0.0.1 ，端口号是6379

[root@localhost bin]# ./redis-cli
127.0.0.1:6379>

如果想要连接指定的ip地址以及端口号，则需要按照

redis-cli -h ip地址 -p 端口号

3.3.4 向Redis服务器发送命令

Ping，测试客户端与Redis的连接是否正常，如果连接正常，回收到pong

127.0.0.1:6379> ping
PONG

3.3.5 退出客户端

127.0.0.1:6379> quit

3.3.6 Redis的停止

(1) 强制结束程序。强制终止Redis进程可能会导致redis持久化数据丢失。

语法:kill -9 pid (2) 正确停止Redis的方式应该是向Redis发送SHUTDOWN命令，方法为（关闭默认的端口）

3.3.7 第三方工具(redis-desktop-manager)操作redis

注意:需要关闭linux防火墙并且修改redis.conf文件中的bind参数

bind linux的ip地址

此时如果通过redis客户端访问的时候，代码如下:

./redis-cli -h 192.168.197.132 -p 6379

四、Redis数据结构

Redis 是一种基于内存的数据库，并且提供一定的持久化功能，它是一种键值（key-value）数据库，使用 key 作为索引找到当前缓存的数据，并且返回给程序调用者。

当前的 Redis 支持 6 种数据类型，它们分别是字符串（String）、列表（List）、集合（set）、哈希结构（hash）、有序集合（zset）和基数（ HyperLogLog ）

五、Redis常用指令

5.1 string类型

5.1.1 常用指令

(1) 赋值

语法：

SET key value

示例：

127.0.0.1:6379> set test 123
OK

(2) 取值

语法：
GET key
示例：
127.0.0.1:6379> get test 
"123"

(3) 取值并赋值

语法：

GETSET key value

示例：

127.0.0.1:6379> getset s2 222 
"111" 
127.0.0.1:6379> get s2 
"222"

(4) 数值增减

注意事项 :

1 、当 value 为整数数据时，才能使用以下命令操作数值的增减。

2 、数值递增都是【原子】操作。

3 、 redis 中的每一个单独的命令都是原子性操作。当多个命令一起执行的时候，就不能保证原子性，不过我们可以使用事务和lua 脚本来保证这一点。

非原子性操作示例：
int i = 1; 
i++; 
System.out.println(i)

(5) 递增数字

语法（increment）：

INCR key

示例：

127.0.0.1:6379> incr num (integer) 
1 
127.0.0.1:6379> incr num (integer) 
2 
127.0.0.1:6379> incr num (integer) 
3

(6) 增加指定的整数

语法：

INCRBY key increment

示例：

127.0.0.1:6379> incrby num 2 
(integer) 5 
127.0.0.1:6379> incrby num 2 
(integer) 7 
127.0.0.1:6379> incrby num 2 
(integer) 9

(7) 递减数值

语法：

DECR key

示例：

127.0.0.1:6379> incr num 
(integer) 1 
127.0.0.1:6379> incr num 
(integer) 2 
127.0.0.1:6379> incr num 
(integer) 3

(8) 减少指定的整数

语法：

DECRBY key decrement

示例:

127.0.0.1:6379> decr num 
(integer) 6 
127.0.0.1:6379> decr num 
(integer) 5 
127.0.0.1:6379> decrby num 3 
(integer) 2 
127.0.0.1:6379> decrby num 3 
(integer) -1

(9) 仅当不存在时赋值

使用该命令可以实现【分布式锁】的功能，后续讲解！！！

语法：

setnx key value

示例：

redis> EXISTS job               # job 不存在 
(integer) 0 
redis> SETNX job "programmer"   # job 设置成功 
(integer) 1 
redis> SETNX job "code-farmer"  # 尝试覆盖 job ，失败 
(integer) 0 
redis> GET job                  # 没有被覆盖 
"programmer"

(10) 其它命令

1.向尾部追加值

APPEND 命令，向键值的末尾追加 value 。

如果键不存在则将该键的值设置为 value ，即相当于 SET key value 。返回值是追加后字符串的总长度。
语法：
APPEND key value
示例：
127.0.0.1:6379> set str hello 
OK
127.0.0.1:6379> append str " world!" 
(integer) 12 
127.0.0.1:6379> get str 
"hello world!"

2.获取字符串长度

STRLEN 命令，返回键值的长度，如果键不存在则返回0。
语法：
STRLEN key
示例：
127.0.0.1:6379> strlen str 
(integer) 0 
127.0.0.1:6379> set str hello 
OK
127.0.0.1:6379> strlen str 
(integer) 5

3.同时设置/获取多个键值

语法：

MSET key value [key value …]
MGET key [key …]

示例：

127.0.0.1:6379> mset k1 v1 k2 v2 k3 v3 
OK
127.0.0.1:6379> get k1 "v1" 
127.0.0.1:6379> mget k1 k3 
1) "v1" 
2) "v3"

5.1.2 应用场景之自增主键

需求：商品编号、订单号采用 INCR 命令生成。

设计： key 命名要有一定的设计

实现：定义商品编号 key ： items:id

192.168.101.3:7003> INCR items:id (integer) 
2 
192.168.101.3:7003> INCR items:id (integer) 
3

5.2 hash类型

hash 类型也叫散列类型，它提供了字段和字段值的映射。字段值只能是字符串类型，不支持散列类型、集合类型等其它类型。如下：

5.2.1 常用指令

(1) 赋值

HSET 命令不区分插入和更新操作，当执行插入操作时 HSET 命令返回 1 ，当执行更新操作时返回 0 。

1.设置一个字段值:

语法：

HSET key field value

示例：

127.0.0.1:6379> hset user username zhangsan 
(integer) 1

2.设置多个字段值

语法：

HMSET key field value [field value ...]

示例：

127.0.0.1:6379> hmset user age 20 username lisi 
OK

3.当字段不存在时赋值

类似 HSET ，区别在于如果字段存在，该命令不执行任何操作
语法：
HSETNX key field value
示例：
127.0.0.1:6379> hsetnx user age 30 # 如果user中没有age字段则设置age值为30，否则不做任何操作 
(integer) 0

(2) 取值

1.获取一个字段值

语法：

HGET key field

示例：

127.0.0.1:6379> hget user username 
"zhangsan“

2.获取多个字段值

语法：

HMGET key field [field ...] 1

示例：

127.0.0.1:6379> hmget user age username 
1) "20" 
2) "lisi"

3.获取所有字段值

语法：

HGETALL key

示例：

127.0.0.1:6379> hgetall user 
1) "age" 
2) "20" 
3) "username" 
4) "lisi"

(3) 删除字段

可以删除一个或多个字段，返回值是被删除的字段个数
语法：
HDEL key field [field ...]
示例：
127.0.0.1:6379> hdel user age 
(integer) 1 
127.0.0.1:6379> hdel user age name 
(integer) 0 
127.0.0.1:6379> hdel user age username 
(integer) 1

(4) 增加数字

语法：

HINCRBY key field increment

示例：

127.0.0.1:6379> hincrby user age 2 # 将用户的年龄加2 
(integer) 22 
127.0.0.1:6379> hget user age # 获取用户的年龄 
"22“

(5) 其它命令

1.判断字段是否存在

语法：

HEXISTS key field

示例：

127.0.0.1:6379> hexists user age 查看user中是否有age字段 
(integer) 1 
127.0.0.1:6379> hexists user name 查看user中是否有name字段 
(integer) 0

2.只获取字段名或字段值

语法：

HKEYS key 
HVALS key

示例：

127.0.0.1:6379> hmset user age 20 name lisi 
OK
127.0.0.1:6379> hkeys user 
1) "age" 
2) "name" 
127.0.0.1:6379> hvals user 
1) "20" 
2) "lisi"

3.获取字段数量

语法：

HLEN key

示例：

127.0.0.1:6379> hlen user 
(integer) 2

4.获取所有字段

获得 hash 的所有信息，包括 key 和 value
语法：
hgetall key

5.2.2 string类型和hash类型的区别

hash类型适合存储那些对象数据，特别是对象属性经常发生【增删改】操作的数据。 string类型也可以存储对象数据，将java对象转成json字符串进行存储，这种存储适合【查询】操作。

5.2.3 应用之存储商品信息

商品信息字段

【商品id、商品名称、商品描述、商品库存、商品好评】

定义商品信息的key

商品ID为1001的信息在 Redis中的key为：[items:1001]

存储商品信息

192.168.101.3:7003> HMSET items:1001 id 3 name apple price 999.9 
OK

获取商品信息

192.168.101.3:7003> HGET items:1001 id 
"3" 
192.168.101.3:7003> HGETALL items:1001 
1) "id" 
2) "3" 
3) "name"
4) "apple" 
5) "price" 
6) "999.9"

5.3 list类型

Redis 的列表类型（ list 类型）可以存储一个有序的字符串列表，常用的操作是向列表两端添加元素，或者获得列表的某一个片段。

列表类型内部是使用双向链表（ double linked list ）实现的，所以向列表两端添加元素的时间复杂度为 o(1) ，获取越接近两端的元素速度就越快。这意味着即使是一个有几千万个元素的列表，获取头部或尾部的10条记录也是极快的。

5.3.1 ArrayList与LinkedList的区别

ArrayList 使用数组方式存储数据，所以根据索引查询数据速度快，而新增或者删除元素时需要设计到位移操作，所以比较慢。

LinkedList 使用双向链表方式存储数据，每个元素都记录前后元素的指针，所以插入、删除数据时只是更改前后元素的指针指向即可，速度非常快。然后通过下标查询元素时需要从头开始索引，所以比较慢，但是如果查询前几个元素或后几个元素速度比较快。

5.3.2 常用命令

(1) LPUSH/RPUSH

语法：

LPUSH key value [value ...] 
RPUSH key value [value ...]

示例：

127.0.0.1:6379> lpush list:1 1 2 3 
(integer) 3 
127.0.0.1:6379> rpush list:1 4 5 6 
(integer) 3

(2) LRANGE

获取列表中的某一片段。将返回 `start` 、 `stop` 之间的所有元素（包含两端的元素），索引从`0` 开始。索引可以是负数，如：“`-1`” 代表最后边的一个元素。

语法：

LRANGE key start stop

示例：

127.0.0.1:6379> lrange list:1 0 2 
1) "2" 
2) "1" 
3) "4"

(3) LPOP/RPOP

从列表两端弹出元素

从列表左边弹出一个元素，会分两步完成：

第一步是将列表左边的元素从列表中移除

第二步是返回被移除的元素值。

语法：

LPOP key 
RPOP key

示例 :

127.0.0.1:6379>lpop list:1 
"3"
127.0.0.1:6379>rpop list:1 
"6"

(4) LLEN

获取列表中元素的个数
语法：
llen key
示例：
127.0.0.1:6379> llen list:1 
(integer) 2

(5) 其它命令

1. LREM

删除列表中指定个数的值

LREM 命令会删除列表中前 count 个值为 value 的元素，返回实际删除的元素个数。根据 count 值的不同，该命令的执行方式会有所不同：

- 当 count>0 时， LREM 会从列表左边开始删除。

- 当 count<0 时， LREM 会从列表后边开始删除。

- 当 count=0 时， LREM 删除所有值为 value 的元素。

语法：
LREM key count value

2. LINDEX

获得指定索引的元素值
语法：
LINDEX key index
示例：
127.0.0.1:6379>lindex l:list 2 
"1"

3. 设置指定索引的元素值

语法：

LSET key index value

示例：

127.0.0.1:6379> lset l:list 2 2 
OK 
127.0.0.1:6379> lrange l:list 0 -1 
1) "6" 
2) "5" 
3) "2" 
4) "2"

4. LTRIM

只保留列表指定片段 , 指定范围和 LRANGE 一致

语法：

LTRIM key start stop

示例：

127.0.0.1:6379> lrange l:list 0 -1 
1) "6" 
2) "5" 
3) "0" 
4) "2" 
127.0.0.1:6379> ltrim l:list 0 2 
OK
127.0.0.1:6379> lrange l:list 0 -1 
1) "6" 
2) "5" 
3) "0"

5. LINSERT

向列表中插入元素。

该命令首先会在列表中从左到右查找值为 pivot 的元素，然后根据第二个参数是BEFORE 还是 AFTER 来决定将 value 插入到该元素的前面还是后面。
语法：
LINSERT key BEFORE|AFTER pivot value
示例：
127.0.0.1:6379> lrange list 0 -1 
1) "3" 
2) "2" 
3) "1" 
127.0.0.1:6379> linsert list after 3 4 
(integer) 4 
127.0.0.1:6379> lrange list 0 -1 
1) "3" 
2) "4" 
3) "2" 
4) "1"

6. RPOPLPUSH

将元素从一个列表转移到另一个列表中

语法：

RPOPLPUSH source destination

示例：

127.0.0.1:6379> rpoplpush list newlist 
"1" 
127.0.0.1:6379> lrange newlist 0 -1 
1) "1" 
127.0.0.1:6379> lrange list 0 -1 
1) "3" 
2) "4" 
3) "2"

5.3.3 应用之商品评论列表

需求：

用户针对某一商品发布评论，一个商品会被不同的用户进行评论，存储商品评论时，要按时间顺序排序。

用户在前端页面查询该商品的评论，需要按照时间顺序降序排序。

分析：

使用 list 存储商品评论信息， KEY 是该商品的 ID ， VALUE 是商品评论信息列表

实现：商品编号为 1001 的商品评论 key 【 items: comment:1001 】

192.168.101.3:7001> LPUSH items:comment:1001 '{"id":1,"name":"商品不错，很 好！！","date":1430295077289}'

5.4 set类型

5.4.1 set类型介绍

set 类型即集合类型，其中的数据是不重复且没有顺序。

集合类型和列表类型的对比：

集合类型的常用操作是向集合中加入或删除元素、判断某个元素是否存在等，由于集合类型的 Redis 内部是使用值为空的散列表实现，所有这些操作的时间复杂度都为 0(1) 。

Redis 还提供了多个集合之间的交集、并集、差集的运算。

5.4.2 常用命令

(1) SADD/SREM 添加元素/删除元素

语法：

SADD key member [member ...]
SREM key member [member ...]

示例：

127.0.0.1:6379> sadd set a b c
(integer) 3
127.0.0.1:6379> sadd set a
(integer) 0
127.0.0.1:6379> srem set c d
(integer) 1

(2) SMEMBERS 获得集合中的所有元素

语法：

SMEMBERS key

示例：

127.0.0.1:6379> smembers set
1) "b"
2) "a”

(3) SISMEMBER 判断元素是否在集合中

语法：

SISMEMBER key member

示例：

127.0.0.1:6379>sismember set a
(integer) 1
127.0.0.1:6379>sismember set h
(integer) 0

5.4.3 集合运算命令

(1) SDIFF 集合的差集运算 A-B：属于A并且不属于B的元素构成的集合。

语法：

SDIFF key [key ...]

示例：

127.0.0.1:6379> sadd setA 1 2 3
(integer) 3
127.0.0.1:6379> sadd setB 2 3 4
(integer) 3
127.0.0.1:6379> sdiff setA setB
1) "1"
127.0.0.1:6379> sdiff setB setA
1) "4"

(2) SINTER 集合的交集运算 A ∩ B：属于A且属于B的元素构成的集合。

语法：

SINTER key [key ...]

示例：

127.0.0.1:6379> sinter setA setB
1) "2"
2) "3"

(3) SUNION 集合的并集运算 A ∪ B：属于A或者属于B的元素构成的集合

语法：

SUNION key [key ...]

示例：

127.0.0.1:6379> sunion setA setB
1) "1"
2) "2"
3) "3"
4) "4"

5.4.4 其他命令

(1) SCARD 获得集合中元素的个数

语法：

SCARD key

示例：

127.0.0.1:6379> smembers setA
1) "1"
2) "2"
3) "3"
127.0.0.1:6379> scard setA
(integer) 3

(2) SPOP

从集合中弹出一个元素。

注意：由于集合是无序的，所有 SPOP 命令会从集合中随机选择一个元素弹出

语法：
SPOP key
示例：
127.0.0.1:6379> spop setA
"1"

5.5 zset类型（sortedset）

5.5.1 zset介绍

在 set 集合类型的基础上，有序集合类型为集合中的每个元素都关联一个分数，这使得我们不仅可以完成插入、删除和判断元素是否存在在集合中，还能够获得分数最高或最低的前N个元素、获取指定分数范围内的元素等与分数有关的操作。

在某些方面有序集合和列表类型有些相似：

1 、二者都是有序的。

2 、二者都可以获得某一范围的元素。

但是，二者有着很大区别：

1 、列表类型是通过链表实现的，获取靠近两端的数据速度极快，而当元素增多后，访问中间数据的速度会变慢。

2 、有序集合类型使用散列表实现，所有即使读取位于中间部分的数据也很快。

3 、列表中不能简单的调整某个元素的位置，但是有序集合可以（通过更改分数实现）

4 、有序集合要比列表类型更耗内存。

5.5.2 常用命令

(1) ZADD

增加元素。

向有序集合中加入一个元素和该元素的分数，如果该元素已经存在则会用新的分数替换原有的分数。返回值是新加入到集合中的元素个数，不包含之前已经存在的元素。

语法：

ZADD key score member [score member ...]

示例：

127.0.0.1:6379> zadd scoreboard 80 zhangsan 89 lisi 94 wangwu
(integer) 3
127.0.0.1:6379> zadd scoreboard 97 lisi
(integer) 0

(2) ZRANGE/ZREVRANGE

获得排名在某个范围的元素列表。
- ZRANGE：按照元素分数从小到大的顺序返回索引从start到stop之间的所有元素（包含两端的元素）
- ZREVRANGE：按照元素分数从大到小的顺序返回索引从start到stop之间的所有元素（包含两端的元素）

语法：

ZRANGE key start stop [WITHSCORES]
ZREVRANGE key start stop [WITHSCORES]

示例：

127.0.0.1:6379> zrange scoreboard 0 2
1) "zhangsan"
2) "wangwu"
3) "lisi“
127.0.0.1:6379> zrevrange scoreboard 0 2
1) " lisi "
2) "wangwu"
3) " zhangsan “

如果需要获得元素的分数的可以在命令尾部加上 WITHSCORES 参数

127.0.0.1:6379> zrange scoreboard 0 1 WITHSCORES
1) "zhangsan"
2) "80"
3) "wangwu"
4) "94"

(3) ZSCORE 获取元素的分数

语法：

ZSCORE key member

示例：

127.0.0.1:6379> zscore scoreboard lisi
"97"

(4) ZREM

删除元素。

移除有序集合 key 中的一个或多个成员，不存在的成员将被忽略。

当 key 存在但不是有序集类型时，返回一个错误。

语法：

ZREM key member [member ...]

示例：

127.0.0.1:6379> zrem scoreboard lisi
(integer) 1

5.5.3 其它命令

(1) ZRANGEBYSCORE 获得指定分数范围的元素。

语法：

ZRANGEBYSCORE key min max [WITHSCORES]

示例：

127.0.0.1:6379> ZRANGEBYSCORE scoreboard 90 97 WITHSCORES
1) "wangwu"
2) "94"
3) "lisi"
4) "97"
127.0.0.1:6379> ZRANGEBYSCORE scoreboard 70 100 limit 1 2
1) "wangwu"
2) "lisi"

(2) ZINCRBY

增加某个元素的分数。

返回值是更改后的分数

语法：

ZINCRBY key increment member

示例：

127.0.0.1:6379> ZINCRBY scoreboard 4 lisi
"101"

(3) ZCARD 获得集合中元素的数量。

语法：

ZCARD key

示例：

127.0.0.1:6379> ZCARD scoreboard
(integer) 3

(4) ZCOUNT 获得指定分数范围内的元素个数

语法：

ZCOUNT key min max

示例：

127.0.0.1:6379> ZCOUNT scoreboard 80 90
(integer) 1

(5) ZREMRANGEBYRANK 按照排名范围删除元素

语法：

ZREMRANGEBYRANK key start stop

示例：

127.0.0.1:6379> ZREMRANGEBYRANK scoreboard 0 1
(integer) 2
127.0.0.1:6379> ZRANGE scoreboard 0 -1
1) "lisi"

(6) ZREMRANGEBYSCORE 按照分数范围删除元素

语法：

ZREMRANGEBYSCORE key min max

示例：

127.0.0.1:6379> zadd scoreboard 84 zhangsan
(integer) 1
127.0.0.1:6379> ZREMRANGEBYSCORE scoreboard 80 100
(integer) 1

(7) ZRANK/ZREVRANK

获取元素的排名。

- ZRANK ：从小到大

- ZREVRANK ：从大到小

语法：

ZRANK key member
ZREVRANK key member

示例：

127.0.0.1:6379> ZRANK scoreboard lisi
(integer) 0
127.0.0.1:6379> ZREVRANK scoreboard zhangsan
(integer) 1

5..5.4 应用之商品销售排行榜

需求：

根据商品销售量对商品进行排行显示

设计：

定义商品销售排行榜（ sorted set 集合）， Key 为 items:sellsort ，分数为商品销售量。

写入商品销售量：商品编号 1001 的销量是 9 ，商品编号 1002 的销量是 10:

192.168.101.3:7007> ZADD items:sellsort 9 1001 10 1002

商品编号 1001 的销量加 1:

192.168.101.3:7001> ZINCRBY items:sellsort 1 1001

商品销量前 10 名：

192.168.101.3:7001> ZREVRANGE items:sellsort 0 9 withscores

5.6 HyperLogLog命令

5.6.1 HyperLogLog命令介绍

HyperLogLog 是一种使用随机化的算法，以少量内存提供集合中唯一元素数量的近似值。
HyperLogLog 可以接受多个元素作为输入，并给出输入元素的基数估算值：
基数：集合中不同元素的数量。比如 {‘apple’, ‘banana’, ‘cherry’, ‘banana’, ‘apple’} 的基数就是3。
估算值：算法给出的基数并不是精确的，可能会比实际稍微多一些或者稍微少一些，但会控制在合理的范围之内。

5.6.2 HyperLogLog的优点

HyperLogLog 的优点是，即使输入元素的数量或者体积非常非常大，计算基数所需的空间总是固定的、并且是很小的。

在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。

但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。

5.6.3 HyperLogLog 相关的一些基本命令。

命令	说明
PFADD key element [element …]	将指定的元素添加到指定的 HyperLogLog 中
PFCOUNT key [key …]	返回给定 HyperLogLog 的基数估算值
PFMERGE destkey sourcekey [sourcekey …]	将多个 HyperLogLog 合并为一个 HyperLogLog

示例:

redis 127.0.0.1:6379> PFADD mykey "redis"
1) (integer) 1
redis 127.0.0.1:6379> PFADD mykey "java"
1) (integer) 1
redis 127.0.0.1:6379> PFADD mykey "mysql"
1) (integer) 1
redis 127.0.0.1:6379> PFCOUNT mykey
(integer) 3

六、Redis的通用命令

(1) keys

返回满足给定 pattern 的所有 key
语法：
keys pattern
示例：
redis 127.0.0.1:6379> keys mylist*
1) "mylist"
2) "mylist5"
3) "mylist6"
4) "mylist7"
5) "mylist8"

(2) del

语法：

DEL key

示例：

127.0.0.1:6379> del test
(integer) 1

(3) exists 确认一个key 是否存在

语法：
exists key
示例：从结果来看，数据库中不存在 HongWan 这个 key ，但是 age 这个 key 是存在的
redis 127.0.0.1:6379> exists HongWan
(integer) 0
redis 127.0.0.1:6379> exists age
(integer) 1

(4) expire（重点）

Redis 在实际使用过程中更多的用作缓存，然而缓存的数据一般都是需要设置生存时间的，即：到期后数据销毁。

语法：

EXPIRE key seconds 设置key的生存时间（单位：秒）key在多少秒后会自动删除
TTL key 查看key生于的生存时间
PERSIST key 清除生存时间
PEXPIRE key milliseconds 生存时间设置单位为：毫秒

示例：

192.168.101.3:7002> set test 1 设置test的值为1
OK
192.168.101.3:7002> get test 获取test的值
"1"
192.168.101.3:7002> EXPIRE test 5 设置test的生存时间为5秒
(integer) 1
192.168.101.3:7002> TTL test 查看test的生于生成时间还有1秒删除
(integer) 1
192.168.101.3:7002> TTL test
(integer) -2
192.168.101.3:7002> get test 获取test的值，已经删除
(nil)

(5) rename 重命名key

语法：

rename oldkey newkey

示例： age 成功的被我们改名为 age_new 了

redis 127.0.0.1:6379[1]> keys *
1) "age"
redis 127.0.0.1:6379[1]> rename age age_new
OK
redis 127.0.0.1:6379[1]> keys *
1) "age_new"

(6) type 显示指定key的数据类型

语法：
type key
示例：这个方法可以非常简单的判断出值的类型
redis 127.0.0.1:6379> type addr
string
redis 127.0.0.1:6379> type myzset2
zset
redis 127.0.0.1:6379> type mylist
list

七、Redis的事务

7.1 Redis事务介绍

Redis 的事务是通过 MULTI 、 EXEC 、 DISCARD 和 WATCH 、UNWATCH这五个命令来完成的。
Redis 的单个命令都是原子性的，所以这里需要确保事务性的对象是命令集合。
Redis 将命令集合序列化并确保处于同一事务的命令集合连续且不被打断的执行
Redis 不支持回滚操作。

7.2 Redis事务命令

(1) MULTI

用于标记事务块的开始。

Redis 会将后续的命令逐个放入队列中，然后使用 EXEC 命令原子化地执行这个命令序列。

语法： multi

(2) EXEC

在一个事务中执行所有先前放入队列的命令，然后恢复正常的连接状态

语法： exec

(3) DISCARD

清除所有先前在一个事务中放入队列的命令，然后恢复正常的连接状态。

语法： discard

(4) WATCH

当某个 [ 事务需要按条件执行 ] 时，就要使用这个命令将给定的 [ 键设置为受监控 ] 的状态。

语法：watch key [key…]

(5) UNWATCH

清除所有先前为一个事务监控的键。

语法： unwatch

7.3 事务演示

示例一：

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set s1 111
QUEUED
127.0.0.1:6379> hset set1 name zhangsan
QUEUED
127.0.0.1:6379> exec
1) OK
2) (integer) 1
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set s2 222
QUEUED
127.0.0.1:6379> hset set2 age 20
QUEUED
127.0.0.1:6379> discard
OK
127.0.0.1:6379> exec
(error) ERR EXEC without MULTI
127.0.0.1:6379> watch s1
OK
127.0.0.1:6379> multi
OK
127.0.0.1:6379> set s1 555
QUEUED
127.0.0.1:6379> exec # 此时在没有exec之前，通过另一个命令窗口对监控的s1字段进行修改
(nil)
127.0.0.1:6379> get s1
111

示例二：

127.0.0.1:6379> multi
OK
127.0.0.1:6379> set u1 user1
QUEUED
127.0.0.1:6379> get u1
QUEUED
127.0.0.1:6379> sadd tag c++ html java
QUEUED
127.0.0.1:6379> smembers tag
QUEUED
127.0.0.1:6379> exec
1) OK
2) "user1"
3) (integer) 3
4) 1) "java"
   2) "html"
   3) "c++"

7.4 事务失败处理

(1) Redis 语法错误（编译期）

(2) Redis 运行错误

(3) Redis 不支持事务回滚（为什么呢）

1、大多数事务失败是因为语法错误或者类型错误，这两种错误，在开发阶段都是可以预见的

2、 Redis 为了性能方面就忽略了事务回滚。

八、Redis发布订阅模式

Redis 发布订阅 (pub/sub) 是一种消息通信模式：发送者 (pub) 发送消息，订阅者 (sub) 接收消息。

Redis 客户端可以订阅任意数量的频道。

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

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

在我们实例中我们创建了订阅频道名为 redisMessage:

127.0.0.1:6379> subscribe redisMessage
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redisMessage"
3) (integer) 1

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

127.0.0.1:6379> publish redisMessage "demo1 test"
(integer) 1
127.0.0.1:6379> publish redisMessage "demo2 test"
(integer) 1
127.0.0.1:6379> publish redisMessage "demo3 test"
(integer) 1

# 订阅者的客户端会显示如下消息

127.0.0.1:6379> subscribe redisMessage
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "redisMessage"
3) (integer) 1
1) "message"
2) "redisMessage"
3) "demo1 test"
1) "message"
2) "redisMessage"
3) "demo2 test"
1) "message"
2) "redisMessage"
3) "demo3 test"

九、Jedis连接Redis

9.1 创建项目，导入依赖

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

注意：

1 ）确认远程服务器是否可以 ping 通 : ping vm 的 ip 地址：

2) 确认防火墙是否关闭或放行

service iptables stop
service iptables status

9.2 链接服务器

9.2.1 方案一 :单实例链接

Jedis jedis = new Jedis(“ip地址”, 端口号);//建立链接

核心代码：

public static void main(String[] args) {
    Jedis jedis=new Jedis("192.168.197.129",6379);
    //设置值
    jedis.set("java001","java工程师");
    String java001 = jedis.get("java001");
    System.out.println(java001);
}

常见异常 :

解决方案:

虚拟机客户端连接的 ip 是 127.0.0.1, 意思是连接的本机 , 其他机器无法连接 , 这里需要修改配置文件 , 将连接地址改为虚拟机的地址, 就可以了

修改 redis.conf 文件里面的 bind 连接地址 , 将连接地址改为自己虚拟机的 ip
bind 192.168.197.129

重新启动服务 ,Jedis 就可以正常连上了

Idea中控制台打印：

服务器上存储：

9.2.2 方案二：连接池

jedis 连接池连接 , 后面会使用 Spring 的配置文件来整合。

// 1.获取连接池配置对象,设置配置项
JedisPoolConfig config = new JedisPoolConfig();
// 1.1最大的连接数
config.setMaxTotal(30);
// 1.2最大的空闲
config.setMaxIdle(10);
// 2.获取连接池
JedisPool jedisPool = new JedisPool(config, "192.168.197.129", 6379);
Jedis jedis = null;
try {
    jedis = jedisPool.getResource();
    // 3.设置数据
    jedis.set("name", "张三");
    String name = jedis.get("name");
    System.out.println("name=" + name);
} catch (Exception e) {
    e.printStackTrace();
} finally {
    if (jedis != null) {
        jedis.close();
    }
    // 4.虚拟机关闭的时候，释放资源
    if (jedisPool != null) {
        jedisPool.close();
    }
}

服务端存储确认：

十、Redis持久化方式

10.1 什么是Redis持久化

由于 redis 的值放在内存中，为防止突然断电等特殊情况的发生，需要对数据进行持久化备份。即将内存数据保存

到硬盘。

10.2 Redis 持久化存储方式

10.2.1 RDB持久化

RDB 是以二进制文件，是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复。

优点：使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 redis 的高性能

缺点： RDB 是间隔一段时间进行持久化，如果持久化之间 redis 发生故障，会发生数据丢失。所以这种方式更适合数据要求不严谨的时候

这里说的这个执行数据写入到临时文件的时间点是可以通过配置来自己确定的，通过配置 redis 在 n 秒内如果超过 m 个 key 被修改这执行一次 RDB 操作。这个操作就类似于在这个时间点来保存一次 Redis 的所有数据，一次快照数据。所有这个持久化方法也通常叫做 snapshots 。

RDB 默认开启， redis.conf 中的具体配置参数如下；

#dbfilename：持久化数据存储在本地的文件
dbfilename dump.rdb
#dir：持久化数据存储在本地的路径，如果是在/redis/redis-5.0.5/src下启动的redis-cli，则数据会存储在当前src目录下
dir ./
##snapshot触发的时机，save
##如下为900秒后，至少有一个变更操作，才会snapshot
##对于此值的设置，需要谨慎，评估系统的变更操作密集程度
##可以通过“save”来关闭snapshot功能
#save时间，以下分别表示更改了1个key时间隔900s进行持久化存储；更改了10个key300s进行存储；更改10000个key60s进行存储。
save 900 1
save 300 10
save 60 10000
##当snapshot时出现错误无法继续时，是否阻塞客户端“变更操作”，“错误”可能因为磁盘已满/磁盘故障/OS级别异常等
stop-writes-on-bgsave-error yes
##是否启用rdb文件压缩，默认为“yes”，压缩往往意味着“额外的cpu消耗”，同时也意味这较小的文件尺寸以及较短的网络传输时间
rdbcompression yes

注意 : 测试时使用 root 用户操作

10.2.2 AOF持久化

Append-Only File ，将 “ 操作 + 数据 ” 以格式化指令的方式追加到操作日志文件的尾部，在 append 操作返回后 ( 已经写入到文件或者将要写入) ，才进行实际的数据变更， “ 日志文件 ” 保存了历史所有的操作过程；当 server 需要数据恢复时，可以直接 replay 此日志文件，即可还原所有的操作过程。 AOF 相对可靠， AOF 文件内容是字符串，非常容易阅读和解析。

优点：可以保持更高的数据完整性，如果设置追加 fifile 的时间是 1s ，如果 redis 发生故障，最多会丢失 1s 的数据；且如果日志写入不完整支持 redis-check-aof 来进行日志修复； AOF 文件没被 rewrite 之前（文件过大时会对命令进行合并重写），可以删除其中的某些命（比如误操作的 flflushall ）。

缺点： AOF 文件比 RDB 文件大，且恢复速度慢。

我们可以简单的认为 AOF 就是日志文件，此文件只会记录 “ 变更操作 ”( 例如： set/del 等 ) ，如果 server 中持续的大量变更操作，将会导致 AOF 文件非常的庞大，意味着 server 失效后，数据恢复的过程将会很长；事实上，一条数据经过多次变更，将会产生多条 AOF 记录，其实只要保存当前的状态，历史的操作记录是可以抛弃的；因为 AOF 持久化模式还伴生了“AOF rewrite” 。

AOF 的特性决定了它相对比较安全，如果你期望数据更少的丢失，那么可以采用 AOF 模式。如果 AOF 文件正在被写入时突然 server 失效，有可能导致文件的最后一次记录是不完整，你可以通过手工或者程序的方式去检测并修正不完整的记录，以便通过 aof 文件恢复能够正常；同时需要提醒，如果你的 redis 持久化手段中有 aof ，那么在 server 故障失效后再次启动前，需要检测 aof 文件的完整性。

AOF 默认关闭，开启方法，修改配置文件 reds.conf：appendonly yes

##此选项为aof功能的开关，默认为“no”，可以通过“yes”来开启aof功能
##只有在“yes”下，aof重写/文件同步等特性才会生效
appendonly yes
##指定aof文件名称
appendfilename appendonly.aof
##指定aof操作中文件同步策略，有三个合法值：always everysec no,默认为everysec
appendfsync everysec
##在aof-rewrite期间，appendfsync是否暂缓文件同步，"no"表示“不暂缓”，“yes”表示“暂缓”，默认为“no”
no-appendfsync-on-rewrite no
##aof文件rewrite触发的最小文件尺寸(mb,gb),只有大于此aof文件大于此尺寸是才会触发rewrite，默认“64mb”，建议“512mb”
auto-aof-rewrite-min-size 64mb
##相对于“上一次”rewrite，本次rewrite触发时aof文件应该增长的百分比。
##每一次rewrite之后，redis都会记录下此时“新aof”文件的大小(例如A)，那么当aof文件增长到A*(1 + p)之后
##触发下一次rewrite，每一次aof记录的添加，都会检测当前aof文件的尺寸。
auto-aof-rewrite-percentage 100

AOF 是文件操作，对于变更操作比较密集的 server ，那么必将造成磁盘 IO 的负荷加重；此外 linux 对文件操作采取了“ 延迟写入 ” 手段，即并非每次 write 操作都会触发实际磁盘操作，而是进入了 buffffer 中，当 buffffer 数据达到阀值时触发实际写入( 也有其他时机 ) ，这是 linux 对文件系统的优化，但是这却有可能带来隐患，如果 buffffer 没有刷新到磁盘，此时物理机器失效( 比如断电 ) ，那么有可能导致最后一条或者多条 aof 记录的丢失。通过上述配置文件，可以得知 redis 提供了 3 中 aof 记录同步选项：

always ：每一条 aof 记录都立即同步到文件，这是最安全的方式，也以为更多的磁盘操作和阻塞延迟，是 IO 开支较大。

everysec ：每秒同步一次，性能和安全都比较中庸的方式，也是 redis 推荐的方式。如果遇到物理服务器故障，有可能导致最近一秒内 aof 记录丢失 ( 可能为部分丢失 ) 。

no ： redis 并不直接调用文件同步，而是交给操作系统来处理，操作系统可以根据 buffffer 填充情况 / 通道空闲时间等择机触发同步；这是一种普通的文件操作方式。性能较好，在物理服务器故障时，数据丢失量会因 OS 配置有关。

其实，我们可以选择的太少， everysec 是最佳的选择。如果你非常在意每个数据都极其可靠，建议你选择一款 “ 关系性数据库” 。

10.2.3 AOF与RDB区别

(1) RDB:

RDB 是在某个时间点将数据写入一个临时文件，持久化结束后，用这个临时文件替换上次持久化的文件，达到数据恢复

优点：使用单独子进程来进行持久化，主进程不会进行任何 IO 操作，保证了 redis 的高性能

缺点： RDB 是间隔一段时间进行持久化，如果持久化之间 redis 发生故障，会发生数据丢失。所以这种方式更适合数据要求不严谨的时候

(2) AOF：

Append-only fifile ，将 “ 操作 + 数据 ” 以格式化指令的方式追加到操作日志文件的尾部，在 append 操作返回后 ( 已经写入到文件或者即将写入) ，才进行实际的数据变更， “ 日志文件 ” 保存了历史所有的操作过程；当 server 需要数据恢复时，可以直接replay 此日志文件，即可还原所有的操作过程。 AOF 相对可靠，它和 mysql 中 bin.log 、 apache.log 、 zookeeper中 txn-log 简直异曲同工。 AOF 文件内容是字符串，非常容易阅读和解析。

优点：可以保持更高的数据完整性，如果设置追加 fifile 的时间是 1s ，如果 redis 发生故障，最多会丢失 1s 的数据；且如果日志写入不完整支持redis-check-aof 来进行日志修复； AOF 文件没被 rewrite 之前（文件过大时会对命令进行合并重写），可以删除其中的某些命令（比如误操作的flflushall ）。

缺点： AOF 文件比 RDB 文件大，且恢复速度慢。

十一、Redis主从复制

持久化保证了即使 redis 服务重启也不会丢失数据，但是当 redis 服务器的硬盘损坏了可能会导致数据丢失，通过redis 的主从复制机制就可以避免这种单点故障（单台服务器的故障）。

主 redis 中的数据和从上的数据保持实时同步 , 当主 redis 写入数据时通过主从复制机制复制到两个从服务上。

主从复制不会阻塞 master ，在同步数据时， master 可以继续处理 client 请求 .

主机 master 配置 : 无需配置

推荐主从模式同步数据 :

工作中一般选用：一主两从或一主一从

数据会同步到从服务器。在这个集群中的几台服务器上都有同样的数据。

11.1 主从搭建步骤:

主机：不用配置。仅仅只需要配置从机 , 从机 slave 配置 :( 这里是伪集群 )

(1) 第一步：复制出一个从机,注意使用root用户

[root@localhost myapps]# cp redis/ redis1 -r
[root@localhost myapps]# ll
总用量 40
drwxr-xr-x. 3 root root 4096 2月 1 09:26 redis
drwxr-xr-x. 3 root root 4096 2月 1 09:27 redis1

(2) 第二步：修改从机的redis.conf

语法： replicaof // replicaof 主机 ip 主机端口号

提示 : 检索文件 : 输入 :/replicaof 当前页没有时，输入 n ，查找下一页

(3) 第三步：修改从机的port地址为6380

在从机 redis.conf 中修改

(4) 第四步：清除从机中的持久化文件

[root@localhost bin]# rm -rf appendonly.aof dump.rdb
[root@localhost bin]# ll
总用量 15440
-rwxr-xr-x. 1 root root 4588902 7月 1 09:27 redis-benchmark
-rwxr-xr-x. 1 root root 22225 7月 1 09:27 redis-check-aof
-rwxr-xr-x. 1 root root 45443 7月 1 09:27 redis-check-dump
-rwxr-xr-x. 1 root root 4691809 7月 1 09:27 redis-cli
lrwxrwxrwx. 1 root root 12 7月 1 09:27 redis-sentinel -> redis-server
-rwxr-xr-x. 1 root root 6450337 7月 1 09:27 redis-server

(5) 第五步：启动从机

[root@localhost redis1]# ./bin/redis-server ./redis.conf

(6) 第六步：启动6380的客户端

[root@localhost redis1]# ./bin/redis-cli -p 6380
127.0.0.1:6380> keys *
1) "mylist"
2) "num"
3) "bookCate1"
4) "newbook"
127.0.0.1:6380>

停止客户端 : ./bin/redis-cli -p 6380 shutdown

注意：

1.主机一旦发生增删改操作，那么从机会自动将数据同步到从机中

2. 从机不能执行写操作 , 只能读

127.0.0.1:6380> get username
"hehe"
127.0.0.1:6380> set username haha
(error) READONLY You can't write against a read only slave.

(7) 复制的过程原理

1.当从库和主库建立 MS(master slaver) 关系后，会向主数据库发送 SYNC 命令；

2.主库接收到 SYNC 命令后会开始在后台保存快照（ RDB 持久化过程），并将期间接收到的写命令缓存起来；

3.快照完成后 , 主 Redis 会将快照文件和所有缓存的写命令发送给从 Redis ；

4.从 Redis 接收到后，会载入快照文件并且执行收到的缓存命令；

5.主 Redis 每当接收到写命令时就会将命令发送从 Redis ，保证数据的一致；【内部完成 , 所以 不支持客户端在从 机人为写数据 。】

(8) 复制架构中出现宕机情况?

从Redis宕机 : 重启就好

主 Redis 宕机 : 从数据库 ( 从机 ) 中执行 SLAVEOF NO ONE 命令，断开主从关系并且提升为主库继续服务 [ 把一个从做为主机，这个时候新主机[ 之前的从机 ] 就具备写入的能力 ] ；主服务器修好后，重新启动后，执行 SLAVEOF 命令，将其设置为从库[ 老主机设置为从机 ] 。 [ 手动执行，过程复杂，容易出错。 ] 是否有更好的方案？

十二、Redis哨兵模式

哨兵模式 ：给集群分配一个站岗的。

哨兵的作用就是对 Redis 系统的运行情况监控，它是一个独立进程 , 它的功能：

1. 监控主数据库和从数据库是否运行正常；

2. 主数据出现故障后自动将从数据库转化为主数据库；

如果主机宕，开启选举工作，选择一个从做主机。

环境准备：一主两从，启动任一从机时，启动哨兵模式

虽然哨兵 (sentinel) 释出为一个单独的可执行文件 redis-sentinel , 但实际上它只是一个运行在特殊模式下的 Redis 服务器，你可以在启动一个普通 Redis 服务器时通过给定 --sentinel 选项来启动哨兵 (sentinel) 。

12.1 第一步：配置哨兵：

哨兵主要是用来监听主服务器的，所以一般把哨兵部署在从服务器上监听。

配置哨兵：

1. 启动哨兵进程，首先需要创建哨兵配置文件 vi sentinel.conf, 可从源码配置 redis5.0.5/sentinel.conf 中复制内容，也可以直接自定义该文件到bin 目录下

2. 在配置中输入 :sentinel monitor mastername 内网 IP(127.0.0.1) 6379 1

3. 说明：

4. mastername 监控主数据的名称，自定义

5. 127.0.0.1 ：监控主数据库的 IP;

6. 6379: 端口

7. 1 ：最低通过票数

12.2 第二步：启动哨兵：

哨兵是一个单独的进程，启动之前确保主从服务是正常的。先启动主服务，后启动从服务

把日志写入指定的文件

[root@localhost bin]# ./redis-sentinel ./sentinel.conf >sent.log &
[1] 3373

启动 redis 服务后，程序会自动配置文件 sentinel.conf ，并生成内容，注意 : 若再起启动需要删除下生成的内容。

哨兵启动方式 :

[root@localhost bin]# ./redis-server sentinel.conf --sentinel

哨兵进程控制台：为 master 数据库添加了一个监控 .

同时多了哨兵进程：

查询配置文件sentinel.conf中生成的内容：

启动哨兵的时候，修改了哨兵的配置文件。如果需要再次启动哨兵，需要删除 myid 唯一标示。

（保险的做法就是启动的一次，新配置一次）

12.3 第三步：主机宕机

机房意外：断电了。硬件故障：硬盘坏了。

列表：

杀死主机：kill -9 pid

[root@localhost redis6380]# kill -9 3342

哨兵控制台：从库自动提升为主库。

哨兵工作 , 链接之前的从机确认：

127.0.0.1:6381> info replication
# Replication
role:slave
master_host:192.168.197.129
master_port:6379

哨兵替代运维。自动监控完成。

同时也会自动修改 redis.conf 的主从配置文件。

replicaof 127.0.0.1 6380

指向了新主机。再次启动原有的主机 , 原有的主机会变为从机。

总结 :

主从集群：主机有写入权限。从机没有，只有可读。

意外宕机方案 :

手动恢复：人为重启服务器，主机宕，把从机设置为主机。

自动恢复：使用哨兵监控。自动切换主从。

十三、Redis集群方案

13.1 redis-cluster架构图

架构细节:

(1) 所有的 redis 节点彼此互联 (PING-PONG 机制 ), 内部使用二进制协议优化传输速度和带宽 .

(2) 节点的 fail 是通过集群中超过半数的节点检测有效时整个集群才生效 .

(3) 客户端与 redis 节点直连 , 不需要中间 proxy 层 . 客户端不需要连接集群所有节点 , 连接集群中任何一个可用节点即可

(4)redis-cluster 把所有的物理节点映射到 [ 0-16383 ]slot 上 ,cluster 负责维护 node<->slot<- >value

Redis 集群中内置了 16384 个哈希槽，当需要在 Redis 集群中放置一个 key-value 时， redis 先对 key 使用 crc16 算法算出一个结果，然后把结果对 16384 求余数，这样每个 key 都会对应一个编号在 0-16383 之间的哈希槽， redis 会根据节点数量大致均等的将哈希槽映射到不同的节点

示例如下：

13.2 redis-cluster投票:容错

心跳机制

(1) 集群中所有 master 参与投票 , 如果半数以上 master 节点与其中一个 master 节点通信超过 (cluster-node-timeout), 认为该master 节点挂掉 .

(2): 什么时候整个集群不可用 (cluster_state:fail)?

a: 如果集群任意 master 挂掉 , 且当前 master 没有 slave ，则集群进入 fail 状态。也可以理解成集群的 [0-16383]slot 映射不完全时进入 fail 状态。

b:如果集群超过半数以上 master 挂掉，无论是否有 slave ，集群进入 fail 状态。

13.3 集群搭建步骤

(1) 第一步:安装redis

(2) 第二步:创建集群目录

[root@localhost redis]# mkdir redis-cluster

(3) 第三步:在集群目录下创建节点目录

搭建集群最少也得需要 3 台主机，如果每台主机再配置一台从机的话，则最少需要 6 台机器。设计端口如下：创建 6 个redis 实例，需要端口号 7001~7006

[root@localhost myapps]# cp redis/ redis-cluster/7001 -r
[root@localhost myapps]# cd redis-cluster/7001
[root@localhost 7001]# ll
drwxr-xr-x. 2 root root 4096 7月 1 10:22 bin
-rw-r--r--. 1 root root 3446 7月 1 10:22 dump.rdb
-rw-r--r--. 1 root root 41404 7月 1 10:22 redis.conf

(4) 第四步：如果存在持久化文件，则删除

[root@localhost 7001]# rm -rf appendonly.aof dump.rdb

(5) 第五步：修改redis.conf配置文件，打开Cluster-enable yes

说明： cluster-enable 是否支持集群

(6) 第六步：修改端口

(7) 第七步：复制出7002-7006机器

[root@localhost redis-cluster]# cp 7001/ 7002 -r
[root@localhost redis-cluster]# cp 7001/ 7003 -r
[root@localhost redis-cluster]# cp 7001/ 7004 -r
[root@localhost redis-cluster]# cp 7001/ 7005 -r
[root@localhost redis-cluster]# cp 7001/ 7006 -r
[root@localhost redis-cluster]# ll
total 28
drwxr-xr-x. 3 root root 4096 Jun 2 00:02 7001
drwxr-xr-x. 3 root root 4096 Jun 2 00:02 7002
drwxr-xr-x. 3 root root 4096 Jun 2 00:02 7003
drwxr-xr-x. 3 root root 4096 Jun 2 00:03 7004
drwxr-xr-x. 3 root root 4096 Jun 2 00:03 7005
drwxr-xr-x. 3 root root 4096 Jun 2 00:03 7006
-rwxr-xr-x. 1 root root 3600 Jun 1 23:52 redis-trib.rb

(8) 第八步：修改7002-7006机器的端口

(9) 第九步：启动7001-7006这六台机器，写一个启动脚本：自定义shel脚本

cd 7001
./bin/redis-server ./redis.conf
cd ..
cd 7002
./bin/redis-server ./redis.conf
cd ..
cd 7003
./bin/redis-server ./redis.conf
cd ..
cd 7004
./bin/redis-server ./redis.conf
cd ..
cd 7005
./bin/redis-server ./redis.conf
cd ..
cd 7006
./bin/redis-server ./redis.conf
cd ..

(10) 第十步：修改start-all.sh文件的权限

[root@localhost redis-cluster]# chmod u+x startall.sh

(11) 第十一步：启动所有的实例

[root@localhost redis-cluster]# ./startall.sh

(12) 第十二步：创建集群（关闭防火墙）

注意：在任意一台上运行不要在每台机器上都运行，一台就够了 redis 5.0.5 中使用 redis-cli --cluster 替代 redis-trib.rb,命令如下

redis-cli --cluster create ip:port ip:port --cluster-replicas 1

[root@localhost redis_cluster]# cd /home/admin/myapps/redis-cluster/7001/bin
[root@localhost bin]# ./redis-cli --cluster create 192.168.197.132:7001 192.168.197.132:7002
192.168.197.132:7003 192.168.197.132:7004 192.168.197.132:7005 192.168.197.132:7006 --clusterreplicas 1
\>>> Creating cluster
Connecting to node 127.0.0.1:7001: OK
Connecting to node 127.0.0.1:7002: OK
Connecting to node 127.0.0.1:7003: OK
Connecting to node 127.0.0.1:7004: OK
Connecting to node 127.0.0.1:7005: OK
Connecting to node 127.0.0.1:7006: OK
\>>> Performing hash slots allocation on 6 nodes...
Using 3 masters:
127.0.0.1:7001
127.0.0.1:7002
127.0.0.1:7003
Adding replica 127.0.0.1:7004 to 127.0.0.1:7001
Adding replica 127.0.0.1:7005 to 127.0.0.1:7002
Adding replica 127.0.0.1:7006 to 127.0.0.1:7003
[OK] All 16384 slots covered.

13.4 连接集群

命令 :

[root@localhost 7001]# ./bin/redis-cli -h 127.0.0.1 -p 7001 -c

-c ：指定是集群连接

[root@localhost 7001]# ./bin/redis-cli -h 127.0.0.1 -p 7001 -c
127.0.0.1:7001> set username java123
-> Redirected to slot [14315] located at 127.0.0.1:7003
OK

关闭防火墙 :service iptables stop

查看防火墙状态 :service iptables status

13.5 查看集群信息

127.0.0.1:7003> cluster info
cluster_state:ok
cluster_slots_assigned:16384
cluster_slots_ok:16384
cluster_slots_pfail:0
cluster_slots_fail:0
cluster_known_nodes:6
cluster_size:3
cluster_current_epoch:6
cluster_my_epoch:3
cluster_stats_messages_sent:1186
cluster_stats_messages_received:1186

13.6 查看集群中节点信息

127.0.0.1:7003> cluster nodes
713218b88321e5067fd8ad25c3bf7db88c878ccf 127.0.0.1:7003 myself,master - 0 0 3 connected 10923-
16383
e7fb45e74f828b53ccd8b335f3ed587aa115b903 127.0.0.1:7001 master - 0 1498877677276 1 connected 0-
5460
b1183545245b3a710a95d669d7bbcbb5e09896a0 127.0.0.1:7006 slave
713218b88321e5067fd8ad25c3bf7db88c878ccf 0 1498877679294 3 connected
8879c2ed9c141de70cb7d5fcb7d690ed8a200792 127.0.0.1:7005 slave
4a312b6fc90bfee187d43588ead99d83b407c892 0 1498877678285 5 connected
4a312b6fc90bfee187d43588ead99d83b407c892 127.0.0.1:7002 master - 0 1498877674248 2 connected
5461-10922
4f8c7455574e2f0aab1e2bb341eae319ac065039 127.0.0.1:7004 slave
e7fb45e74f828b53ccd8b335f3ed587aa115b903 0 1498877680308 4 connected

13.7 Jedis连接集群

13.7.1 关闭防火墙

注意 : 如果 redis 重启，需要将 redis 中生成的 dump.rdb 和 nodes.conf 文件删除，然后再重启。

13.7.2 代码实现

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

注意 jedis 的版本，其他版本有可能报错 :java.lang.NumberFormatException: For input string: "7002@17002"

public static void main(String[] args) throws IOException {
    // 创建一连接，JedisCluster对象,在系统中是单例存在
    Set<HostAndPort> nodes = new HashSet<HostAndPort>();
    nodes.add(new HostAndPort("192.168.197.132", 7001));
    nodes.add(new HostAndPort("192.168.197.132", 7002));
    nodes.add(new HostAndPort("192.168.197.132", 7003));
    nodes.add(new HostAndPort("192.168.197.132", 7004));
    nodes.add(new HostAndPort("192.168.197.132", 7005));
    nodes.add(new HostAndPort("192.168.197.132", 7006));
    JedisCluster cluster = new JedisCluster(nodes);
    // 执行JedisCluster对象中的方法，方法和redis指令一一对应。
    cluster.set("test1", "test111");
    String result = cluster.get("test1");
    System.out.println(result);
    //存储List数据到列表中
    cluster.lpush("site-list", "java");
    cluster.lpush("site-list", "c");
    cluster.lpush("site-list", "mysql");
    // 获取存储的数据并输出
    List<String> list = cluster.lrange("site-list", 0 ,2);
    for(int i=0; i<list.size(); i++) {
        System.out.println("列表项为: "+list.get(i));
    }
    // 程序结束时需要关闭JedisCluster对象
    cluster.close();
    System.out.println("集群测试成功！");
}

十四、Redis高端面试-缓存穿透，缓存击穿，缓存雪崩问题

14.1 缓存的概念

什么是缓存 ?

广义的缓存就是在第一次加载某些可能会复用数据的时候，在加载数据的同时，将数据放到一个指定的地点做保存。再下次加载的时候，从这个指定地点去取数据。这里加缓存是有一个前提的，就是从这个地方取数据，比从数据源取数据要快的多。

java 狭义一些的缓存，主要是指三大类

1. 虚拟机缓存（ehcache， JBoss Cache ）

2. 分布式缓存（redis， memcache ）

3. 数据库缓存

正常来说，速度由上到下依次减慢

缓存取值图 :

14.2 缓存雪崩

缓存雪崩产生的原因 ：

缓存雪崩通俗简单的理解就是：由于原有缓存失效（或者数据未加载到缓存中），新缓存未到期间（缓存正常从 Redis中获取，如下图）所有原本应该访问缓存的请求都去查询数据库了，而对数据库 CPU 和内存造成巨大压力，严重的会造成数据库宕机，造成系统的崩溃。

缓存失效的时候如下图：

解决方案:

1 ：在缓存失效后，通过加锁或者队列来控制读数据库写缓存的线程数量。比如对某个 key 只允许一个线程查询数据和写缓存，其他线程等待。虽然能够在一定的程度上缓解了数据库的压力但是与此同时又降低了系统的吞吐量。

public Users getByUsers(Long id) {
    // 1.先查询redis
    String key = this.getClass().getName() + "-" + 
    Thread.currentThread().getStackTrace([1].getMethodName()+ "-id:" + id;
    String userJson = redisService.getString(key);
    if (!StringUtils.isEmpty(userJson)) {
        Users users = JSONObject.parseObject(userJson, Users.class);
        return users;
    }
    Users user = null;
    try {
        lock.lock();
        // 查询db
        user = userMapper.getUser(id);
        redisService.setSet(key, JSONObject.toJSONString(user));
    } catch (Exception e) {
    } finally {
        lock.unlock(); // 释放锁
    }
    return user;
}

注意 : 加锁排队只是为了减轻数据库的压力，并没有提高系统吞吐量。假设在高并发下，缓存重建期间 key 是锁着的，这是过来1000 个请求 999 个都在阻塞的。同样会导致用户等待超时，这是个治标不治本的方法。

2 ：分析用户的行为，不同的 key ，设置不同的过期时间，让缓存失效的时间点尽量均匀。

14.3 缓存穿透

缓存穿透是指用户查询数据，在数据库没有，自然在缓存中也不会有。这样就导致用户查询的时候，在缓存中找不到，每次都要去数据库再查询一遍，然后返回空。这样请求就绕过缓存直接查数据库，这也是经常提的缓存命中率问题。

解决方案 :

1.如果查询数据库也为空，直接设置一个默认值存放到缓存，这样第二次到缓冲中获取就有值了，而不会继续访问数据库，这种办法最简单粗暴。

2.把空结果，也给缓存起来，这样下次同样的请求就可以直接返回空了，既可以避免当查询的值为空时引起的缓存穿透。同时也可以单独设置个缓存区域存储空值，对要查询的key 进行预先校验，然后再放行给后面的正常缓存处理逻辑。

public String getByUsers2(Long id) {
    // 1.先查询redis
    String key = this.getClass().getName() + "-" + Thread.currentThread().getStackTrace()
    [1].getMethodName()+ "-id:" + id;
    String userName = redisService.getString(key);
    if (!StringUtils.isEmpty(userName)) {
        return userName;
    }
    System.out.println("######开始发送数据库DB请求########");
    Users user = userMapper.getUser(id);
    String value = null;
    if (user == null) {
        // 标识为null
        value = "";
    } else {
        value = user.getName();
    }
    redisService.setString(key, value);
    return value;
}

注意：再给对应的 ip 存放真值的时候，需要先清除对应的之前的空缓存。

14.4 缓存击穿

对于一些设置了过期时间的 key ，如果这些 key 可能会在某些时间点被超高并发地访问，是一种非常 “ 热点 ” 的数据。这个时候，需要考虑一个问题：缓存被“ 击穿 ” 的问题，这个和缓存雪崩的区别在于这里针对某一 key 缓存，前者则是很多key 。

热点 key:

某个 key 访问非常频繁，当 key 失效的时候有大量线程来构建缓存，导致负载增加，系统崩溃。

解决办法：

①使用锁，单机用 synchronized,lock 等，分布式用分布式锁。

②缓存过期时间不设置，而是设置在 key 对应的 value 里。如果检测到存的时间超过过期时间则异步更新缓存。

十五、Redis高端面试-分布式锁

15.1 使用分布式锁要满足的几个条件：

1. 系统是一个分布式系统（关键是分布式，单机的可以使用 ReentrantLock 或者 synchronized 代码块来实现）

2. 共享资源（各个系统访问同一个资源，资源的载体可能是传统关系型数据库或者 NoSQL ）

3. 同步访问（即有很多个进程同时访问同一个共享资源。）

15.2 什么是分布式锁？

线程锁：主要用来给方法、代码块加锁。当某个方法或代码使用锁，在同一时刻仅有一个线程执行该方法或该代码段。线程锁只在同一JVM 中有效果，因为线程锁的实现在根本上是依靠线程之间共享内存实现的，比如 synchronized是共享对象头，显示锁 Lock 是共享某个变量（state）。

进程锁：为了控制同一操作系统中多个进程访问某个共享资源，因为进程具有独立性，各个进程无法访问其他进程的资源，因此无法通过synchronized 等线程锁实现进程锁。

分布式锁：当多个进程不在同一个系统中，用分布式锁控制多个进程对资源的访问。

15.3 应用的场景

线程间并发问题和进程间并发问题都是可以通过分布式锁解决的，但是强烈不建议这样做！因为采用分布式锁解决这些小问题是非常消耗资源的！分布式锁应该用来解决分布式情况下的多进程并发问题才是最合适的。

有这样一个情境，线程 A 和线程 B 都共享某个变量 X 。

如果是单机情况下（单 JVM ），线程之间共享内存，只要使用线程锁就可以解决并发问题

如果是分布式情况下（多 JVM ），线程 A 和线程 B 很可能不是在同一 JVM 中，这样线程锁就无法起到作用了，这时候就要用到分布式锁来解决。

分布式锁可以基于很多种方式实现，比如 zookeeper 、 redis... 。不管哪种方式，他的基本原理是不变的：用一个状态值表示锁，对锁的占用和释放通过状态值来标识。

这里主要讲如何用 redis 实现分布式锁。

15.4 使用redis的setNX命令实现分布式锁

15.4.1 实现的原理：

Redis 为单进程单线程模式，采用队列模式将并发访问变成串行访问，且多客户端对 Redis 的连接并不存在竞争关系。redis 的 SETNX 命令可以方便的实现分布式锁。

15.4.2 基本命令解析

(1) setNX（SET if Not eXists）

语法：
SETNX key value
将 key 的值设为 value ，当且仅当 key 不存在。

若给定的 key 已经存在，则 SETNX 不做任何动作。

SETNX 是『SET if Not eXists』 ( 如果不存在，则 SET) 的简写

返回值：

设置成功，返回 1 。

设置失败，返回 0 。

redis> EXISTS job # job 不存在
(integer) 0
redis> SETNX job "programmer" # job 设置成功
(integer) 1
redis> SETNX job "code-farmer" # 尝试覆盖 job ，失败
(integer) 0
redis> GET job # 没有被覆盖
"programmer"

所以我们使用执行下面的命令 SETNX 可以用作加锁原语 (locking primitive) 。比如说，要对关键字 (key) foo 加锁，客户端可以尝试以下方式：

SETNX lock.foo <current Unix time + lock timeout + 1>

如果 SETNX 返回 1 ，说明客户端已经获得了锁， SETNX 将键 lock.foo 的值设置为锁的超时时间（当前时间 + 锁的有效时间）。之后客户端可以通过 DEL lock.foo 来释放锁。

如果 SETNX 返回 0 ，说明 key 已经被其他客户端上锁了。如果锁是非阻塞 (non blocking lock) 的，我们可以选择返回调用，或者进入一个重试循环，直到成功获得锁或重试超时(timeout) 。

(2) getSET

先获取 key 对应的 value 值。若不存在则返回 nil ，然后将旧的 value 更新为新的 value 。

语法：

GETSET key value

将给定 key 的值设为 value ，并返回 key 的旧值 (old value) 。

当 key 存在但不是字符串类型时，返回一个错误。

返回值：

返回给定 key 的旧值 [ 之前的值 ] 。

当 key 没有旧值时，也即是， key 不存在时，返回 nil 。

注意的关键点：(回答面试的核心点)

1 、同一时刻只能有一个进程获取到锁。 setnx

2 、释放锁：锁信息必须是会过期超时的，不能让一个线程长期占有一个锁而导致死锁；

（最简单的方式就是 del ，如果在删除之前死锁了。）

ex:

53秒设置--58秒到期

当前时间为56秒，没有过期

当前时间为 59 秒，过期 . （当前时间大于设置的时间）

死锁情况是在判断超时后，直接操作业务，设置过期时间，执行业务，然后删除释放锁。其他进程再次通过 setnx 来抢锁。

解决死锁：

上面的锁定逻辑有一个问题： 如果一个持有锁的客户端失败或崩溃了不能释放锁，该怎么解决 ？

我们可以通过锁的键对应的时间戳来判断这种情况是否发生了，如果当前的时间已经大于 lock.foo 的值，说明该锁已失效，可以被重新使用。

发生这种情况时，可不能简单的通过 DEL 来删除锁，然后再 SETNX 一次（讲道理， 删除锁的操作应该是锁拥有 者执行的，这里只需要等它超时即可 ），当多个客户端检测到锁超时后都会尝试去释放它，这里就可能出现一个竞态条件, 让我们模拟一下这个场景：

C0 操作超时了，但它还持有着锁， C1 和 C2 读取 lock.foo 检查时间戳，先后发现超时了。 C1 发送 DEL lock.foo C1 发送SETNX lock.foo 并且成功了。 C2 发送 DEL lock.foo C2 发送 SETNX lock.foo 并且成功了。这样一来， C1 ， C2 都拿到了锁！问题大了！

幸好这种问题是可以避免的，让我们来看看C3这个客户端是怎样做的：

C3发送 SETNX lock.foo 想要获得锁，由于 C0 还持有锁，所以 Redis 返回给 C3 一个 0 C3 发送 GET lock.foo 以检查锁是否超时了，

如果没超时，则等待或重试。反之，如果已超时，C3 通过下面的操作来尝试获得锁： GETSET lock.foo 通过 GETSET， C3 拿到的时间戳如果仍然是超时的，那就说明， C3 如愿以偿拿到锁了。如果在 C3 之前，有个叫 C4 的客户端比C3 快一步执行了上面的操作，那么 C3 拿到的时间戳是个未超时的值，这时， C3 没有如期获得锁，需要再次等待或重试。留意一下，尽管C3 没拿到锁，但它改写了 C4 设置的锁的超时值，不过这一点非常微小的误差带来的影响可以忽略不计。

注意：为了让分布式锁的算法更稳键些，持有锁的客户端在解锁之前应该再检查一次自己的锁是否已经超时，再去做DEL 操作，因为可能客户端因为某个耗时的操作而挂起，操作完的时候锁因为超时已经被别人获得，这时就不必解锁了。

伪代码 :

public static boolean lock(String lockName) {
    Jedis jedis = RedisPool.getJedis();
    //lockName可以为共享变量名，也可以为方法名，主要是用于模拟锁信息
    System.out.println(Thread.currentThread() + "开始尝试加锁！");
    Long result = jedis.setnx(lockName, String.valueOf(System.currentTimeMillis() + 5000));
    if (result != null && result.intValue() == 1){
        System.out.println(Thread.currentThread() + "加锁成功！");
        jedis.expire(lockName, 5);
        System.out.println(Thread.currentThread() + "执行业务逻辑！");
        jedis.del(lockName);
        return true;
    } else {//判断是否死锁
        String lockValueA = jedis.get(lockName);
        //得到锁的过期时间，判断小于当前时间，说明已超时但是没释放锁，通过下面的操作来尝试获得锁。下面逻辑防止死锁[已经过期但是没有释放锁的情况]
        if (lockValueA != null && Long.parseLong(lockValueA) <                     
            System.currentTimeMillis()){
            String lockValueB = jedis.getSet(lockName,
            String.valueOf(System.currentTimeMillis() + 5000));
            //这里返回的值是旧值，如果有的话。之前没有值就返回null,设置的是新超时。
            if (lockValueB == null || lockValueB.equals(lockValueA)){
                System.out.println(Thread.currentThread() + "加锁成功！");
                jedis.expire(lockName, 5);
                System.out.println(Thread.currentThread() + "执行业务逻辑！");
                jedis.del(lockName);
                return true;
            } else {
                return false;
            }
        } else {
            return false;
        }
    }
}