redis分布式锁(基于springboot实现)

Separes

 关注

阅读 39

2021-09-18

分布式锁应该具备哪些条件

  • 在分布式系统环境下，一个方法在同一时间只能被一个机器的一个线程执行
  • 高可用的获取锁与释放锁
  • 高性能的获取锁与释放锁
  • 具备可重入特性（可理解为重新进入，由多于一个任务并发使用，而不必担心数据错误）
  • 具备锁失效机制，防止死锁
  • 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

分布式锁的实现方式

  1. 基于数据库乐观锁/悲观锁
  2. Redis分布式锁（本文）：利用setnx命令。此命令是原子性操作，只有key不存在的情况下，才能set，就意味着线程获取到了锁
  3. Zookeeper分布式锁：利用 Zookeeper 的顺序临时节点，来实现分布式锁和等待队列。Zookeeper 设计的初衷，就是为了实现分布式锁服务的
  4. Memcached：利用add命令。此命令是原子性操作，只有key不存在的情况下，才能add，也就意味着线程获取到了锁

redis是如何实现加锁的？

在redis中，有一条命令，实现锁

SETNX key value

该命令的作用是将 key 的值设为 value ，当且仅当 key 不存在。若给定的 key 已经存在，则 SETNX 不做任何动作。设置成功，返回 1 ；设置失败，返回 0

使用 redis 来实现锁的逻辑就是这样的

线程 1 获取锁  -- > setnx lockKey lockvalue
              -- >  1  获取锁成功
线程 2 获取锁  -- > setnx lockKey lockvalue 
              -- >  0  获取锁失败  （继续等待，或者其他逻辑）
线程 1 释放锁  -- > 
线程 2 获取锁  -- > setnx lockKey lockvalue
              -- > 1 获取成功

接下来我们将基于springboot实现redis分布式锁

1. 引入redis、springmvc、lombok依赖

<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.0</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>cn.miao.redis</groupId>
    <artifactId>springboot-caffeine-demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>springboot-redis-lock-demo</name>
    <description>Demo project for Redis Distribute Lock</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

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

        <!--redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
            <version>2.1.4.RELEASE</version>
        </dependency>

        <!--springMvc-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.3.3.RELEASE</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.projectlombok/lombok -->
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.12</version>
            <scope>provided</scope>
        </dependency>

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

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

2. 新建RedisDistributedLock.java并书写加锁解锁逻辑

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;

import java.nio.charset.StandardCharsets;

/**
 * @author miao
 * redis 加锁工具类
 */
@Slf4j
public class RedisDistributedLock {

    /**
     * 超时时间
     */
    private static final long TIMEOUT_MILLIS = 15000;

    /**
     * 重试次数
     */
    private static final int RETRY_TIMES = 10;

    /***
     * 睡眠时间
     */
    private static final long SLEEP_MILLIS = 500;

    /**
     * 用来加锁的lua脚本
     * 因为新版的redis加锁操作已经为原子性操作
     * 所以放弃使用lua脚本
     */
    private static final String LOCK_LUA =
            "if redis.call(\"setnx\",KEYS[1],ARGV[1]) == 1 " +
                    "then " +
                    "    return redis.call('expire',KEYS[1],ARGV[2]) " +
                    "else " +
                    "    return 0 " +
                    "end";

    /**
     * 用来释放分布式锁的lua脚本
     * 如果redis.get(KEYS[1]) == ARGV[1],则redis delete KEYS[1]
     * 否则返回0
     * KEYS[1] , ARGV[1] 是参数，我们只调用的时候 传递这两个参数就可以了
     * KEYS[1] 主要用來传递在redis 中用作key值的参数
     * ARGV[1] 主要用来传递在redis中用做 value值的参数
     */
    private static final String UNLOCK_LUA =
            "if redis.call(\"get\",KEYS[1]) == ARGV[1] "
                    + "then "
                    + "    return redis.call(\"del\",KEYS[1]) "
                    + "else "
                    + "    return 0 "
                    + "end ";

    /**
     * 检查 redisKey 是否上锁
     *
     * @param redisKey redisKey
     * @param template template
     * @return Boolean
     */
    public static Boolean isLock(String redisKey, String value, RedisTemplate<Object, Object> template) {

        return lock(redisKey, value, template, RETRY_TIMES);
    }

    private static Boolean lock(String redisKey,
                                String value,
                                RedisTemplate<Object, Object> template,
                                int retryTimes) {

        boolean result = lockKey(redisKey, value, template);

        while (!(result) && retryTimes-- > 0) {
            try {

                log.debug("lock failed, retrying...{}", retryTimes);
                Thread.sleep(RedisDistributedLock.SLEEP_MILLIS);
            } catch (InterruptedException e) {

                return false;
            }
            result = lockKey(redisKey, value, template);
        }

        return result;
    }


    private static Boolean lockKey(final String key,
                                   final String value,
                                   RedisTemplate<Object, Object> template) {
        try {

            RedisCallback<Boolean> callback = (connection) -> connection.set(
                    key.getBytes(StandardCharsets.UTF_8),
                    value.getBytes(StandardCharsets.UTF_8),
                    Expiration.milliseconds(RedisDistributedLock.TIMEOUT_MILLIS),
                    RedisStringCommands.SetOption.SET_IF_ABSENT
            );

            return template.execute(callback);
        } catch (Exception e) {

            log.info("lock key fail because of ", e);
        }

        return false;
    }


    /**
     * 释放分布式锁资源
     *
     * @param redisKey key
     * @param value    value
     * @param template redis
     * @return Boolean
     */
    public static Boolean releaseLock(String redisKey,
                                      String value,
                                      RedisTemplate<Object, Object> template) {
        try {
            RedisCallback<Boolean> callback = (connection) -> connection.eval(
                    UNLOCK_LUA.getBytes(),
                    ReturnType.BOOLEAN,
                    1,
                    redisKey.getBytes(StandardCharsets.UTF_8),
                    value.getBytes(StandardCharsets.UTF_8)
            );

            return template.execute(callback);
        } catch (Exception e) {

            log.info("release lock fail because of ", e);
        }

        return false;
    }

}

补充：

1. spring-data-redis 有StringRedisTempla和RedisTemplate两种，但是我选择了RedisTemplate，因为他比较万能。他们的区别是：当你的redis数据库里面本来存的是字符串数据或者你要存取的数据就是字符串类型数据的时候，那么你就使用StringRedisTemplate即可， 但是如果你的数据是复杂的对象类型，而取出的时候又不想做任何的数据转换，直接从Redis里面取出一个对象，那么使用RedisTemplate是 更好的选择。
2. 选择lua脚本是因为，脚本运行是原子性的，在脚本运行期间没有客户端可以操作，所以在释放锁的时候用了lua脚本，
而redis最新版加锁时保证了Redis值和自动过期时间的原子性，所用没用lua脚本

3. 创建测试类 TestController 

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @author miao
 */
@RestController
@Slf4j
public class TestController {

    @Resource
    private RedisTemplate<Object, Object> redisTemplate;

    @PostMapping("/order")
    public String createOrder() throws InterruptedException {

        log.info("开始创建订单");

        Boolean isLock = RedisDistributedLock.isLock("testLock", "456789", redisTemplate);

        if (!isLock) {

            log.info("锁已经被占用");
            return "fail";
        } else {
            //.....处理逻辑
        }

        Thread.sleep(10000);
        //一定要记得释放锁，否则会出现问题
        RedisDistributedLock.releaseLock("testLock", "456789", redisTemplate);

        return "success";
    }
}

4. 使用postman进行测试

5. redis分布式锁的缺点

6.redis分布式锁的优化

** 分布式锁对比**

数据库分布式锁实现

缺点：

  1. db操作性能较差，并且有锁表的风险
  2. 非阻塞操作失败后，需要轮询，占用cpu资源
  3. 长时间不commit或者长时间轮询，可能会占用较多连接资源

Redis分布式锁实现

缺点：

  1. 锁删除失败 过期时间不好控制
  2. 非阻塞，操作失败后，需要轮询，占用cpu资源

ZK分布式锁实现

缺点：

  1. 性能不如redis实现，主要原因是写操作（获取锁释放锁）都需要在Leader上执行，然后同步到follower。

总结

从理解的难易程度角度（从低到高）数据库 > Redis >Zookeeper
从实现的复杂性角度（从低到高）Zookeeper >= Redis > 数据库
从性能角度（从高到低）缓存 > Zookeeper >= 数据库
从可靠性角度（从高到低）Zookeeper > Redis > 数据库

相关推荐

基于 SpringBoot + Redis 实现分布式锁

sunflower821 106 0 0

分布式锁2：基于redis实现分布式锁

回望这一段人生 95 0 0

分布式锁实现方案-基于Redis实现的分布式锁

曾宝月 49 0 0

分布式锁之基于redis实现分布式锁（二）

冬冬_79d4 39 0 0

springboot实现分布式锁，java实现分布式锁，redisson实现分布式锁

单调先生 134 0 0

基于redis实现的分布式锁

扒皮狼 189 0 0

SpringBoot基于Zookeeper实现分布式锁

小亦同学321 51 0 0

springboot+redis+lua实现分布式锁

科牛 25 0 0

【Redis】Redis实现分布式锁

婉殇成长笔记 28 0 0

基于SpringBoot AOP面向切面编程实现Redis分布式锁

十里一走马 84 0 0

精彩评论（0）

0 0 举报