分布式锁设计与实践

2022/1/27 6:04:25

编程Tag： 设计 return 获取节点 param 实践分布式 lock public

本文主要是介绍分布式锁设计与实践，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

分布式锁设计与实践

分布式锁是什么

分布式锁即在分布式环境下锁定共享资源，让请求处理串行化，实际表现为互斥锁。分布式锁可以解决业务中的幂等性问题。

分布式锁是主要是可重入的排它锁，主要具有如下特点：

1）释放锁的节点，一定是持有锁的节点。

2）持有锁的节点宕机了，不会因此而出现死锁，就像synchronized一样，当持有锁的线程异常退出了，锁会被自动释放。

3）排他性，很好理解，一个节点持有锁，其他节点就不能持有了。

4）可重入性，很好理解，持有锁的节点，可以重复持有锁。

分布式锁应用场景

两个常见的应用场景：（1）防止用户重复下单（2）mq消息去重

防止用户重复下单

正常情况下用户短时间内肯定是只会下单一次的，但是如果出现了超时用户点击了多次，那么如果前端的同学也没有控制住，这个时候如果服务层面不做任何控制就会出现同时下单多笔的情况，这个肯定是不正常的。当然如果在我们的服务没做水平扩容前我们加个本地锁就搞定了，但是如果我们的服务做了水平扩容:

这个时候就算我们的服务层面有本地锁，也还是不能解决问题的。比如第一次下单请求落在业务逻辑层1，而后面的重复下单落在业务逻辑层2、业务逻辑层3、业务逻辑层N，那我们的重复下单就真的重复下单了，所以这个时候我们需要分布式锁来解决这个问题:

这样每次下单请求过来的时候我们可以使用UserId作为关键信息先去分布式锁服务拿锁，如果能拿到请求继续执行，如果拿不到(被其他进程或线程占用)就丢弃掉当前请求。

MQ消息去重

还是用户下单的操作，这一次我们为了提升吞度量采用了水平分层异步架构，用户进行下单操作，这里我假设在网关层生成了订单号(正常情况下应该是专门的订单服务做处理)，然后把请求丢到MessageQueue。这种场景下我们如果要对MQ消息去重，一般我们会有两种想法:

发送端去重

消费端去重

我们先来分析发送端去重，如果第一次发送端生成了消息丢到MQ，正常情况下MQ是会返回一个Ack给发送端的，那如果超时了，这个时候发送端重试又生成一个消息丢到MQ，我们可以思考下这种情况下我们能做到去重吗？肯定是做不到的吧，因为2次生成消息就会生成2个不同的MsgId。那我们再来看看消费端去重，其实这个时候我们要想既然MsgId我们不能唯一识别，那什么是我们能唯一识别的？肯定是订单号吧，那我们只要在消费端对订单号加锁就ok了吧。同样的如果没有做水平扩容的情况下我们本地锁是不是就可以搞定？但是水平扩容之后呢?

同样的道理，如果消息第一次被业务逻辑层1消费，后面重复的消息被业务逻辑层2、业务逻辑层3、业务逻辑层N消费，本地锁在这种情况下肯定搞不定。我们也需要分布式锁来做这个事:

这样每次消费消息的时候我们可以使用OrderId作为关键信息先去分布式锁服务拿锁，如果能拿到请求继续执行，如果拿不到(被其他进程或线程占用)就丢弃掉当前消息。

这里有一个需要注意的事情，我们的分布式锁只能处理短时间内的重复请求或者消息，如果过了锁的时间，那锁是无法处理的，这个时候怎么办呢？很简单，使用幂等处理。

通过分析上面2种业务场景，我们可以得出一个结论。这两种场景存在的共性是要处理共享的资源（订单、消息），那解决方案也就是实现共享资源的互斥，共享资源的串行化，那最后转换出来就是锁的问题了，锁又有本地锁和分布式锁。在分布式环境下，本地锁已经不奏效了，只能使用分布式锁。

分布式锁设计与实现
1. redis实现分布式锁

使用redis唯一工作线程来处理

利用 SETNX 和 SETEX

基本命令主要有：

SETNX(SET If Not Exists)：当且仅当 Key 不存在时，则可以设置，否则不做任何动作。

SETEX：可以设置超时时间

其原理为：通过 SETNX 设置 Key-Value 来获得锁，随即进入死循环，每次循环判断，如果存在 Key 则继续循环，如果不存在 Key，则跳出循环，当前任务执行完成后，删除 Key 以释放锁。

这种方式可能会导致死锁，为了避免这种情况，需要设置超时时间。

下面，请看具体的实现步骤。

1.创建一个 Maven 工程并在 pom.xml 加入以下依赖：

 1 <parent>
 2         <groupId>org.springframework.boot</groupId>
 3         <artifactId>spring-boot-starter-parent</artifactId>
 4         <version>2.0.2.RELEASE</version>
 5         <relativePath/> <!-- lookup parent from repository -->
 6     </parent>
 7 
 8     <properties>
 9         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
10         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
11         <java.version>1.8</java.version>
12     </properties>
13 
14     <dependencies>
15         <dependency>
16         <groupId>org.springframework.boot</groupId>
17         <artifactId>spring-boot-starter-test</artifactId>
18         <scope>test</scope>
19         </dependency>
20 
21         <!-- 开启web-->
22         <dependency>
23             <groupId>org.springframework.boot</groupId>
24             <artifactId>spring-boot-starter-web</artifactId>
25         </dependency>
26 
27         <!-- redis-->
28         <dependency>
29             <groupId>org.springframework.boot</groupId>
30             <artifactId>spring-boot-starter-data-redis</artifactId>
31         </dependency>
32     </dependencies>

2.创建启动类 Application.java：

1 @SpringBootApplication
2 public class Application {
3 
4     public static void main(String[] args) {
5         SpringApplication.run(Application.class,args);
6     }
7 
8 }

3.添加配置文件 application.yml：

1 server:
2   port: 8080
3 spring:
4   redis:
5     host: localhost
6     port: 6379

4.创建全局锁类 Lock.java：

/**
 * 全局锁，包括锁的名称
 */
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;
    }

}

5.创建分布式锁类 DistributedLockHandler.java：

@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());
        }
    }

}

6.最后创建 HelloController 来测试分布式锁。

@RestController
public class LockController {

    @Autowired
    private DistributedLockHandler distributedLockHandler;

    @RequestMapping("/index")
    public String index(){
        Lock lock=new Lock("lynn","min");
        if(distributedLockHandler.tryLock(lock)){
            try {
                //为了演示锁的效果，这里睡眠5000毫秒
                System.out.println("执行方法");
                Thread.sleep(5000);
            }catch (Exception e){
                e.printStackTrace();
            }
            distributedLockHandler.releaseLock(lock);
        }
        return "hello world!";
    }
}

7.测试。

启动 Application.java，连续访问两次浏览器：http://localhost:8080/index，控制台可以发现先打印了一次"执行方法"，说明后面一个线程被锁住了，5秒后又再次打印了"执行方法"，说明锁被成功释放。

通过这种方式创建的分布式锁存在以下问题：

高并发的情况下，如果两个线程同时进入循环，可能导致加锁失败。

SETNX 是一个耗时操作，因为它需要判断 Key 是否存在，因为会存在性能问题。

因此，Redis 官方推荐 Redlock 来实现分布式锁。

使用Redlock

通过 Redlock 实现分布式锁比其他算法更加可靠，继续改造上一例的代码。

1.pom.xml 增加以下依赖：

<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.7.0</version>
</dependency>

2.增加以下几个类：

 1 /**
 2  * 获取锁后需要处理的逻辑
 3  */
 4 public interface AquiredLockWorker<T> {
 5     T invokeAfterLockAquire() throws Exception;
 6 }
 7 /**
 8  * 获取锁管理类
 9  */
10 public interface DistributedLocker {
11 
12     /**
13      * 获取锁
14      * @param resourceName  锁的名称
15      * @param worker 获取锁后的处理类
16      * @param <T>
17      * @return 处理完具体的业务逻辑要返回的数据
18      * @throws UnableToAquireLockException
19      * @throws Exception
20      */
21     <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws UnableToAquireLockException, Exception;
22 
23     <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception;
24 
25 }
26 /**
27  * 异常类
28  */
29 public class UnableToAquireLockException extends RuntimeException {
30 
31     public UnableToAquireLockException() {
32     }
33 
34     public UnableToAquireLockException(String message) {
35         super(message);
36     }
37 
38     public UnableToAquireLockException(String message, Throwable cause) {
39         super(message, cause);
40     }
41 }
42 /**
43  * 获取RedissonClient连接类
44  */
45 @Component
46 public class RedissonConnector {
47     RedissonClient redisson;
48     @PostConstruct
49     public void init(){
50         redisson = Redisson.create();
51     }
52 
53     public RedissonClient getClient(){
54         return redisson;
55     }
56 
57 }
58 @Component
59 public class RedisLocker  implements DistributedLocker{
60 
61     private final static String LOCKER_PREFIX = "lock:";
62 
63     @Autowired
64     RedissonConnector redissonConnector;
65     @Override
66     public <T> T lock(String resourceName, AquiredLockWorker<T> worker) throws InterruptedException, UnableToAquireLockException, Exception {
67 
68         return lock(resourceName, worker, 100);
69     }
70 
71     @Override
72     public <T> T lock(String resourceName, AquiredLockWorker<T> worker, int lockTime) throws UnableToAquireLockException, Exception {
73         RedissonClient redisson= redissonConnector.getClient();
74         RLock lock = redisson.getLock(LOCKER_PREFIX + resourceName);
75         // Wait for 100 seconds seconds and automatically unlock it after lockTime seconds
76         boolean success = lock.tryLock(100, lockTime, TimeUnit.SECONDS);
77         if (success) {
78             try {
79                 return worker.invokeAfterLockAquire();
80             } finally {
81                 lock.unlock();
82             }
83         }
84         throw new UnableToAquireLockException();
85     }
86 }

3.修改LockController：

@RestController
public class HelloController {

    @Autowired
    private DistributedLocker distributedLocker;

    @RequestMapping("index")
    public String index()throws Exception{
        distributedLocker.lock("test",new AquiredLockWorker<Object>() {

            @Override
            public Object invokeAfterLockAquire() {
                try {
                    System.out.println("执行方法！");
                    Thread.sleep(5000);
                }catch (Exception e){
                    e.printStackTrace();
                }
                return null;
            }

        });
        return "hello world!";
    }
}

4.按照上节的测试方法进行测试，我们发现分布式锁也生效了。

Redlock 是 Redis 官方推荐的一种方案，因此可靠性比较高。

Zookeeper实现分布式锁
1. 节点概念

让我们来回顾一下Zookeeper节点的概念：

Zookeeper的数据存储结构就像一棵树，这棵树由节点组成，这种节点叫做Znode。

Znode分为四种类型：

1.持久节点（PERSISTENT）

默认的节点类型。创建节点的客户端与zookeeper断开连接后，该节点依旧存在。

2.持久节点顺序节点（PERSISTENT_SEQUENTIAL）

所谓顺序节点，就是在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号：

3.临时节点（EPHEMERAL）

和持久节点相反，当创建节点的客户端与zookeeper断开连接后，临时节点会被删除：

4.临时顺序节点（EPHEMERAL_SEQUENTIAL）

顾名思义，临时顺序节点结合和临时节点和顺序节点的特点：在创建节点时，Zookeeper根据创建的时间顺序给该节点名称进行编号；当创建节点的客户端与zookeeper断开连接后，临时节点会被删除。

zookeeper分布式锁的原理

Zookeeper分布式锁恰恰应用了临时顺序节点。具体如何实现呢？让我们来看一看详细步骤：

获取锁

首先，在Zookeeper当中创建一个持久节点ParentLock。当第一个客户端想要获得锁时，需要在ParentLock这个节点下面创建一个临时顺序节点 Lock1。

之后，Client1查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock1是不是顺序最靠前的一个。如果是第一个节点，则成功获得锁。

这时候，如果再有一个客户端 Client2 前来获取锁，则在ParentLock下载再创建一个临时顺序节点Lock2。

Client2查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock2是不是顺序最靠前的一个，结果发现节点Lock2并不是最小的。这意味着Client2抢锁失败，进入了等待状态。

于是，Client2向排序仅比它靠前的节点Lock1注册Watcher，用于监听Lock1节点是否存在。

这时候，如果又有一个客户端Client3前来获取锁，则在ParentLock下载再创建一个临时顺序节点Lock3。

Client3查找ParentLock下面所有的临时顺序节点并排序，判断自己所创建的节点Lock3是不是顺序最靠前的一个，结果同样发现节点Lock3并不是最小的。

于是，Client3向排序仅比它靠前的节点Lock2注册Watcher，用于监听Lock2节点是否存在。这意味着Client3同样抢锁失败，进入了等待状态。

这样一来，Client1得到了锁，Client2监听了Lock1，Client3监听了Lock2。这恰恰形成了一个等待队列，很像是Java当中ReentrantLock所依赖的

释放锁

释放锁分为两种情况：

1.任务完成，客户端显示释放

当任务完成时，Client1会显示调用删除节点Lock1的指令。

2.任务执行过程中，客户端崩溃

获得锁的Client1在任务执行过程中，如果Duang的一声崩溃，则会断开与Zookeeper服务端的链接。根据临时节点的特性，相关联的节点Lock1会随之自动删除。

由于Client2一直监听着Lock1的存在状态，当Lock1节点被删除，Client2会立刻收到通知。这时候Client2会再次查询ParentLock下面的所有节点，确认自己创建的节点Lock2是不是目前最小的节点。如果是最小，则Client2顺理成章获得了锁。

同理，如果Client2也因为任务完成或者节点崩溃而删除了节点Lock2，那么Client3就会接到通知。

最终，Client3成功得到了锁。

代码实现

创建 DistributedLock 类：

public class DistributedLock implements Lock, Watcher{

private ZooKeeper zk;

private String root = "/ParentLock";//根

private String lockName;//竞争资源的标志

private String waitNode;//等待前一个锁

private String myZnode;//当前锁

private CountDownLatch latch;//计数器

private CountDownLatch connectedSignal=new CountDownLatch(1);

private int sessionTimeout = 30000;

/**

* 创建分布式锁,使用前请确认config配置的zookeeper服务可用

* @param config localhost:2181

* @param lockName 竞争资源标志,lockName中不能包含单词_lock_

*/

public DistributedLock(String config, String lockName){

this.lockName = lockName;

// 创建一个与服务器的连接

try {

zk = new ZooKeeper(config, sessionTimeout, this);

connectedSignal.await();

Stat stat = zk.exists(root, false);//此去不执行 Watcher

if(stat == null){

// 创建根节点

zk.create(root, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);

}

} catch (IOException e) {

throw new LockException(e);

} catch (KeeperException e) {

throw new LockException(e);

} catch (InterruptedException e) {

throw new LockException(e);

}

}

/**

* zookeeper节点的监视器

*/

public void process(WatchedEvent event) {

//建立连接用

if(event.getState()== Event.KeeperState.SyncConnected){

connectedSignal.countDown();

return;

}

//其他线程放弃锁的标志

if(this.latch != null) {

this.latch.countDown();

}

}

 

public void lock() {

try {

if(this.tryLock()){

System.out.println("Thread " + Thread.currentThread().getId() + " " +myZnode + " get lock true");

return;

}

else{

waitForLock(waitNode, sessionTimeout);//等待锁

}

} catch (KeeperException e) {

throw new LockException(e);

} catch (InterruptedException e) {

throw new LockException(e);

}

}

public boolean tryLock() {

try {

String splitStr = "_lock_";

if(lockName.contains(splitStr))

throw new LockException("lockName can not contains \\u000B");

//创建临时子节点

myZnode = zk.create(root + "/" + lockName + splitStr, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);

System.out.println(myZnode + " is created ");

//取出所有子节点

List<String> subNodes = zk.getChildren(root, false);

//取出所有lockName的锁

List<String> lockObjNodes = new ArrayList<String>();

for (String node : subNodes) {

String _node = node.split(splitStr)[0];

if(_node.equals(lockName)){

lockObjNodes.add(node);

}

}

Collections.sort(lockObjNodes);

 

if(myZnode.equals(root+"/"+lockObjNodes.get(0))){

//如果是最小的节点,则表示取得锁

System.out.println(myZnode + "==" + lockObjNodes.get(0));

return true;

}

//如果不是最小的节点，找到比自己小1的节点

String subMyZnode = myZnode.substring(myZnode.lastIndexOf("/") + 1);

waitNode = lockObjNodes.get(Collections.binarySearch(lockObjNodes, subMyZnode) - 1);//找到前一个子节点

} catch (KeeperException e) {

throw new LockException(e);

} catch (InterruptedException e) {

throw new LockException(e);

}

return false;

}

public boolean tryLock(long time, TimeUnit unit) {

try {

if(this.tryLock()){

return true;

}

return waitForLock(waitNode,time);

} catch (Exception e) {

e.printStackTrace();

}

return false;

}

private boolean waitForLock(String lower, long waitTime) throws InterruptedException, KeeperException {

Stat stat = zk.exists(root + "/" + lower,true);//同时注册监听。

//判断比自己小一个数的节点是否存在,如果不存在则无需等待锁,同时注册监听

if(stat != null){

System.out.println("Thread " + Thread.currentThread().getId() + " waiting for " + root + "/" + lower);

this.latch = new CountDownLatch(1);

this.latch.await(waitTime, TimeUnit.MILLISECONDS);//等待，这里应该一直等待其他线程释放锁

this.latch = null;

}

return true;

}

public void unlock() {

try {

System.out.println("unlock " + myZnode);

zk.delete(myZnode,-1);

myZnode = null;

zk.close();

} catch (InterruptedException e) {

e.printStackTrace();

} catch (KeeperException e) {

e.printStackTrace();

}

}

public void lockInterruptibly() throws InterruptedException {

this.lock();

}

public Condition newCondition() {

return null;

}

 

public class LockException extends RuntimeException {

private static final long serialVersionUID = 1L;

public LockException(String e){

super(e);

}

public LockException(Exception e){

super(e);

}

}

}

ETCD实现分布式锁

ETCD实现分布式锁内容较多，篇幅太长，以后专题介绍

三种分布式锁比较

CAP定律

2000年，Eric Brewer教授提出CAP猜想，2年后，被Seth Gilbert和Nancy Lynch大佬所证明。

CAP是指：Consistency（强一致性），Availability（可用性），Partition tolerance（分区容错性），分布式系统，三者同时满足两者。

由于分布式系统中，一定是存在分区的，所有cap模型中常用的只能是是：AP或CP模型了。

@Component

publicclassDistributedLockHandler {

    privatestatic 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) {

        returngetLock(lock, LOCK_TRY_TIMEOUT,LOCK_TRY_INTERVAL,LOCK_EXPIRE);

/**

     * 尝试获取全局锁

     * @param lock    锁的名称

     * @param timeout 获取超时时间单位ms

     * @return true 获取成功，false获取失败

*/

    public boolean tryLock(Lock lock, long timeout) {

        returngetLock(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) {

        returngetLock(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) {

        returngetLock(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())) {

                returnfalse;

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

                    returntrue;

                } else {//存在锁

                    logger.debug("lock is exist!！！");

                if (System.currentTimeMillis()- startTime > timeout) {//尝试超过了设定值之后直接跳出循环

                    returnfalse;

                Thread.sleep(tryInterval);

            while (template.hasKey(lock.getName())) ;

        } catch (InterruptedException e) {

            logger.error(e.getMessage());

            returnfalse;

        returnfalse;

/**

     * 释放锁

*/

    publicvoidreleaseLock(Lock lock) {

        if (!StringUtils.isEmpty(lock.getName())) {

            template.delete(lock.getName());

这篇关于分布式锁设计与实践的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！

分布式锁设计与实践

分布式锁是什么

分布式锁应用场景

防止用户重复下单

MQ消息去重

分布式锁设计与实现

redis实现分布式锁

利用 SETNX 和 SETEX

使用Redlock

Zookeeper实现分布式锁

节点概念

zookeeper分布式锁的原理

代码实现

ETCD实现分布式锁

三种分布式锁比较

CAP定律

相关编程文章