SpringCloud 分布式锁

时间 2019-12-24

原文原文链接

A 简单实现：java

1、简介通常来讲，对数据进行加锁时，程序先经过acquire获取锁来对数据进行排他访问，而后对数据进行一些列的操做，最后须要释放锁。 Redis 自己用 watch命令进行了加锁，这个锁是乐观锁。使用 watch命令对于频繁访问的键会引发性能的问题。web

2、redis命令介绍 SETNX命令（SET if Not eXists）当且仅当 key 不存在，将 key 的值设为 value ，并返回1；若给定的 key 已经存在，则 SETNX 不作任何动做，并返回0。redis

SETEX命令设置超时时间算法

GET命令返回 key 所关联的字符串值，若是 key 不存在那么返回特殊值 null 。spring

DEL命令删除给定的一个或多个 key ,不存在的 key 会被忽略。apache

3、实现思路因为redis的setnx命令天生就适合用来实现锁的功能，这个命令只有在键不存在的状况下为键设置值。获取锁以后，其余程序再设置值就会失败，即获取不到锁。获取锁失败。只需不断的尝试获取锁，直到成功获取锁，或者到设置的超时时间为止。api

另外为了防治死锁，即某个程序获取锁以后，程序出错，没有释放，其余程序没法获取锁，从而致使整个分布式系统没法获取锁而致使一系列问题，甚至致使系统没法正常运行。这时须要给锁设置一个超时时间，即setex命令，锁超时后，从而其它程序就能够获取锁了。安全

4、编码实现本文采用springboot结合redis 取实现的，因此你须要装一个redis。springboot

首先pom.xml引入建立springboot工程，引入redis 。

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

        <!-- 开启web-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

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

2.建立一个锁类并发

/**
 * 全局锁，包括锁的名称
 */
public class Lock {
    private String name;
    private String value;

    public Lock(String name, String value) {
        this.name = name;
        this.value = value;
    }

    public String getName() {
        return name;
    }

    public String getValue() {
        return value;
    }

}

3.建立分布式锁的具体方法,思路已经说清楚了，代码注释也写好了，就不讲解了。

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 */
@Component
public class DistributedLockHandler {

    private static final Logger logger = LoggerFactory.getLogger(DistributedLockHandler.class);
    private final static long LOCK_EXPIRE = 30 * 1000L;//单个业务持有锁的时间30s，防止死锁
    private final static long LOCK_TRY_INTERVAL = 30L;//默认30ms尝试一次
    private final static long LOCK_TRY_TIMEOUT = 20 * 1000L;//默认尝试20s

    @Autowired
    private StringRedisTemplate template;

    /**
     * 尝试获取全局锁
     *
     * @param lock 锁的名称
     * @return true 获取成功，false获取失败
     */
    public boolean tryLock(Lock lock) {
        return getLock(lock, LOCK_TRY_TIMEOUT, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
    }

    /**
     * 尝试获取全局锁
     *
     * @param lock    锁的名称
     * @param timeout 获取超时时间 单位ms
     * @return true 获取成功，false获取失败
     */
    public boolean tryLock(Lock lock, long timeout) {
        return getLock(lock, timeout, LOCK_TRY_INTERVAL, LOCK_EXPIRE);
    }

    /**
     * 尝试获取全局锁
     *
     * @param lock        锁的名称
     * @param timeout     获取锁的超时时间
     * @param tryInterval 多少毫秒尝试获取一次
     * @return true 获取成功，false获取失败
     */
    public boolean tryLock(Lock lock, long timeout, long tryInterval) {
        return getLock(lock, timeout, tryInterval, LOCK_EXPIRE);
    }

    /**
     * 尝试获取全局锁
     *
     * @param lock           锁的名称
     * @param timeout        获取锁的超时时间
     * @param tryInterval    多少毫秒尝试获取一次
     * @param lockExpireTime 锁的过时
     * @return true 获取成功，false获取失败
     */
    public boolean tryLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
        return getLock(lock, timeout, tryInterval, lockExpireTime);
    }


    /**
     * 操做redis获取全局锁
     *
     * @param lock           锁的名称
     * @param timeout        获取的超时时间
     * @param tryInterval    多少ms尝试一次
     * @param lockExpireTime 获取成功后锁的过时时间
     * @return true 获取成功，false获取失败
     */
    public boolean getLock(Lock lock, long timeout, long tryInterval, long lockExpireTime) {
        try {
            if (StringUtils.isEmpty(lock.getName()) || StringUtils.isEmpty(lock.getValue())) {
                return false;
            }
            long startTime = System.currentTimeMillis();
            do{
                if (!template.hasKey(lock.getName())) {
                    ValueOperations<String, String> ops = template.opsForValue();
                    ops.set(lock.getName(), lock.getValue(), lockExpireTime, TimeUnit.MILLISECONDS);
                    return true;
                } else {//存在锁
                    logger.debug("lock is exist!！！");
                }
                if (System.currentTimeMillis() - startTime > timeout) {//尝试超过了设定值以后直接跳出循环
                    return false;
                }
                Thread.sleep(tryInterval);
            }
            while (template.hasKey(lock.getName())) ;
        } catch (InterruptedException e) {
            logger.error(e.getMessage());
            return false;
        }
        return false;
    }

    /**
     * 释放锁
     */
    public void releaseLock(Lock lock) {
        if (!StringUtils.isEmpty(lock.getName())) {
            template.delete(lock.getName());
        }
    }

}

4.用法：

@Autowired
DistributedLockHandler distributedLockHandler;
Lock lock=new Lock("lockk","sssssssss);
if(distributedLockHandler.tryLock(lock){
    doSomething();
    distributedLockHandler.releaseLock();
}

5、注意点

在使用全局锁时为了防止死锁采用 setex命令，这种命令须要根据具体的业务具体设置锁的超时时间。另一个就是锁的粒度性。好比在redis实战中有个案列，为了实现买卖市场交易的功能，把整个交易市场都锁住了，致使了性能不足的状况，改进方案只对买卖的商品进行加锁而不是整个市场。

B 优化实现:

setNx是一个耗时操做，由于它须要查询这个键是否存在，就算redis的百万的qps，在高并发的场景下，这种操做也是有问题的。关于redis实现分布式锁，redis官方推荐使用redlock。

1、redlock简介在不一样进程须要互斥地访问共享资源时，分布式锁是一种很是有用的技术手段。实现高效的分布式锁有三个属性须要考虑：

安全属性：互斥，无论何时，只有一个客户端持有锁效率属性A:不会死锁效率属性B：容错，只要大多数redis节点可以正常工做，客户端端都能获取和释放锁。

Redlock是redis官方提出的实现分布式锁管理器的算法，这个算法会比通常的普通方法更加安全可靠。

2、怎么用java使用 redlock 在pom文件引入redis和redisson依赖：

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

AquiredLockWorker接口类，，主要是用于获取锁后须要处理的逻辑：

/**
 * 获取锁后须要处理的逻辑
 */
public interface AquiredLockWorker<T> {
     T invokeAfterLockAquire() throws Exception;
}

DistributedLocker 获取锁管理类：

/**
 * 获取锁管理类
 */
public interface DistributedLocker {

     /**
      * 获取锁
      * @param resourceName  锁的名称
      * @param worker 获取锁后的处理类
      * @param <T>
      * @return 处理完具体的业务逻辑要返回的数据
      * @throws UnableToAquireLockException
      * @throws Exception
      */
     <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws UnableToAquireLockException, Exception;

     <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception;

}

UnableToAquireLockException ，不能获取锁的异常类：

/**
 * 异常类
 */
public class UnableToAquireLockException extends RuntimeException {

    public UnableToAquireLockException() {
    }

    public UnableToAquireLockException(String message) {
        super(message);
    }

    public UnableToAquireLockException(String message, Throwable cause) {
        super(message, cause);
    }
}

RedissonConnector 链接类：

/**
 * 获取RedissonClient链接类
 */
@Component
public class RedissonConnector {
    RedissonClient redisson;
    @PostConstruct
    public void init(){
        redisson = Redisson.create();
    }

    public RedissonClient getClient(){
        return redisson;
    }

}

RedisLocker 类，实现了DistributedLocker：

import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.TimeUnit;

@Component
public class RedisLocker  implements DistributedLocker{

    private final static String LOCKER_PREFIX = "lock:";

    @Autowired
    RedissonConnector redissonConnector;
    @Override
    public <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws InterruptedException, UnableToAquireLockException, Exception {

        return lock(resourceName, worker, 100);
    }

    @Override
    public <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception {
        RedissonClient redisson= redissonConnector.getClient();
        RLock lock = redisson.getLock(LOCKER_PREFIX + resourceName);
      // Wait for 100 seconds seconds and automatically unlock it after lockTime seconds
        boolean success = lock.tryLock(100, lockTime, TimeUnit.SECONDS);
        if (success) {
            try {
                return worker.invokeAfterLockAquire();
            } finally {
                lock.unlock();
            }
        }
        throw new UnableToAquireLockException();
    }
}

测试类：

@Autowired
    RedisLocker distributedLocker;
    @RequestMapping(value = "/redlock")
    public String testRedlock() throws Exception{

        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(5);
        for (int i = 0; i < 5; ++i) { // create and start threads
            new Thread(new Worker(startSignal, doneSignal)).start();
        }
        startSignal.countDown(); // let all threads proceed
        doneSignal.await();
        System.out.println("All processors done. Shutdown connection");
        return "redlock";
    }

     class Worker implements Runnable {
        private final CountDownLatch startSignal;
        private final CountDownLatch doneSignal;

        Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
            this.startSignal = startSignal;
            this.doneSignal = doneSignal;
        }

        public void run() {
            try {
                startSignal.await();
                distributedLocker.lock("test",new AquiredLockWorker<Object>() {

                    @Override
                    public Object invokeAfterLockAquire() {
                        doTask();
                        return null;
                    }

                });
            }catch (Exception e){

            }
        }

        void doTask() {
            System.out.println(Thread.currentThread().getName() + " start");
            Random random = new Random();
            int _int = random.nextInt(200);
            System.out.println(Thread.currentThread().getName() + " sleep " + _int + "millis");
            try {
                Thread.sleep(_int);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " end");
            doneSignal.countDown();
        }
    }