redis 系列之——分布式锁

 关注
发布于: 2020 年 07 月 09 日
Redis系列目录﻿
redis系列之——分布式锁
﻿
redis系列之——缓存穿透、缓存击穿、缓存雪崩
﻿
redis系列之——Redis为什么这么快？
﻿
redis系列之——数据持久化（RDB和AOF）
﻿
redis系列之——一致性hash算法
﻿
redis系列之——高可用（主从、哨兵、集群）
﻿
redis系列之——事物及乐观锁
﻿
redis系列之——数据类型geospatial：你隔壁有没有老王？
﻿
redis系列之——数据类型bitmaps：今天你签到了吗？
﻿
布隆过滤器是个啥！
﻿
端午节最后一天了，三天假期过得太快了，更坏的消息是，下周还要上六天班呢！
﻿
﻿
趁着假期的尾巴，和大家再叨逼叨redis是如何实现分布式锁的。这期也是为了填之前《redis系列之——缓存穿透、缓存击穿、缓存雪崩》留下的坑。
﻿
这一期不是专门聊分布式锁的，所以不会涉及到各种分布式锁实现及相关的比较，只是聊一下如何使用redis实现分布式锁。感觉上一个坑还没填完，这里又挖了一个大坑，各种分布式锁实现及相关的比较后面有时间我会跪着填上的，请大家给我多一点点时间，多一点点温柔。
﻿
本期是硬核输出，也是面试高频，更是实战必会。
﻿
什么是分布式锁﻿
先说一个场景，消费者在购物网站上下单或收银员在POS机上下单，由于网络等问题，在连续点击了两下，后端网站如何处理，如何响应？对于这个问题，前端需要处理，后端也需要处理。这里主要说后端，后端不光要处理重复订单问题，还有处理幂等问题。幂等问题简单来说就是相同的请求，要有相同的响应结果，这里就不展开了。重复订单该如何处理？
﻿
对于一个小的访问量不大的网站，部署了一个tomcat，这个问题可以简单的通过JVM提供的同步锁synchronized实现。但是当网站访问量越来越大时，需要扩展机器，synchronized就不能起作用了。相同的下单参数连续两次请求后端服务器，可能会被分发到两个tomcat上，就会出现synchronized失效问题。
﻿
﻿
分布式锁要解决的就是多机器部署时，相同请求并发访问时资源竞争问题。请求到达每个tomcat时，首先要去redis中注册锁，注册成功返回true则说明获得了锁，可以继续处理相关的业务，处理完成后释放锁。同一时刻只能有一个tomcat能获得锁，其他没获得锁的tomcat则多次尝试继续获得锁，没有获得锁不能处理业务。获得锁的tomcat释放锁后，其他的tomcat才能有一个获得锁。
﻿
﻿
这里是使用redis做外部存储介质存储锁的，使用zookeeper也是类似的。万变不离其宗，原理都一样，只是技术选型有差别。
﻿
redis实现﻿
废话就不说了，直接上代码。
﻿
1.pom.xml﻿
     <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.3.1.RELEASE</version>
     </parent>   
     <dependency>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-data-redis</artifactId>
     </dependency>
﻿
这里需要注意，我使用的是spring-boot。redis的相关jar使用的是spring-boot-starter-data-redis。
﻿
注意版本是2以上，版本1和版本2在redis锁的实现上有个关键的差异，后面会说到。
﻿
2.application.yml﻿
server:
  port: 8080
spring:
  application:
    name: java-summary
  redis:
    database: 10
    password: 123456
    timeout: 20000
    host: 127.0.0.1
    port: 6379
    pool:
      max-idle: 20
      min-idle: 20
      max-wait: 10000
      max-active: 5000
﻿
3.RedisConfiguration.java﻿
package com.wuxiaolong.redis;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
 * Description:
 *
 * @author 诸葛小猿
 * @date 2020-06-27
 *
 * 去除redis序列化时，key-value的乱码问题
 */
@Configuration
public class RedisConfiguration {
    @Bean
    @Primary
	public RedisTemplate setRedisTemplate(RedisTemplate redisTemplate) {
		RedisSerializer stringSerializer = new StringRedisSerializer();
		redisTemplate.setKeySerializer(stringSerializer);
		redisTemplate.setValueSerializer(stringSerializer);
		redisTemplate.setHashKeySerializer(stringSerializer);
		redisTemplate.setHashValueSerializer(stringSerializer);
		return redisTemplate;
	}
}
﻿
这个配置类就是为了解决redis的key和value在序列化时的乱码问题，将序列化的方式设置为string格式
﻿
4.RedisLock.java﻿
package com.wuxiaolong.redis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
/**
 * Description:
 *
 * @author 诸葛小猿
 * @date 2020-06-27
 */
@Component
@Slf4j
public class RedisLock {
    @Autowired
    private RedisTemplate redisTemplate;
    /**
     * 添加元素
     *
     * @param key
     * @param value
     */
    public void set(String key, String value) {
        if (key == null || value == null) {
            return;
        }
        redisTemplate.opsForValue().set(key, value);
    }
    /**
     * 如果已经存在返回false，否则返回true
     *
     * @param key
     * @param value
     * @return
     */
    public Boolean setNx(String key, String value, Long expireTime, TimeUnit mimeUnit) {
        if (key == null || value == null) {
            return false;
        }
        //spiring boot 1.5 版本setIfAbsent只能设置key-value,需要单独对key设置过期时间，因为是两步操作，所以不是原子性
        //Boolean tf =  redisTemplate.opsForValue().setIfAbsent(key, value);
        //redisTemplate.expire(key, expireTime, mimeUnit);
        // 在spiring boot 2 可以直接使用 redisTemplate的setIfAbsent设置key-value和过期时间，是原子性
        Boolean tf =redisTemplate.opsForValue().setIfAbsent(key,value,expireTime, mimeUnit);
        return tf;
    }
    /**
     * 获取数据
     *
     * @param key
     * @return
     */
    public Object get(String key) {
        if (key == null) {
            return null;
        }
        return redisTemplate.opsForValue().get(key);
    }
    /**
     * 删除
     *
     * @param key
     * @return
     */
    public void remove(Object key) {
        if (key == null) {
            return ;
        }
        redisTemplate.delete(key);
    }
    /**
     * 加锁
     *
     * @param key
     * @param waitTime 等待时间,在这个时间内会多次尝试获取锁，超过这个时间还没获得锁，就返回false
     * @param interval 间隔时间,每隔多长时间尝试一次获的锁
     * @param expireTime key的过期时间
     */
    public Boolean lock(String key, Long waitTime,Long interval, Long expireTime) {
        String value = UUID.randomUUID().toString().replaceAll("-", "").toLowerCase();
        Boolean flag = setNx(key, value, expireTime, TimeUnit.MILLISECONDS);
        // 尝试获取锁 成功返回
        if (flag) {
            return flag;
        } else {
            // 获取失败
            // 现在时间
            long newTime = System.currentTimeMillis();
            // 等待过期时间
            long loseTime = newTime + waitTime;
            // 不断尝试获取锁成功返回
            while (System.currentTimeMillis() < loseTime) {
                Boolean testFlag = setNx(key, value, expireTime, TimeUnit.MILLISECONDS);
                if (testFlag) {
                    return testFlag;
                }
                try {
                    Thread.sleep(interval);
                } catch (InterruptedException e) {
                    log.error("获取锁异常",e);
                }
            }
        }
        return false;
    }
    /**
     * 释放锁
     *
     * @param key
     * @return
     */
    public void unLock(String key) {
         remove(key);
    }
    public Boolean setIfAbsent(String key, String value){
        Boolean tf =  redisTemplate.opsForValue().setIfAbsent(key, value);
        redisTemplate.expire(key, 60, TimeUnit.DAYS);
        return tf;
    }
}
﻿
这个就是redis锁的核心实现。上面提到的spring-boot版本问题就是这里的setNx方法中的redisTemplate的setIfAbsent的差异。
﻿
setIfAbsent的作用是，在存储key和value时，如果key不存在则保存后返回true，说明获得锁；如果key存在则保存后返回false，说明这个锁正在使用中，不能获取。
﻿
在获取锁时，有两步操作，首先是要存储这个key-value，然后需要对其设置一个过期时间，防止出现死锁。在spring-boot的版本为1时，只提供了setIfAbsent(key, value)的API，设置成功后需要调用redisTemplate.expire(key, expireTime, mimeUnit)对这个key设置过期时间，这是两步是非原子性的操作，如果第一步执行成功，第二步执行失败，就可能出现死锁，虽然概率很低。在spring-boot的版本为2时，提供了一个原子性的API setIfAbsent(key,value,expireTime, mimeUnit)，在保存key-value的同时设置了过期时间，推荐使用。
﻿
4.RedisLockTest.java﻿
package com.wuxiaolong.redis;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
/**
 * Description:
 *
 * @author 诸葛小猿
 * @date 2020-06-27
 */
@RestController
@Slf4j
public class RedisLockTest {
    @Autowired
    private RedisLock redisLock;
    public static final String ORDER_LOCK_PREFIX = "order:lock:";
    @RequestMapping(value = "save/order",method = RequestMethod.POST)
    public String saveOrder(@RequestBody Map<String,String> order){
        // 获取商户号、门店号、收银机号、收银单号，组成全局唯一的订单号
        String orderSn = order.get("merchantSn")+order.get("storeSn")+order.get("workstationSn")+order.get("checkSn");
        // 使用全局唯一的订单号做分布式锁的key
        String lockKey = ORDER_LOCK_PREFIX + orderSn;
        try{
            // 加锁 每隔100毫秒获取一次锁；1分钟内拿不到锁就返回false；锁的过期时间5分钟（防止死锁）
            Boolean tf = redisLock.lock(lockKey,1L * 60 * 1000,100L,5L * 60 * 1000);
            if(tf){
                // todo 处理拿到锁时的业务
                return "success";
            }else {
                // todo 处理没有拿到锁时的业务
                return "fail";
            }
        }catch (Exception e){
            log.error("业务异常",e);
        }finally {
            // 一定要在finally中释放锁
            redisLock.unLock(lockKey);
        }
        return "success";
    }
}
﻿
这是测试类的实例。首先需要获得全局唯一的订单号，然后使用这个订单号为key。在try块的最前面先去redis中获得锁，根据是否获得锁做不同的操作。需要根据自己的业务定义waitTime,interval, expireTime这三个参数。一定要在最后的finally中释放锁。
﻿
完成，收工！！
﻿
﻿
【传播知识，共享价值】，感谢小伙伴们的关注和支持，我是【诸葛小猿】，一个彷徨中奋斗的互联网民工。
﻿
﻿