一、Redis事务与分布式锁

1.1 Redis事务

1.1.1 Redis事务简介

Redis 事务的本质是一组命令的集合。在Redis中开启事务后，事务中的命令并不会立即执行，而是会推送到一个事务队列中，该队列积攒此次事务的所有命令，等到事务提交（执行）后，会逐步执行此队列中的命令，执行队列中的命令过程是一个整体，不会被其他客户端所干扰。在事务中如果有命令执行错误，那么此次事务队列中的所有命令取消。

• 开启事务，并创建队列

multi

• 执行事务

exec

• 取消事务

discard

Redis在开启事务时，之后执行的所有的操作均会保存到任务队列中。如果执行了exec命令，那么redis将会逐步执行队列中的指令，如果执行discard指令，那么将会取消队列中的所有指令。

事务队列中某个命令出现错误：

1.1.2 watch监视

我们知道redis在开启事务时，其实就是创建了一个事务队列，此次事务所执行的所有命令将不会被立即执行，而是等事务执行之后（exec），才会将队列中的命令一起执行。有时候我们希望在执行队列中的指令之前不允许其他客户端对某个值进行修改，那么此时需要就需要加锁。

• watch：监视某个key

watch key1 [key2……]

如果被监视的值在exec命令之前被其他客户端修改过了，那么此次事务队列中的命令全部失效。

• unwatch：取消监视的所有key

unwatch

tips：执行exec会释放所有被监视的key

1.1.3 Jedis实现Redis事务

在Jdis中提供有Transaction类，用于执行Redis事务相关指令；

• 示例代码：

package com.dfbz.demo01;

import org.junit.Before;
import org.junit.Test;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.Transaction;

/**
 * @author lscl
 * @version 1.0
 * @intro:
 */
public class Demo01 {

    private JedisPool pool;

    @Before
    public void before() {
        pool = new JedisPool("192.168.18.155", 6379);
    }

    @Test
    public void test1() {
        Jedis jedis = pool.getResource();
        // 开启事务
        Transaction tx = jedis.multi();
        try {
            tx.set("name", "zhangsan");

            int i = 1 / 0;

            tx.set("age", "20");

            // 执行事务
            tx.exec();
        } catch (Exception e) {
            e.printStackTrace();

            // 取消事务
            tx.discard();
        }
    }
}

• 锁的监视：

@Test
public void test2() {
    Jedis jedis = pool.getResource();
    jedis.set("count", "0");

    // 监视count
    jedis.watch("count");
    // 开启事务
    Transaction tx = jedis.multi();

    try {
        tx.incr("count");
        // 执行事务
        tx.exec();
    } catch (Exception e) {
        e.printStackTrace();
        // 取消事务
        tx.discard();
    }
}

@Test
public void test3() {
    Jedis jedis = pool.getResource();
    jedis.incr("count");
}

1.1.4 RedisTemplate 实现Redis事务

• 1）引入依赖：

<?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.dfbz</groupId>
    <artifactId>01_Redis_tx_lock</artifactId>
    <version>1.0-SNAPSHOT</version>

    <parent>
        <artifactId>spring-boot-starter-parent</artifactId>
        <groupId>org.springframework.boot</groupId>
        <version>2.0.1.RELEASE</version>
    </parent>

    <dependencies>
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>3.7.0</version>
        </dependency>

        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.13.1</version>
            <scope>test</scope>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>

        <dependency>
            <groupId>com.fasterxml.jackson.core</groupId>
            <artifactId>jackson-databind</artifactId>
        </dependency>
    </dependencies>

</project>

• 2）启动类：

注意：RedisTemplate默认情况下是不支持事务的，需要手动开启；

package com.dfbz;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

/**
 * @author lscl
 * @version 1.0
 * @intro:
 */
@SpringBootApplication
public class RedisApplication {

    @Bean
    public RedisTemplate redisTemplate(RedisTemplate redisTemplate){
        // key序列化
        redisTemplate.setKeySerializer(new StringRedisSerializer());

        // value序列化
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());

        // 开启事务支持
        redisTemplate.setEnableTransactionSupport(true);
        return redisTemplate;
    }

    public static void main(String[] args) {
        SpringApplication.run(RedisApplication.class);
    }
}

• 测试类：

package com.dfbz.demo01;

import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.web.servlet.server.Session;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.SessionCallback;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.ArrayList;

/**
 * @author lscl
 * @version 1.0
 * @intro:
 */
@RunWith(SpringRunner.class)
@SpringBootTest
public class Demo02 {

    @Autowired
    private RedisTemplate redisTemplate;

    @Test
    public void test1() {

        try {
            // 开启事务
            redisTemplate.multi();

            redisTemplate.opsForValue().set("name", "zhangsan");

            int i = 1 / 0;
            redisTemplate.opsForValue().set("age", "20");

            // 执行事务
            redisTemplate.exec();

            System.out.println("事务执行成功！");
        } catch (Exception e) {
            e.printStackTrace();

            // 取消事务
            redisTemplate.discard();

            System.out.println("事务取消！");
        }
    }


    @Test
    public void test2() {

        try {
            redisTemplate.opsForValue().set("count", 0);

            // 开启key的监视
//            redisTemplate.watch("count");

            // 开启事务
            redisTemplate.multi();

            // 自增操作
            redisTemplate.opsForValue().increment("count", 1);

            // 执行事务
            redisTemplate.exec();

            System.out.println("事务执行成功！");
        } catch (Exception e) {
            e.printStackTrace();

            // 取消事务
            redisTemplate.discard();

            System.out.println("事务取消！");
        }
    }

    @Test
    public void test3() {           // 使用SessionCallback执行事务

        redisTemplate.execute(new SessionCallback<String>() {
            @Override
            public String execute(RedisOperations redisOperations) throws DataAccessException {

                try {
                    redisTemplate.opsForValue().set("count", 0);

                    // 开启key的监视
//            redisTemplate.watch("count");

                    // 开启事务
                    redisTemplate.multi();

                    // 自增操作
                    redisTemplate.opsForValue().increment("count", 1);

                    // 执行事务
                    redisTemplate.exec();

                    return "事务执行成功！";
                } catch (Exception e) {
                    e.printStackTrace();

                    // 取消事务
                    redisTemplate.discard();

                    return "事务取消！";
                }
            }
        });
    }
}

1.3 Redis 实现分布式锁

在并发编程中，我们通常通过锁来保证多个多线程操作同一资源的一致性，在Java中，提供有synchroized、Lock等来实现锁的功能，但Java中的锁只能保证在单体项目中（单个JVM进程中），无法应用在分布式环境下。

在分布式环境下，想要实现多个服务（多个进程）操作同一资源时，保证数据的一致性。可以使用redis来实现分布式锁。

1.3.1 Redis分布式锁场景

12306在售票时，是有多个售票窗口的，也就是售票的服务是有多个的，不管是在哪个售票微服卖出去的票都需要减掉库存中的余票，此时库存微服也是分布式的，即也存在多个减库存的微服接口，调用其中任意某个库存微服都可以实现减票操作。我们在单体项目时，只需将关键代码加上同步锁（多个售票窗口保证锁是同一个）。

伪代码如下：

public class Ticket implements Runnable {

    //票数
    private static Integer ticket = 100;

    //锁对象
    private static Object obj = new Object();

    @Override
    public void run() {

        while (true) {
            
            //加上同步代码块,把需要同步的代码放入代码块中,同步代码块中的锁对象必须保证一致！
            synchronized (obj) {
                
                if (ticket <= 0) {
                    break;      //票卖完了
                }
                System.out.println(Thread.currentThread().getName() + "正在卖第: " + (101 - ticket) + "张票");

                ticket--;
            }
        }
    }
}

售票窗口：

public class Demo1 {
    public static void main(String[] args) throws InterruptedException {
		
        Ticket ticket=new Ticket();
        //开启三个窗口,卖票
        Thread t1=new Thread(ticket,"南昌西站");
        Thread t2=new Thread(ticket,"南昌东站");
        Thread t3=new Thread(ticket,"南昌站");
        
        t1.start();
        t2.start();
        t3.start();
    }
}

在单体项目中，以上代码是可以保证票数不会被超卖。但是在分布式环境中无法保证，原因是分布式的每台机器上都有各自的锁。

我们需要将锁抽取出来，所有的微服公用同一把锁，谁拿到了锁，谁就可以操作资源。没有拿到锁的只能等待。

1.3.2 Redis实现分布式锁

• 设置分布式锁

setnx lock-key value

示例：

setnx ticket_lock 1

说明：设置一个key，如果成功返回1，失败返回0

分布式锁案例代码：

while(true){
    // 获取到了分布式锁才可以进行卖票操作            
    if("1".equals(redis.setnx("ticket_lock","1"))){
        if (ticket <= 0) {
            break;      //票卖完了
        }
        System.out.println(Thread.currentThread().getName() + "正在卖第: " + (101 - ticket) + "张票");

        ticket--;  
        redis.del("ticket_lock");
    }else{
        Thread.sleep(10);
    }
}

1.3.3 分布式锁改良

上面设置的分布式锁，可以保证分布式环境下的资源一致问题，但是有一个非常大的弊端，那就是如果获取到分布式锁的那个微服出现故障（如宕机），那么锁将永远不会释放，造成锁的永远阻塞，其他窗口再也卖不出任何的票，因此为了防止这种现象，我们可以给分布式锁设置一个过期时间，如果时间到了那么锁自动释放。

expire lock-key second
expire ticket_lock 5