Redis学习笔记（常用数据类型，发布订阅，事务和锁机制，持久化，集群，雪崩，缓存击穿，分布式锁）

一、NoSQL数据库简介

解决扩展性问题，如果需要对功能进行改变（比如增删功能），用框架有一定的规范要求，无形中解决了扩展性问题。

Redis是一种典型的NoSQL数据库。

NoSQL的基础作用：

1. nginx负载均衡反向代理多台服务器，会出现session问题（代理到不同服务器，session存放分散）：解决方式1：存储到客户端的Cookie中（安全性无法保证）。解决方式2：session复制（数据冗余）。解决方式3：用户信息存储到Nosql数据库。不需要I/O操作，直接存到内存中。所以NoSQL可以减少CPU、I/O的压力，直接通过内存进行存取。

2. 当数据量变大，数据库的读写效率会降低，一般的操作是进行水平切分、垂直切分、读写分离，但这样的操作会破坏业务逻辑来换取性能。但现在可以通过设置缓存数据库（频繁查询的数据可以放入）来减少io的读操作，提高访问效率。

NoSQL（Not Only SQL）: 泛指非关系型的数据库，不依赖业务逻辑方式存储，以简单的key-value模式存储，大大增加了数据库的扩展能力。

NoSQL特点：1. 不遵循SQL标准。2.不支持ACID（原子性，一致性，隔离性，持久性），但不是不支持事务。3.远超于SQL的性能。

NoSQL适用场景：1. 对数据高并发的读写（秒杀场景）。2. 海量数据的读写。3.对数据高可扩展性的。

NoSQL不适用场景：1. 需要事务支持。2.基于sql的结构化查询存储，处理复杂的关系，需要即席查询。3. 用不着sql的和用了sql也不行的情况，考虑用NoSQL。

不持久化：关电脑关机了，内存中的数据就不存在了。支持持久化：将数据存入硬板中。

二、Redis概述和安装

redis简介：

redis多样的数据结构存储持久化数据：

redis特点：1. 端口号6379。2. 有16个数据库，默认是用0号库。3. redis支持多数据类型，支持持久化，支持单线程+多路IO复用技术。

下载：

第1步：先从官网redis.io上下载redis-7.2.1.tar.gz。然后打开linux虚拟机，用xshell连接，选择一个目录，拖入压缩文件（我的地址：home - user - opt）。

第2步：检查环境：先gcc --version看有没有c语言的环境，如果没有则输入yum install gcc自动安装。

第3步：解压文件：tar -zxvf r然后按Tag键补全，然后cd进入解压后的文件，输入make进行编译。最后要输入sudo make install进行安装，这一步很重要。

第4步：首先进入/usr/local/bin目录查看，文件会被默认安装在该目录下，出现如下文件表示成功。

第5步：首先进入redis文件夹，cd /home/user/opt/redis-7.2.1，然后将文件复制到etc下，cp redis.conf /etc/redis.conf ，（cp是复制，将redis.conf文件复制到etc目录下，命名为redis.conf），最后先sudo su进入root身份，再进入etc文件夹cd /etc/，在etc下vi redis.conf，将darmonize no 改为 yes

第5步：后台启动 cd /usr/local/bin，输入：redis-server /etc/redis.conf 即可启动服务器。

注意：1、redis-server后面有一个空格。2、如果提示“bash：redis-server：未找到命令...”则输入 ln -s /usr/local/bin/redis-server /usr/bin/redis-server 相当于让后面的路径可以引用前面路径的文件。3、如果提示如下内容：WARNING Memory overcommit must be enabled!先输入：echo 1 > /proc/sys/vm/overcommit_memory 将内存超额提交机制设置为始终允许分配请求，即启用内存超额提交。然后redis-server /etc/redis.conf进行重启。

— — — — —知识加油站 — — — — —

ln是 Linux 和 Unix 操作系统中的一个命令，用于创建链接（links），包括硬链接和符号链接（symbolic link）。ln -s 是一个选项，表示创建的链接是一个符号链接（symbolic link）。

符号链接是一种特殊类型的文件，它只包含了对另一个文件或目录的引用，而不是实际的数据。通过符号链接，可以在不改变实际文件位置的情况下，在不同的位置引用同一个文件或目录。

— — — — — — — — — — — — — —

第6步：检验。输入ps -ef | grep redis 查看当前进程。先输入ln -s /usr/local/bin/redis-cli /usr/bin/redis-cli，然后用客户端访问输入redis-cli，然后输入ping如果返回PONG表示连通成功。

第7步：退出redis：可以直接在redis-cli下用shutdown退出。也可以用ps -ef | grep redis找到进程，用kill -9 端口号，杀死进程。

三、常用五大数据类型

3.0 key键操作

首先要按照前面的方式启动服务器，然后输入代码：/usr/local/bin/redis-cli，连接上redis：

命令1：keys *

命令2：set key键名 value值（如set k1 rucy）

命令3：type key键名，查看key的类型

命令4：exists key键名，判断是否存在key，返回1为有，返回0为无

命令5：del key键名，删除key。unlink key 根据value选择非阻塞删除（异步删除，后台慢慢删）。

命令6：expire key键名时间秒，

命令7：ttl key键名，查看键会否过期，1是还未过期，-2为过期，-1永不过期

命令8：select 数字，命令切换数据库。dbsize查看当前库里有多少key。flushdb清空当前库。flushall清空所有。

3.1 String字符串

String是redis最基本的类型，一个key对应一个Value。String类型是二进制安全的，意味着redis的string可以包含任何数据。比如jpg文件或序列化的对象。一个redis中字符串value最多可以是512M。

String的数据结构为简单动态字符串（Simple Dynamic String，缩写SDS）。是可以修改的字符串，内部结构实现上类似于Java的ArrayList，采用预分配冗余空间的方式来减少内存的频繁分配。

如下图所示，内部为当前字符串实际分配的空间capacity一般要高于实际字符串的长度len，当字符串长度小于1M时，扩容都是加倍现有的空间，如果超过1M，扩容时只会多扩1M的空间。需要注意的是字符串最大长度为512M。

命令1：set key键名 value值，可以加入一个键值对。

命令2：get 键名，可以获取key里的值。

命令3：append key名追加的内容，返回的是长度

命令4：strlen key名，可以返回长度

命令5：setnx key名 value值，只有当key值不存在，在设置值，当返回0表示不能设置成功。

命令6：incr key名，增加1

命令7：decr key名，减少1

命令8：incrby key名步长，在数字键的value上增加步长的值。decrby key名步长，减少步长的值。

原子性操作：不会被线程调度机制打断的操作。这种操作一旦开始，就一直运行到结束，中间不会切换到另外一个线程。Redis是单线程。

命令9：mset key1 value1 key2 value2 ，批量设置键和值

命令10：mget key1 key2 key3，批量获取键的值

命令11：msetnx key1 value1 key2 value2，批量设置键和值，如果有一个键存在则设置不成功

命令12：getrange 键起始数终止数，获取键范围内的值

命令13：setrange 键起始位置替换的值，如果有1个键是已存在的都会设置失败，具有原子性

命令14：setex key键过期时间 value值，设置带有过期时间的key键

命令15：getset key键 value值，以新值换旧值

3.2 List列表

单键多值，Redis列表是简单的字符串列表，按照插入顺序排序。可以添加一个元素到列表的头部（左边）或者尾部（右边）。它的底层实际是个双向链表，对两端的操作性能很高，通过索引下标的操作中间的节点性能会较差。

数据结构：List的数据结构为快速链表quickList。首先在列表元素较少的情况下使用一块连续的内存存储，结构是ziplist，也即是压缩列表。它将所有元素挨在一起存储，分配的是一块连续内存。只有数据较多时才会改成quicklist。因为普通链表需要的附加指针空间太大，会比较浪费空间。所以Redis将链表和zipset结合起来组成了quicklist，也就是将多个zipset使用双向指针串起来使用。满足了快速的插入删除性能，不会出现太大的空间冗余。

命令1：lpush key键值值... 表示从key键的列表左边加入值

命令2：lrange key键 0 -1，表示取出键中的所有值（比如lpush k1 v1 v2 v3，那么v1在最右边，v3在最左边，等到输出的话第1个就是v3）

命令3：rpop key键，从右边删除value。lpop key键，从左边删除value。

命令4：rpoplpush key1键 key2键，从key1列表右边吐出一个值，插到key键列表左边

命令5：lindex key键 index值，按照索引下标获取元素（从左到右）

命令6：linsert key键 before value值新value值，会在key键的value值前插入新的value值。

命令7：lrem key键数量n value值，从左边删除n个value

命令8：lset key键 index值 value值，将列表key下标为index的值替换为value

3.3 Set集合

set可以自动排重，当需要存储一个列表数据，又不希望有重复数据可以选择。set是String类型的无序集合，底层是一个value为null的hash表，所以添加、删除、查找的复杂度都是O（1）。

数据结构：set数据结构是dict字典，字典是用哈希表实现的。Redis的set结构内部使用hash结构，所有value指向同一个内部值。

命令1：sadd key键 value值.. ，添加键值对。

命令2：smembers key键，取出键内的值。

命令3：sismember key键 value值，查看是否在键内存在唯一value值。返回1代表存在，返回0代表不存在。

命令4：scard key键，返回集合中元素的个数

命令5：srem key键 value值...，可以删除掉key键中的value值

命令6：spop key键，从集合中随机吐出一个值

命令7：srandmember key键数字，随机从集合中取出n个值，不会删除。

命令8：smove 源key键目标key键 value值，将源key键中的value值转移到目标key键中。

命令9：sinter key键1 key键2，取两个key键中value值的交集。

命令10：sunion key键1 key键2，取两个key键中value值的并集。

命令11：sdiff key键1 key键2，返回两个集合的差集元素（key1中的，不包含key2中的）

3.4 Hash哈希

redis hash是一个键值对集合。是一个string类型的field和value的映射表，hash特别适合用于存储对象，类似Java里面的Map<String,Object>。用户user为查找的key，存储的value用户对象包含：ID、姓名、年龄等信息

Hash类型对应的数据结构有2中：ziplist压缩列表，hashtable哈希表。当field-value长度较短且个数较少时，使用ziplist，否则使用hashtable。

命令1：hset key键名 field名 value名，给key集合中的field键赋值value值（hset user:1001 id 1）

命令2：hget key键名 field名，从key集合field中取出value

命令3：hmset key键名 field1名 value1 field2名 value2，批量设置hash的值（hmset user:1002 id 2 name lisi age 30）

命令4：hexists key键名 field名，查看哈希表key中，给定域field是否存在（hexists user:1002 name）。

命令5：hkeys key键名，查看hash集合的所有field。

命令6：hvals key键名，列出该hash集合的所有value

命令7：hincrby key键 field名数字，为哈希表key中的域field的值加上增量，增量为输入的数字。

命令8：hsetnx key键 field名 value值，将哈希表key中的域field的值设置为value，当且仅当域field不存在。

3.5 Zset有序集合

redis有序集合zset与普通集合set非常相似，是一个没有重重复元素的字符串集合。不同之处是有序集合的每个成员都关联了一个评分（score)，这个评分（score）被用来按照从最低分到最高分的方式排序集合中的成员。集合的成员是唯一的，但评分可以是重复的。因为元素是有序的，所以可以很快的根据评分（score）或者次序（position）来获取一个范围的元素。访问有序集合的中间元素是非常快的，因此能够使用永续集合作为一个没有重复成员的智能列表。

数据结构：SortedSet（zset）是Redis提供的一个非常特别的数据结构，一方面它等价于Java的数据结构Map<String,Double>，它可以给每一个元素value赋予一个权重score，另一方面它又类似于TreeSet，内部的元素会按照权重score进行排序，可以得到每个元素的名次，还可以通过score的范围来获取元素列表。

zset底层使用了2个数据结构：1.hash，hash的作用就是关联元素value和权重score，保证元素value的唯一性，可以通过value找到相应的score值。2.跳跃表，跳跃表的目的在于给元素vale排序，根据score的范围获取元素列表。

命令1：zadd key键 score1名 value1值 score2名 value2值，将一个或多个member元素及其score值假如到有序集key当作（topen 200 java 300 c++ 400 mysql 500 php
）。

命令2：zrange key键名 0 -1，返回有序集key中，下标在start到stop之间的元素。排序是从小到大。

命令3：zrangebyscore key键名起始数终止数，显示评分在起始数到终止数间的成员结果。

命令4：zrevrangebyscore key 大数小数，返回有序集key中，所有score值结语max和min之间的成员。

命令5：zincrby key键名增加量 value，为元素的score加上增量（zincrby topen 50 java）。

命令6：zcount key键名下限上限，统计该集合，分数区间内的元素个数。

命令7：zrank key键名 value，返回该值在集合中的排名，从0开始（zrank topen php）。

四、Redis6配置文件详解

vi /etc/redis.conf可进入配置文件查看配置。

units单位：配置大小单位，开头定义了一些基本的度量单位，只支持bytes，不支持bit，大小写不敏感。

protected-mode：保护模式，yes不支持远程访问，no支持。

port：设置端口号。

bind：默认情况bind=127.0.0.1只能接收本机的访问请求。不写的情况下，无限制接收任何ip地址的访问。生产环境要写你应用服务器的地。服务器是需要远程访问的，所以需要将其注释掉。如果开启了protected-mode，那么在没有设定bind ip且没有设置密码的情况下，redis只接收本机的响应。

tcp-backlog：设置tcp的backlog，backlog其实是一个连接队列，backlog队列总和=未完成三次握手队列+已经完成三次握手队列。在高并发环境下你需要一个高backlog值来避免慢客户端连接问题。

timeout：是超时设置，为0代表永不超时。

TCP keepalive：检查心跳，看是否还在继续操作。

daemonize：redis后台启动。

loglevel：debug在开发环境中能看到更详细信息。verbose能够看到有用的信息。notice生产环境使用，默认。warning是非常重要且有用的信息。

logfile：可以更改存放日志的文件

databases：默认的数据库数量。

CLIENTS：maxclients设置redis同时可以与多少个客户端进行连接，默认是10000个，如果达到极限，redis会拒绝新的连接请求，并且向连接请求方发出“max number of clients reached”以作回应。

maxmemory：必须设置，否则将内存占满，造成服务器宕机。设置redis可以使用的内存量，一旦达到上线，redis会试图移除数据，移除规则可以通过maxmemory-policy指定。

五、Redis6的发布和订阅

Redis发布订阅（pub/sub）是一种消息通信模式：发送者pub发送消息，订阅者sub接收消息。Redis客户端可以订阅任意数量的频道。

首先在Xshell开启2个会话，在每个会话中启动与redis服务器的连接，可以使用同一个Linux服务器的ip。在服务器1中输入SUBSCRIBE channel1，用于订阅频道1。然后服务器2中输入publish channel1 待发送的文字。

六、Redis6新数据类型

6.1 Bitmaps

Bitmaps本身不是一种数据类型，实际上它就是字符串（key-value），但是它可以对字符串的位进行操作。

Bitmaps单独提供了一套命令，所以在Redis中使用Bitmaps和使用字符串的方法不太相同。可以把Bitmaps想象成一个以位为单位的数组，数组的每个单元只能存储0和1，数组的下标在Bitmaps中叫做偏移量。

命令1：setbit key键偏移量 value值，设置Bitmaps中某个偏移量的值（0或1）（例子：setbit users:20210101 1 1）

命令2：getbit key键偏移量，获取Bitmaps中某个偏移量的值（getbit users:20210101 6）

命令3：bitcount key键，获得key键里1的数量（例子：bitcount users:20210101）

命令4：bitop and(or/not/xor) destkey值，是一个复合操作，可以做多个Bitmaps的and、or、not、xor操作并将结果保存在destkey中。

6.2 HyperLogLog

命令1：pfadd key键名元素，向key里添加元素（例子：pfadd program "c++" "mysql"）

命令2：pfcount key键名，查看key中有几个元素

命令3：pfmerge 新key键旧key键1 旧key键2，将旧的key键里的元素合并到新的key键里。

6.3 Geospatial

命令1：geoadd key键名经度维度成员，添加地点的经纬度（geoadd china:city 121.47 31.23 shanghai）

命令2：geopos key键名成员，获取地点的经纬度

命令3：geodist key键名成员1 成员2 长度单位，获取两个地点直线距离（geodist china:city shanghai beijing km ）

命令4：georadius key键名经度维度范围长度单位，以给定的经纬度未中心，某一半径内的元素。

七、Jedis操作Redis6

IDEA新建maven工程，GroupID写com.atguigu，ArtifactID写jedis_redisdemo，引入Jedis所需要的jar包：

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

在java下创建com.atguigu.jedis然后创建一个测试类JedisDemo1，填写主机地址（就是虚拟机的ip），填写端口号。调用jedis里的ping方法，如果能连上redis就会返回PONG，否则会报错：

public class JedisDemo1 {
    public static void main(String[] args){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        String value = jedis.ping();
        System.out.println(value);
    }
}

需要注意前面配置文件里要将bind注释掉，将protected mode改为no。注意虚拟机里有防火墙，想要连接必须关闭防火墙，关闭防火墙：systemctl stop firewalld，

5. 在同一个类下编写如下代码，可以输出数据库中所有key和value

@Test
    public  void demo1(){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        Set<String> keys = jedis.keys("*");
        for(String key:keys){
            System.out.println(key);
        }
    }

jedis.set("key键","value值") 设置键值对。

keys.size() 返回key的数量

jedis.exists("key键")

jedis.ttl("key键")

jedis.get("key键")

jedis.mset("key键1","value值1","key键2","value值2"); 设置多个键值对

List<String> mget = jedis.mget("key键1","key键2"); 获取多个键值对

@Test
    public  void demo1(){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        jedis.lpush("key1","lucy","mary","jack");
        List<String> values = jedis.lrange("key1", 0, -1);
        System.out.println(values);
    }
@Test
    public  void demo1(){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        jedis.sadd("name","lucy","jack");
        Set<String> names = jedis.smembers("name");
        System.out.println(names);
    }
@Test
    public  void demo1(){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        jedis.hset("users","age","20");
        String hget = jedis.hget("users","age");
        System.out.println(hget);
    }
@Test
    public  void demo1(){
        Jedis jedis = new Jedis("192.168.182.140",6379);
        jedis.zadd("china",100d,"shanghai");
        Set<String> china = jedis.zrange("china", 0, -1);
        System.out.println(china);
    }

手机验证码案例：

1. 随机生成6位数字码：用Random。

2. 验证码在2分钟内有效：把验证码放到redis里面，设置过期时间120秒。

3. 判断验证码是否正确：从redis中获取验证码和输入的验证码进行比较。

4. 每个手机每天只能发送3次验证码：incr每次发送后+1，当大于2不能再发送

先创建一个类PhoneCode，创建一个main方法：

public class PhoneCode {
    public static void main(String [] argv){
        }
}

然后创建一个生成验证码的方法：

//生成6位数字验证码
public static String getCode(){
    Random random = new Random();
    String code = "";
    for(int i=0;i<6;i++){
        int rand = random.nextInt(10);
        code += rand;
    }
    return code;
}

3.将结果进行输出进行验证

public static void main(String [] argv){
    String code = getCode();
    System.out.println(code);
}

4. setex key键过期时间 value值，设置带有过期时间的key键

//每个手机每天只能发送三次，验证码放到redis中，设置过期时间
public static void verifyCode(String phone){
    //拼接redis
    Jedis jedis = new Jedis("192.168.182.140",6379);
    //拼接key
    //手机发送次数key
    String countKey = "VerifyCode"+phone+":count";
    //验证码key
    String codeKey = "VerifyCode"+phone+":code";
    //每个手机每天只能发送3次
    String count = jedis.get(countKey);
    if(count==null){
        //没有发送次数，第一次发送，设置发送次数为1
        jedis.setex(countKey,24*60*60,"1");
    }else if(Integer.parseInt(count)<=2){
        jedis.incr(countKey);
    }else if(Integer.parseInt(count)>2){
        System.out.println("今天发送次数已经超过三次");
        jedis.close();
        return;
    }
    String vcode = getCode();
    jedis.setex(codeKey,120,vcode);
    jedis.close();
}

//生成6位数字验证码
public static String getCode(){
    Random random = new Random();
    String code = "";
    for(int i=0;i<6;i++){
        int rand = random.nextInt(10);
        code += rand;
    }
    return code;
}

public static void main(String [] argv){
    verifyCode("17359456898");
}

5. 点击启动，然后看是否成功：

keys *
get VerifyCode17359456898:code
get VerifyCode17359456898:count

public static void main(String [] argv){
    //verifyCode("17359456898");
    getRedisCode("17359456898","272212");
}

八、SpringBoot整合Redis

1. Group：com.atguigu。Artifact：redis_springboot。Java Version：8。

记得更改版本号为2.2.1.RELEASE：

<parent>
		<groupId>org.springframework.boot</groupId>
		<artifactId>spring-boot-starter-parent</artifactId>
		<version>2.2.1.RELEASE</version>
		<relativePath/> <!-- lookup parent from repository -->
</parent>

2.引入依赖：

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

3. 配置文件application.properties配置：

spring.redis.host=192.168.182.140
spring.redis.port=6379
spring.redis.database=0
spring.redis.timeout=1800000
spring.redis.lettuce.pool.max-active=20
spring.redis.lettuce.pool.max-idle=5
spring.redis.lettuce.pool.min-idle=0

4.在java/config下创建RedisConfig配置类：

@EnableCaching
@Configuration
public class RedisConfig extends CachingConfigurerSupport {
    @Bean
    public RedisTemplate<String,Object> redisTemplate(RedisConnectionFactory factory){
        RedisTemplate<String,Object> template = new RedisTemplate<>();
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        template.setConnectionFactory(factory);
        template.setKeySerializer(redisSerializer);
        template.setValueSerializer(jackson2JsonRedisSerializer);
        template.setHashValueSerializer(jackson2JsonRedisSerializer);
        return template;
    }
    @Bean
    public CacheManager cacheManager(RedisConnectionFactory factory){
        RedisSerializer<String> redisSerializer = new StringRedisSerializer();
        Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
        ObjectMapper om = new ObjectMapper();
        om.setVisibility(PropertyAccessor.ALL,JsonAutoDetect.Visibility.ANY);
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jackson2JsonRedisSerializer.setObjectMapper(om);
        RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
            .entryTtl(Duration.ofSeconds(600))
            .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(redisSerializer))
            .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
            .disableCachingNullValues();
        RedisCacheManager cacheManager = RedisCacheManager.builder(factory)
            .cacheDefaults(config)
                .build();
        return cacheManager;
    }
}

5.在java/controller下建立类RedisTestController：

@RestController
@RequestMapping("/redisTest")
public class RedisTestController {
    @Autowired
    private RedisTemplate redisTemplate;
    @GetMapping
    public String testRedis(){
        //设置值到redis
        redisTemplate.opsForValue().set("name","lucy");
        //从redis获取值
        String name = (String)redisTemplate.opsForValue().get("name");
        return name;
    }
}

6.测试，输入网址，返回lucy

九、事务和锁机制

Multi是开启事务，Exec会执行命令，discard放弃组队：

如果组队阶段某个命令出错，则所有命令都不会执行：

如果执行阶段某个命令报出错误，则只有报错的命令不会被执行而其它命令都会执行，不会回滚：

十、事务和锁机制

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

乐观锁（optimistic Lock）：每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号（数据库和操作都被赋予版本号，当操作执行前需要看操作的版本号和数据库的版本号是否一致，不一致时拒绝执行，只有当一致时操作才能执行）等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量。Redis就是利用这种check-and-set机制实现事务的。场景：抢票，秒杀...

WATCH key ：在执行multi之前，先执行watch key1[key2]，可以监视一个或多个key，如果在事务执行之前这个（或这些）key被其他命令所改动，那么事务将被打断。

1.首先打开2个终端，注意一定要连接上redis，

2. 终端1：set balance 100，watch balance。终端2：watch balance。

3. 终端1：multi（Multi是开启事务），终端2：multi。

4. 终端1：incrby balance 10。终端2：incrby balance 20。

5. 终端1：exec执行，成功，终端2：exec执行，失败

Redis事务三特性：

1.单独的隔离操作：事务中所有命令都会序列化、按顺序地执行。事务在执行过程中，不会被其他客户端发送来的命令请求所打断。

2.没有隔离级别的概念：队列中的命令没有提交之前都不会实际被执行因为事务提交前任何指令都不会被实际执行。

3.不保证原子性：事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚。

秒杀：

1. 在webapp下创建index.jsp。定义了一个表单，效果如下图：

<%@ page language="java" contentType="text/html; charset=UTF-8"
         pageEncoding="UTF-8"%>
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN" "http://www.w3.org/TR/html4/loose.dtd">
<html>
<head>
    <meta http-equiv="Content-Type" content="text/html; charset=UTF-8">
    <title>Insert title here</title>
</head>
<body>
<h1>iPhone 13 Pro !!!  1元秒杀！！！
</h1>
<form id="msform" action="${pageContext.request.contextPath}/doseckill" enctype="application/x-www-form-urlencoded">
    <input type="hidden" id="prodid" name="prodid" value="0101">
    <input type="button"  id="miaosha_btn" name="seckill_btn" value="秒杀点我"/>
</form>
</body>
<script  type="text/javascript" src="${pageContext.request.contextPath}/script/jquery/jquery-3.1.0.js"></script>
<script  type="text/javascript">
    $(function(){
        $("#miaosha_btn").click(function(){
            var url=$("#msform").attr("action");
            $.post(url,$("#msform").serialize(),function(data){
                if(data==0){alert("秒杀还没开始，请等待" );}
                if(data==1){alert("秒杀成功了..." );}
                if(data==2){alert("已经秒杀成功,不能成功秒杀" );}
                if(data==3){
                    alert("秒杀已经结束" );
                    $("#miaosha_btn").attr("disabled",true);
                }
                if(data==4){alert("用户或商品id为空" );}
            } );
        })
    })
</script>
</html>

【$("#miaosha_btn").click(function(){】意思是：当点击id为miaosha_btn按钮时，会触发事件【var url=$("#msform").attr("action");】意思是：定义变量url的值是id为msform的表单中的action元素，其中action="${pageContext.request.contextPath}/doseckill,pageContext.request.contextPath应该对应根路径，注意这里的doseckill对应web.xml中的<servlet-name>【$.post(url,$("#msform").serialize(),function(data){】应该是发送了一个post请求到url页面，然后data是返回的参数。【alert("秒杀已经结束" );】是在页面顶端弹出一个提示框。【$("#miaosha_btn").attr("disabled",true);】是让按钮变成不可点击状态。

2. webapp下的WEB-INF中的web.xml文件内容要更改为如下：

<?xml version="1.0" encoding="UTF-8"?>
<web-app xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee
		 http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd" version="2.5">
  <servlet>
    <description></description>
    <display-name>doseckill</display-name>
    <servlet-name>doseckill</servlet-name>
    <servlet-class>com.atguigu.SecKillServlet</servlet-class>
  </servlet>
  <servlet-mapping>
    <servlet-name>doseckill</servlet-name>
    <url-pattern>/doseckill</url-pattern>
  </servlet-mapping>
</web-app>

【<servlet-name>doseckill</servlet-name> <servlet-class>com.atguigu.SecKillServlet</servlet-class>】这段代码大致的意思是如果访问路径为/doseckill，就去请求com/atguigu下的SecKillServlet类。

记得将script/jquery/jquery-3.1.0.js文件放在webapp目录下。

3. 在java/com/atguigu包下创建SecKillServlet类

public class SecKillServlet extends HttpServlet {
    private static final long serialVersionUID = 1L;
    public SecKillServlet(){super();}
    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
        String userid = new Random().nextInt(50000)+""; //用随机数生成用户Id
        String prodid = request.getParameter("prodid"); //获取商品id
        int isSuccess = SecKill_redis.doSecKill(userid,prodid); //调用方法做秒杀过程，判断返回的结果成功还是失败
        response.getWriter().print(isSuccess);
    }
}

调用SecKillServlet类之后，会自动调用doPost方法，关键是下面这段代码【SecKill_redis.doSecKill(userid,prodid);】调用了SecKill_redis类中的doSecKill方法。

4.在java/com/atguigu包下创建SecKill_redis类：

doSecKill是核心方法，用于设定秒杀核心逻辑：

public class SecKill_redis {
    public static void main(String[] args){
        Jedis jedis = new Jedis("192.168.182.142",6379);
        System.out.println(jedis.ping());
        jedis.close();
    }
    public static int doSecKill(String uid,String prodid) throws IOException {
        //1.uid和prodid非空判断
        if(uid==null || prodid==null) return 4;
        //2.jedis连接redis
        Jedis jedis = new Jedis("192.168.182.144",6379);
        //3.拼接key，库存key，秒杀成功用户key(sk是stock缩写，表示库存)
        String kcKey = "sk:" + prodid + ":qt";
        String userKey = "sk:" + prodid + ":user";
        //4.获取库存，如果库存为null，秒杀还未开始
        String kc = jedis.get(kcKey);
        if(kc==null) {
            System.out.println("秒杀还没开始，请等待");
            jedis.close();
            return 0;
        }
        //5.判断用户是否重复秒杀操作
        if(jedis.sismember(userKey,uid)){
            System.out.println("已经秒杀成功,不能成功秒杀");
            jedis.close();
            return 2;
        }
        //6.判断如果商品数量,库存数量小于1,秒杀结束
        if(Integer.parseInt(kc)<=0){
            System.out.println("秒杀已经结束");
            jedis.close();
            return 3;
        }
        //7.秒杀过程(秒杀到库存-1，把秒杀成功用户添加到清单里面)
        jedis.decr(kcKey);
        jedis.sadd(userKey, uid);
        System.out.println("秒杀成功了...");
        jedis.close();
        return 1;
    }
}

场景：在某天的某个时间，商家对商品进行低价销售，有一定的数量限额。操作：商品库存 -1，秒杀成功者清单 +1。

测试：设置库存为10：set sk:0101:qt 10。然后刷新网页，点击秒杀点我。会有弹窗显示秒杀成功。get sk:0101:qt。smembers sk:0101:user。

秒杀并发模拟：

1. 使用ab模拟测试，先通过如下命令进行安装：yum install httpd-tools

2. 通过vi postfile创建一个文件夹，写入如下内容：prodid=0101&

3. 输入指令：ab -n 1000 -c 100 -p ~/postfile -T application/x-www-form-urlencoded http://192.168.173.172:8080/doseckill

注意3点：首先类型那里不用加任何的引号，其次192.168.173.172是windows的主机ip而不是虚拟机，最后doseckill是action的后缀。

-n请求的数量，-c是请求中的并发数量，-p提交的参数，-t参数的类型，~/postfile是通过vi创建的哪个文件，'application/x-www-form-urlencoded'是参数的类型，http://主机的地址/seckill/doseckill是url。http://192.168.173.172:8080。

连接池：

连接超时问题通过连接池解决。

连接池：节省每次连接redis服务带来的消耗，把连接好的实例反复利用，通过参数管理连接行为。

创建JedisPoolUtil类，getJedisPoolInstance方法，加synchronized锁，setMaxTotal最大连接数，setBlockWhenExhausted超过了是否进行等待，setTestOnBorrow检测是否连接状态。

new JedisPool第1个是基本配置，第2个是ip地址（ip地址要修改为虚拟机的ip地址），第3个是端口号，第4个是超时时间。

public class JedisPoolUtil {
	private static volatile JedisPool jedisPool = null;
	private JedisPoolUtil() {
	}
	public static JedisPool getJedisPoolInstance() {
		if (null == jedisPool) {
			synchronized (JedisPoolUtil.class) {
				if (null == jedisPool) {
					JedisPoolConfig poolConfig = new JedisPoolConfig();
					poolConfig.setMaxTotal(200);
					poolConfig.setMaxIdle(32);
					poolConfig.setMaxWaitMillis(100*1000);
					poolConfig.setBlockWhenExhausted(true);
					poolConfig.setTestOnBorrow(true);  // ping  PONG
					jedisPool = new JedisPool(poolConfig, "192.168.182.145", 6379, 60000 );
				}
			}
		}
		return jedisPool;
	}
	public static void release(JedisPool jedisPool, Jedis jedis) {
		if (null != jedis) {
			jedisPool.close();
		}
	}
}

超卖问题：

出现问题：连接超时问题，超卖问题（东西没了还能买，库存最终为负数）。

 public static int doSecKill(String uid,String prodid) throws IOException {
        //1.uid和prodid非空判断
        if(uid==null || prodid==null) return 4;
       //2. 通过连接池得到jedis对象
        JedisPool jedisPoolInstance = JedisPoolUtil.getJedisPoolInstance();
        Jedis jedis = jedisPoolInstance.getResource();
        //3.拼接key，库存key，秒杀成功用户key(sk是stock缩写，表示库存)
        String kcKey = "sk:" + prodid + ":qt";
        String userKey = "sk:" + prodid + ":user";
        jedis.watch(kcKey); //加入库存key因为库存会不断变化
        //4.获取库存，如果库存为null，秒杀还未开始
        String kc = jedis.get(kcKey);
        if(kc==null) {
            System.out.println("秒杀还没开始，请等待");
            jedis.close();
            return 0;
        }
        //5.判断用户是否重复秒杀操作
        if(jedis.sismember(userKey,uid)){
            System.out.println("已经秒杀成功,不能重复秒杀");
            jedis.close();
            return 2;
        }
        //6.判断如果商品数量,库存数量小于1,秒杀结束
        if(Integer.parseInt(kc)<=0){
            System.out.println("秒杀已经结束");
            jedis.close();
            return 3;
        }
        //7.秒杀过程(秒杀到库存-1，把秒杀成功用户添加到清单里面)
        //使用事务
        Transaction multi = jedis.multi();
        //组队操作
        multi.decr(kcKey);
        multi.sadd(userKey,uid);
        List<Object> results = multi.exec();
        if(results==null || results.size()==0){
            System.out.println("秒杀失败了...");
            jedis.close();
            return 5;
        }
        System.out.println("秒杀成功了...");
        jedis.close();
        return 1;
}

先用watch监视库存【jedis.watch(kcKey);】开启事务【Transaction multi = jedis.multi();】进行组队【multi.decr(kcKey);multi.sadd(userKey,uid);】最终执行【List<Object> results = multi.exec();】

库存遗留问题：

原因：因为版本号不一致，导致其它人不能往下购买。

解决：使用LUA脚本语言

ab -n 2000 -c 300 -p ~/postfile -T application/x-www-form-urlencoded http://192.168.173.172:8080/doseckill

定义了2个变量，分别传入1个用户id和1个商品id：

拼接库存key和用户key

调用redis命令看用户是否在清单中存在：

秒杀过了就不能再秒杀第2次，返回2表示秒杀过了：

查看库存如果<=0则秒杀结束，返回0表示秒杀完了结束：

否则就让库存量-1，加入用户的信息，返回1，秒杀成功

带有Lua脚本的SecKill_redisByScript类内容如下：

public class SecKill_redisByScript {
	private static final  org.slf4j.Logger logger =LoggerFactory.getLogger(SecKill_redisByScript.class) ;
	public static void main(String[] args) {
		JedisPool jedispool =  JedisPoolUtil.getJedisPoolInstance();
		Jedis jedis=jedispool.getResource();
		System.out.println(jedis.ping());
		Set<HostAndPort> set=new HashSet<HostAndPort>();
	//	doSecKill("201","sk:0101");
	}
	static String secKillScript ="local userid=KEYS[1];\r\n" + 
			"local prodid=KEYS[2];\r\n" + "local qtkey='sk:'..prodid..\":qt\";\r\n" + 
			"local usersKey='sk:'..prodid..\":usr\";\r\n" + 
			"local userExists=redis.call(\"sismember\",usersKey,userid);\r\n" + 
			"if tonumber(userExists)==1 then \r\n" + "   return 2;\r\n" + "end\r\n" + 
			"local num= redis.call(\"get\" ,qtkey);\r\n" + 
			"if tonumber(num)<=0 then \r\n" + "   return 0;\r\n" + "else \r\n" + 
			"   redis.call(\"decr\",qtkey);\r\n" + 
			"   redis.call(\"sadd\",usersKey,userid);\r\n" + "end\r\n" + "return 1" ;
	static String secKillScript2 = 
			"local userExists=redis.call(\"sismember\",\"{sk}:0101:usr\",userid);\r\n" +
			" return 1";

	public static boolean doSecKill(String uid,String prodid) throws IOException {
		JedisPool jedispool =  JedisPoolUtil.getJedisPoolInstance();
		Jedis jedis=jedispool.getResource();
		 //String sha1=  .secKillScript;
		String sha1=  jedis.scriptLoad(secKillScript);
		Object result= jedis.evalsha(sha1, 2, uid,prodid);
		  String reString=String.valueOf(result);
		if ("0".equals( reString )  ) {
			System.err.println("已抢空！！");
		}else if("1".equals( reString )  )  {
			System.out.println("抢购成功！！！！");
		}else if("2".equals( reString )  )  {
			System.err.println("该用户已抢过！！");
		}else{
			System.err.println("抢购异常！！");
		}
		jedis.close();
		return true;
	}
}

更改SecKillServlet类中的doPost方法，将调用的方法改为调用SecKill_redisByScript中的doSecKill方法：

boolean b = SecKill_redisByScript.doSecKill(userid, prodid);//调用方法做秒杀过程，判断返回的结果成功还是失败
response.getWriter().print(b);

十一、持久化RDB和AOF

持久化：将数据存入硬盘

11.1 RDB（Redis Database）

RDB：在指定的时间间隔内将内存中的数据集快照写入磁盘，也就是行话讲的Snapshot快照，它恢复时是将快照文件直接读到内存里。

备份是如何执行的：

Fork：

在/usr/local/bin目录下输入：vim /etc/redis.conf。找到SNAPSHOTTING。

dbfilename dump.rdp的意思是进行rdb后文件的名字为dump.rdg。

stop-writes-on-bgsave-error如果为yes代表：当Redis无法写入磁盘的话，直接关掉Redis的写操作，推荐yes。

rdbcompression压缩文件。

rdbchecksum如果为yes代表：检查完整性，推荐yes。

sava 秒钟写操作次数。默认60分钟内有1个key发生变化，即进行持久化操作。5分钟内有10个key发生变化进行持久化操作。1分钟如果1000key发生变化就进行持久化操作。

操作步骤：在redis.conf配置文件中位置的下方写入save 20 3，wq!，然后ps -ef | grep redis杀掉进程，然后重新启动redis-server /etc/redis.conf，输入ll，查看dump.rdb文件的大小，

bgsave：Redis会在后台异步进行快照操作，快照同时还可以响应客户端请求

优势：

劣势：

11.2 AOF

— — — — — — — — — — — — — — — —

先输入： cd /usr/local/bin，再输入：redis-server /etc/redis.conf 启动服务器，/usr/bin/redis-cli 进入redis