Redis入门(4): Redis_事务_锁机制_秒杀

秒杀项目的 github 地址如下：https://github.com/MRtianyanxiaobai/redis_Seckill

Redis 的事务

Redis的事务定义

Redis事务是一个单独的隔离操作：

事务中的所有命令都会序列化、按顺序地执行。
事务在执行的过程中，不会被其他客户端发送来的命令请求所打断。
Redis事务的主要作用就是串联多个命令防止别的命令插队。

组队过程：Multi、Exec、discard

从输入Multi 命令开始，输入的命令都会依次进入命令队列中，但不会执行
直到输入 Exec 后，Redis会将之前的命令队列中的命令依次执行。
组队的过程中可以通过 discard 来放弃组队。

案例

组队成功，提交成功
组队阶段报错，提交失败
组队成功，提交有成功有失败情况

事务的错误处理

组队阶段，一错全错： 组队中某个命令出现了报告错误，执行时整个的所有队列都会被取消。
运行阶段，一错在己： 如果执行阶段某个命令报出了错误，则只有报错的命令不会被执行，而其他的命令都会执行，不会回滚。

悲观锁

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

乐观锁

乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁。
但是在更新的时候会判断一下在此期间别人有没有去更新这个数据（CAS机制），可以使用版本号等机制。
乐观锁适用于多读的应用类型，这样可以提高吞吐量。Redis就是利用这种check-and-set机制实现事务的。

监视器：WATCH key [key …]

在执行 multi 之前，先执行 watch key1 [key2] ,可以监视一个(或多个) key
如果在事务执行之前这个(或这些) key 被其他命令所改动，那么事务将被打断。
正常情况如下：
现在 watch 一个 key 后，在事务开始之前对其进行提前修改：

取消监视： unwatch

取消 WATCH 命令对所有 key 的监视。
如果在执行 WATCH 命令之后，EXEC 命令或 DISCARD 命令先被执行了的话，那么就不需要再执行UNWATCH 了。
关闭 watch 之后，事务就可以执行成功了

http://doc.redisfans.com/transaction/exec.html

Redis中的事务三特性（顺序性，隔离性，原子性）

顺序性

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

隔离性：Redis 中没有隔离级别的概念

队列中的命令没有提交之前都不会实际被执行，因为事务提交前任何指令都不会被实际执行

不保证原子性

事务中如果有一条命令执行失败，其后的命令仍然会被执行，没有回滚

事务秒杀并发模拟

需求

模拟秒杀商品的功能。
为了简化案例，这里我们先不使用数据库，直接使用 Redis 来做实时的数据更新。主要需要维护的变量如下：
在这里插入图片描述

用户点击秒杀按钮，redis中的商品库存和成功者清单需要进行准确且实时的更新。

基础代码编写

1、创建前端页面

首先，我们创建一个简单的前端页面：

<%@ 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>

<%--秒杀表单，用户填写表单信息，单价按钮之后，提交表单信息--%>
<%--简化案例，假设这里只有一个商品，商品号位 0101--%>
<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(){
      
      
    //利用 ajax 对请求返回的结果进行处理回显
    $("#miaosha_btn").click(function(){
      
      
      var url=$("#msform").attr("action");
      $.post(url,$("#msform").serialize(),function(data){
      
      
        if(data=="false"){
      
      
          alert("抢光了" );
          $("#miaosha_btn").attr("disabled",true);
        }
      } );
    })
  })
</script>
</html>

在这里插入图片描述

2、导入需要依赖的包

    <dependencies>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.2.0</version>
        </dependency>
        <!-- https://mvnrepository.com/artifact/javax.servlet/javax.servlet-api -->
        <dependency>
            <groupId>javax.servlet</groupId>
            <artifactId>javax.servlet-api</artifactId>
            <version>3.0.1</version>
            <scope>provided</scope>
        </dependency>

    </dependencies>

3、编写 servlet并且配置xml

编写 servlet

public class secKillServlet extends HttpServlet
{
      
      
    private static final  long serialVersionUID = 1L;
    public  secKillServlet(){
      
      
        super();
    }

    //post 方法发送秒杀请求
    @Override
    protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
      
      
        //假设当前登录的 userid
        String userid = new Random().nextInt(50000)+"";
        //默认 1101
        String proid = req.getParameter("prodid");

        //进行秒杀，此时该方法还未实现
        boolean isSuccess = Seckill_redis.doSecKill(userid,proid);
        //写回页面
        resp.getWriter().print(isSuccess);

    }
}

配置web.xml

    <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>

4、Seckill_redis.doSecKill 秒杀过程的实现

主要步骤如下：

1、uid 和 prodid 非空的判断
2、连接redis
3、拼接key
4、获取库存，如果库存本身为 null，那么秒杀还没有开始
5、判断用户是否重复秒杀
6、判断如果商品数量：库存量小于1：秒杀结束
7、秒杀的过程
- 把秒杀成功的用户添加到成功表中
- 库存减一

具体代码如下：

package com.atguigu;

import redis.clients.jedis.Jedis;

public class Seckill_redis {
      
      

    //真正的秒杀函数
    public static boolean doSecKill(String userid, String proid) {
      
      
        // 1、uid 和 prodid 非空的判断
        if (userid==null || proid==null){
      
      
            return false;
        }


        // 2、连接redis
        Jedis jedis = new Jedis("xxxx:22",6379);
        jedis.auth("xxx");

        //  3、拼接
        // 3.1 库存的key
        // 3.2 成功名单的key
        String KcKey= "sk:"+proid+":qt";
        String userKey="sk:"+prodid+":user";

        //4. 获取库存，如果库存本身为 null，那么秒杀还没有开始
        String kc = jedis.get(KcKey);
        if(kc==null){
      
      
            System.out.println("秒杀没有开始");
            jedis.close();
            return  false;
        }
        //5. 判断用户是否重复秒杀
        //从set中能够去除userid则表示已经秒杀了，不能在秒杀了
        if(jedis.sismember(userKey,userid)){
      
      
            System.out.println("已经秒杀成功了，不能重复秒杀");
            jedis.close();
            return false;
        }

        //6. 判断如果商品数量：库存量小于1：秒杀结束
        Integer intkc = Integer.parseInt(kc);
        if(intkc<=0){
      
      
            System.out.println("秒杀已经结束，库存已经清零！！");
            jedis.close();
            return  false;
        }

        //7. 秒杀的过程
        // 7.1 把秒杀成功的用户添加到成功表中
        // 7.2 库存减一
        jedis.decr(KcKey);
        jedis.sadd(userKey,userid);
        System.out.println("秒杀成功了。。。");
        jedis.close();

        return  true;

    }
}

5. 测试

首先现在 redis 中添加上面的两个key，并且设置库存的具体值(设置为 8)
- sk:0101:qt: 表示 0101号商品现在的库存量。
- sk:0101:user: 表示购买 0101号商品成功的用户名单。这个set在第一次add时会自动添加
部署环境，点击按钮观察结果：
此时 redis 中两个 key 的值如下：

利用 Jmeter 进行多线程测试：

下载网址：http://jmeter.apache.org/download_jmeter.cgi

1、新建一个线程组。

在这里插入图片描述

2、设置线程组参数。

参数详解:

线程数：虚拟用户数。一个虚拟用户占用一个进程或线程。设置多少虚拟用户数在这里也就是设置多少个线程数。
Ramp-Up Period(in seconds)准备时长：设置的虚拟用户数需要多长时间全部启动。如果线程数为10，准备时长为2，那么需要2秒钟启动10个线程，也就是每秒钟启动5个线程。
循环次数：每个线程发送请求的次数。如果线程数为10，循环次数为100，那么每个线程发送100次请求。总请求数为10*100=1000 。如果勾选了“永远”，那么所有线程会一直发送请求，一到选择停止运行脚本。
Delay Thread creation until needed：直到需要时延迟线程的创建。
调度器：设置线程组启动的开始时间和结束时间(配置调度器时，需要勾选循环次数为永远)
持续时间（秒）：测试持续时间，会覆盖结束时间
启动延迟（秒）：测试延迟启动时间，会覆盖启动时间
启动时间：测试启动时间，启动延迟会覆盖它。当启动时间已过，手动只需测试时当前时间也会覆盖它。
结束时间：测试结束时间，持续时间会覆盖它。

这里配置为：100个线程，同时启动，循环10次。

3、新增http请求默认值。

在上一步创建的线程组上，新增http请求默认值，所有的请求都会使用设置的默认值，这设置协议为http，IP为localhost，端口为8080。
在这里插入图片描述

4、添加要压测的http请求。

在这里插入图片描述

下图第一个红框内的协议、IP、端口不需要设置，会使用步骤c中设置的默认值，只需设置请求路径Path即可，这里填入 /doseckill/doseckill,添加参数：prodid=0101

5、新增监听器，用于查看压测结果。

这里添加三种：聚合报告、图形结果、用表格查看结果，区别在于结果展现形式不同。
在这里插入图片描述

6、点击运行按钮开始压测，并查看结果。

在这里插入图片描述

7、查看后台结果：

在这里插入图片描述
结果出现了下面两个问题：

超卖问题：
- 首先秒杀成功的案例已经超过了原来的 10个。
- 其次库存已经显示清零了，还在秒杀。
- 打开redis 观察当前库存量：
连接超时问题： 由于很多线程访问 redis，但是redis的连接数固定，有些线程就会进行等待，进而造成连接超时问题。

问题的解决

1、连接超时问题的解决：连接池

使用单例模式，创建一个数据库连接池，设置相关参数，节省每次连接redis服务带来的消耗，把连接好的实例反复利用。
链接池参数

MaxTotal： 控制一个pool可分配多少个jedis实例，通过pool.getResource()来获取；如果赋值为-1，则表示不限制；如果pool已经分配了MaxTotal个jedis实例，则此时pool的状态为exhausted。
maxIdle： 控制一个pool最多有多少个状态为idle(空闲)的jedis实例；
MaxWaitMillis： 表示当borrow一个jedis实例时，最大的等待毫秒数，如果超过等待时间，则直接抛JedisConnectionException；
testOnBorrow： 获得一个jedis实例的时候是否检查连接可用性（ping()）；如果为true，则得到的jedis实例均是可用的；

获取连接池的单例模式代码如下：

package com.atguigu;

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

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

	private JedisPoolUtil() {
      
      
	}

	public static JedisPool getJedisPoolInstance() {
      
      
		if (null == jedisPool) {
      
      
			synchronized (JedisPoolUtil.class) {
      
      
				if (null == jedisPool) {
      
      
					JedisPoolConfig poolConfig = new JedisPoolConfig();
					poolConfig.setMaxTotal(200);
					poolConfig.setMaxIdle(32);
					poolConfig.setMaxWaitMillis(100*1000);
					poolConfig.setBlockWhenExhausted(true);
					poolConfig.setTestOnBorrow(true);  // ping  PONG
				 
					jedisPool = new JedisPool(poolConfig, "192.168.44.168", 6379, 60000 );
				}
			}
		}
		return jedisPool;
	}

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

}

只需要使用 getresource 就能获得相应得到jedis 了

//通过连接池得到jedis对象
JedisPool jedisPool = JedisPoolUtill.getJedisPoolInstance();
Jedis jedis = jedisPool.getResource();

2、超卖问题的解决：乐观锁

思路： 如下所示，添加一个乐观锁，即添加一个版本号，只有在版本号未变的情况下，才会对其进行操作。
在这里插入图片描述
解决步骤：

利用 watch 操作，对需要加锁的变量添加一个监控器
利用 multi 和 exec 构造事务。
根据 exec 的返回值判断事务是否成功，如果我们监控的变量改变了，那么exec 就会返回空或者0。如果事务执行成功会返回指令数。

具体代码如下：

//添加乐观锁的秒杀过程
jedis.watch(KcKey);
.................//对Kckey的一些检查，判断是否东西是否已经卖完
.................

//开启事务
Transaction mutil= jedis.multi();
//添加事务操作要用 mutil.xxx
mutil.decr(KcKey);
mutil.sadd(userKey,userid);
List<Object> res = mutil.exec();//执行事务，结果以list返回
if(res==null || res.size()==0){
      
      
    System.out.println("秒杀失败了");
    jedis.close();
    return false;
}

System.out.println("秒杀成功了。。。");
jedis.close();

注意： 我们加锁的地方，务必放在检查的前面。如果放在后面，那么前面点的检查时，Kckey 可能已经发生了修改。

结果如下：

为了方便查看，我们可以访问redis中的具体值，观察结果：

3、库存遗留问题

如果此时将库存量设置为 500，观察结果：
结果如下：
可以看到，虽然此时我们一共发送了 100*10 个请求，但是真正秒杀成功的只有 25 个用户。而redis中真实库存为 500，因此造成了 库存遗留问题。
这个问题的主要原因是乐观锁造成的，简单的说就是，watch 的库存量发生了变化，然后该请求就会秒杀失败。由于秒杀失败的请求太多，导致库存还有剩余。
由于redis 默认不能使用非乐观锁，所以不能使用非乐观锁。

解决方法使用 LUA 脚本

Lua 是一个小巧的脚本语言，Lua 脚本可以很容易的被C/C++ 代码调用，也可以反过来调用C/C++的函数，Lua并没有提供强大的库，一个完整的Lua解释器不过200k，所以Lua不适合作为开发独立应用程序的语言，而是作为嵌入式脚本语言。
LUA脚本是类似redis事务，有一定的原子性，不会被其他命令插队，可以完成一些redis事务性的操作。
但是注意redis的lua脚本功能，只有在Redis 2.6以上的版本才可以使用。redis 2.6版本以后，通过lua脚本解决争抢问题，实际上是redis 利用其单线程的特性，用任务队列的方式解决多任务并发问题。

LUA 的基本格式如下：

//设置中 id并且拼接两种key
local userid=KEYS[1]; 
local prodid=KEYS[2];
local qtkey="sk:"..prodid..":qt";
local usersKey="sk:"..prodid.":usr'; 
//和java代码的逻辑类型，先判断是否该用户已买
local userExists=redis.call("sismember",usersKey,userid);
if tonumber(userExists)==1 then 
  return 2;
end
//再判断该是否还存在库存
local num= redis.call("get" ,qtkey);
if tonumber(num)<=0 then 
  return 0; 
else //最后进行添加操作
  redis.call("decr",qtkey);
  redis.call("sadd",usersKey,userid);
end
return 1;

上面是基础的 LUA 脚本，在 java 中我们需要使用字符串来封装他，并利用它完成我能的检查以及存取操作：

package com.atguigu;

import org.slf4j.LoggerFactory;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class Seckill_redisScript {
      
      
    private static final  org.slf4j.Logger logger = LoggerFactory.getLogger(Seckill_redisScript.class) ;

    public static void main(String[] args) {
      
      
        JedisPool jedispool =  JedisPoolUtill.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 =  JedisPoolUtill.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;
    }
}

在servlet 中直接进行调用：

        //利用 lua 进行redis 操作
        boolean isSuccess = Seckill_redisScript.doSecKill(userid,proid);

测试结果如下：
简单说，就是让每个线程进行的操作为原子操作，可以理解为变相的添加了悲观锁。