redis：什么是分布式锁？实现分布式锁的三种方式_什么是分布式锁,redis如何实现?

在很多场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等，那具体什么是分布式锁，分布式锁应用在哪些业务场景，如何来实现

应用场景

• 比如春运时，您需要在 12306 上抢购回家火车票，但 Redis 数据库中只剩一张票了，此时有多个用户来预订购买，那么这张票会被谁抢走呢？redis服务器又是如何处理这种情景呢？在这个过程就需要使用分布式锁
• 双11热卖过程中，怎样避免最后一件商品不被多人同时购买（超卖问题）

为什么要使用分布式锁

问： 多线程并发的时候，如何实现对共享资源安全的访问

答：通过锁来实现

问：单机多进程多线程的时候，如何实现对共享资源安全的访问

答：通过共享内存、管道等实现

问： 多机多进程的时候，如何实现对共享资源安全的访问

答：通过分布式锁

我们在开发的时候，如果需要对某一个共享变量进行多线程同步访问的时候，可以使用我们学到的锁进行处理，并且可以完美的运行，毫无bug

注意这是单机应用，后来业务发展，需要做集群，一个应用需要部署到几台机器上然后做负载均衡，大致如下图：

上图可以看到，

• 变量A存在三个服务器内存中（这个变量A主要体现是在一个类中的一个成员变量，是一个有状态的对象）
• 如果不加任何控制的话，变量A同时都会在分配一块内存，三个请求发过来同时对这个变量操作，显然结果是不对的！
• 即使不是同时发过来，三个请求分别操作三个不同内存区域的数据，变量A之间不存在共享，也不具有可见性，处理的结果也是不对的！

如果我们业务中确实存在这个场景的话，我们就需要一种方法解决这个问题！

• 为了保证一个方法或者属性在高并发的情况下的同一时间只能被同一个线程执行，在传统的单机应用单机部署的情况下，可以使用并发处理相关的功能进行互斥控制。
• 但是，随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统之后，由于分布式系统多线程，多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的应用并不能提供分布式锁的能力。
• 为了解决这个问题就需要一种跨机器的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

分布式锁，顾名思义，就是分布式项目开发中用到的锁，可以用来控制分布式系统之间同步访问共享资源

思路是：在整个系统提供一个全局、唯一的获取锁的“东西”，然后每个系统在需要加锁时，都去问这个“东西”拿到一把锁，这样不同的系统拿到的就可以认为是同一把锁。至于这个“东西”，可以是Redis、Zookeeper，也可以是数据库。

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

在实现分布式锁的三种实现方法之前，先了解下分布式锁应该具备哪些条件。

• 互斥性： 在任何时刻，对于同一条数据，只有一台线程能够持有锁；
• 独占性： 加锁和解锁必须是由同一个线程来设置
• 高可用性：在分布式场景下，一小部分服务器宕机不影响正常使用，这种情况就需要将提供分布式锁的服务以集群的方式部署；
• 具备锁失效机制，防止死锁：如果客户端没有主动释放锁，服务器会在一段时间之后自动释放锁，防止客户端宕机或者网络不可达时产生死锁；
• 具备非阻塞锁特性，即没有获取到锁将直接返回获取锁失败

分布式锁的三种实现方式

目前几乎所有大型网站以及应用都是分布式的，分布式场景中的数据一致性一致是一个比较重要的问题。分布式的CAP理论告诉我们“任何一个分布式系统都无法同时满足一致性（Consistency）、可用性（Availability）和分区容错性（Partition tolerance），最多只能同时满足两项。”所以，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大部分场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证“最终一致性”，只要这个最终时间是在系统可以接受的范围内既可了。

在很多场景中，我们为了保证数据的最终一致性，需要很多技术方案来支持，比如分布式事务、分布式锁等。有时，我们需要保证一个方法在同一时间只能被同一个线程执行

基于数据库实现分布式锁

基于数据库的实现方式的核心思想是：在数据库中创建一个表，表中包含“方法名”等字段，并在方法名字段上创建唯一索引，想要执行某个方法，就使用这个方法向表中插入数据，成功插入则获取锁，执行完成之后删除对应的行数据释放锁

（1）创建一个表：

DROP TABLE IF EXISTS `method_lock`;
CREATE TABLE `method_lock` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
  `method_name` varchar(64) NOT NULL COMMENT '锁定的方法名',
  `desc` varchar(255) NOT NULL COMMENT '备注信息',
  `update_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uidx_method_name` (`method_name`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COMMENT='锁定中的方法';
1
2
3
4
5
6
7
8
9

（2）想要执行某个方法，就使用这个方法名向表中插入数据：

INSERT INTO method_lock (method_name, desc) VALUES ('methodName', '测试的methodName');
1

因为我们对method_name做了唯一性约束，这里如果有多个请求同时提交到数据库的话，数据库会保证只有一个操作可以成功，那么我们就可以认为操作成功的那个线程获得了该方法的锁，可以执行方法体内容。

（3）成功插入则获取锁，执行完成后删除对应的行数据释放锁：

delete from method_lock where method_name ='methodName';
1

注意：这只是使用基于数据库的一种方法，使用数据库实现分布式锁还有很多其他的玩法！

使用基于数据库的这种实现方式很简单，但是对于分布式锁应该具备的条件来说，它有一些问题需要解决及优化：

• 因为是基于数据库实现的，数据库的可用性以及性能将直接影响分布式锁的可用性以及性能，所以，数据库需要双机部署，数据同步，主备切换
• 不具备可重入特性，因为同一个线程在释放锁之前，行数据一直存在，无法再次成功插入数据，所以，需要在表中新增一列，用于记录当前获取到锁的机器以及线程信息，在再次获取锁的时候，先查询表中机器以及线程信息释放和当前机器以及线程相同，如果相同则直接获取锁；
• 没有锁失效机制，因为有可能出现成功插入数据之后，服务器宕机了，对应的数据没有删除，等服务恢复之后一直获取不到锁，所以，需要在表中新增一列，用于记录失效时间，并且需要有定时任务清楚这些失效的数据.
• 不具备阻塞锁特性，获取不到锁直接返回失败，所以需要优化获取逻辑，循环多次去获取。
• 在实施的过程中会遇到各种不同的问题，为了解决这些问题，实现方式将会越来越复杂；依赖数据库需要一定的资源开销，性能问题需要考虑。

基于Zookeeper实现分布式锁

ZooKeeper是一个为分布式应用提供一致性服务的开源组件，它内部是一个分层的文件系统目录树结构，规定同一个目录下只能有一个唯一文件名。基于ZooKeeper实现分布式锁的步骤如下：

（1）创建一个目录mylock；
（2）线程A想获取锁就在mylock目录下创建临时顺序节点；
（3）获取mylock目录下所有的子节点，然后获取比自己小的兄弟节点，如果不存在，则说明当前线程顺序号最小，获得锁；
（4）线程B获取所有节点，判断自己不是最小节点，设置监听比自己次小的节点；
（5）线程A处理完，删除自己的节点，线程B监听到变更事件，判断自己是不是最小的节点，如果是则获得锁。

这里推荐一个Apache的开源库Curator，它是一个ZooKeeper客户端，Curator提供的InterProcessMutex是分布式锁的实现，acquire方法用于获取锁，release方法用于释放锁。

优点：具备高可用、可重入、阻塞锁特性，可解决失效死锁问题。

缺点：因为需要频繁的创建和删除节点，性能上不如Redis方式。

基于缓存（Redis等）实现分布式锁

选用Redis实现分布式锁原因：

• Redis有很高的性能：加锁和释放锁开销较小
• Redis命令对此支持较好，实现起来比较方便

原理

分布式锁的本质其实就是要在 Redis 里面占一个“坑”，当别的进程也要来占时，发现已经有人蹲了，就只好放弃或者稍做等待。

单机版本方案

方案一：setnx/expired/del

占坑一般使用setnx指令，只允许被一个客户端占坑。先来先占，用完了，再调用del指令释放“坑”

（1）SETNX

SETNX key val：

• 当且仅当key不存在时，set一个key为val的字符串，返回1；
• 若key存在，则什么都不做，返回0。

（2）expire

• expire key timeout： 为key设置一个超时时间，单位为second，超过这个时间锁会自动释放，避免死锁。

（3）delete

• delete key：删除key

在使用Redis实现分布式锁的时候，主要就会使用到这三个命令。

127.0.0.1:6379> setnx lock:codehole true
(integer) 1
127.0.0.1:6379> expire lock:codehole 5
(integer) 1
// ---- do sonmething ---
127.0.0.1:6379> del lock:codehole
1
2
3
4
5
6

实现思想：

• 获取锁的时候，使用setnx加锁，并使用expire命令为锁添加一个超时时间，超过该时间则自动释放锁，锁的value值为一个随机生成的UUID，通过此在释放锁的时候进行判断。
• 获取锁的时候还设置一个获取的超时时间，若超过这个时间则放弃获取锁。
• 释放锁的时候，通过UUID判断是不是该锁，若是该锁，则执行delete进行锁释放。

简单实现代码：

#连接redis
redis_client = redis.Redis(host="localhost",
                           port=6379,
                           password=password,
                           db=10)

#获取一个锁
lock_name：锁定名称
acquire_time: 客户端等待获取锁的时间
time_out: 锁的超时时间
def acquire_lock(lock_name, acquire_time=10, time_out=10):
    """获取一个分布式锁"""
    identifier = str(uuid.uuid4())
    end = time.time() + acquire_time
    lock = "string:lock:" + lock_name
    while time.time() < end:
        if redis_client.setnx(lock, identifier):
            # 给锁设置超时时间, 防止进程崩溃导致其他进程无法获取锁
            redis_client.expire(lock, time_out)
            return identifier
        elif not redis_client.ttl(lock):
            redis_client.expire(lock, time_out)
        time.sleep(0.001)
    return False

#释放一个锁
def release_lock(lock_name, identifier):
    """通用的锁释放函数"""
    lock = "string:lock:" + lock_name
    pip = redis_client.pipeline(True)
    while True:
        try:
            pip.watch(lock)
            lock_value = redis_client.get(lock)
            if not lock_value:
                return True

            if lock_value.decode() == identifier:
                pip.multi()
                pip.delete(lock)
                pip.execute()
                return True
            pip.unwatch()
            break
        except redis.excetions.WacthcError:
            pass
    return False

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

测试刚才实现的分布式锁

• 例子中使用50个线程模拟秒杀一个商品，使用–运算符来实现商品减少，从结果有序性就可以看出是否为加锁状态。

def seckill():
    identifier=acquire_lock('resource')
    print(Thread.getName(),"获得了锁")
    release_lock('resource',identifier)


for i in range(50):
    t = Thread(target=seckill)
    t.start()
1
2
3
4
5
6
7
8
9

方案二：set/del

但是上面逻辑还是会有问题，如下图，如果在setnx和expire之间服务器进程突然死掉了，就会导致expire得不到执行，导致死锁

这个问题的根源在于setnx和expire的两条指令不是原子指令。如果这两条指令可以一起执行就不会出现问题。为此在redis2.8版本中引入了set 指令的参数，使得setnx和expire指令可以一起执行

127.0.0.1:6379> set lock:codehole true ex 5 nx

(integer) 1

// ---- do sonmething ---
127.0.0.1:6379> del lock:codehole
1
2
3
4
5
6

遇到的问题

某个客户端宕机了导致锁一直被占用，进而死锁（设置超时时间）

场景分析

依赖分布式锁的机制，某个用户操作redis时对应的客户端宕机了，而且此时已经获取到了锁，导致锁一直被持有，其他客户端拿不到锁，这就是死锁，如何解决呢？

• 由于锁操作由用户控制加锁解锁，必定会存在加锁未解锁的风险
• 需要解锁操作不能仅依赖用户控制，系统级别要给出对应的保底处理方案
  
  

解锁方案

1. 使用expire为锁key添加时间限制，到时不释放锁，则放弃锁

setnx lock-key 001 # 设置分布式锁
expire lock-key second # 设置单位为秒
pexpire lock-key milliseconds # 设置单位为毫秒

1
2
3
4

2. 或者直接设置的时候添加时间限制

set lock-key value NX PX 毫秒数 
# 比如为key为name设置分布式锁
set lock-name 001 NX PX 5000

1
2
3
4

• value可以是任意值
• NX代码只有lock-key不存在时才设置值

实际开发中如何知道设置多少时间才合适

由于操作通过都是微秒或者毫秒级，因此锁设定时间不宜过大。具体时间需要业务测试后确定

比如：持有锁的操作最长执行时间127ms，最短执行时间7ms。

• 测试百万次最长执行时间对应命令的最大耗时，测试百万次网络延迟平均耗时
• 时间设定推荐：最大耗时120%+平均网络延迟110%
• 如果二者相差2个数量级，取其中单个耗时较长即可.

流程图

锁误解除

问题

• 假如某线程成功得到了锁，并且设置的超时时间是30秒。
• 如果某些原因导致线程 A 执行的很慢很慢，过了 30 秒都没执行完，这时候锁过期自动释放，线程 B 得到了锁。
• 随后，线程 A 执行完了任务，线程 A 接着执行 del 指令来释放锁。但这时候线程 B 还没执行完，线程A实际上 删除的是线程 B 加的锁。

解决

• 我们应该在删除锁之前, 判断这个锁是否是自己设置的锁, 如果不是（例如自己的锁已经超时释放）, 那么就不要删除了.

• 怎么知道当前加锁的是不是自己呢？

  • 我们可以在set 锁时，存入自己的信息！删除锁前, 判断下里面的值是不是与自己相等. 如果不等，就不要删除了.
  • 具体在加锁的时候把当前线程的id当做value，可生成一个 UUID 标识当前线程，在删除之前验证key对应的value是不是自己线程的id。即：随机值+当前线程的id, 通过UUID.randomUUID().toString()+Thread.currentThread().getId()获取到.

• 但是，这样做又隐含了一个新的问题，判断和释放锁是两个独立操作，不是原子性。解决：用 lua 脚本做验证标识和解锁操作。

tag = random.nextint()   $ 随机数
if redis.set(key, tag, nx = true, ex = 5):
	do_something();
	redis.delifequals(key, tag);
1
2
3
4

超时解锁导致并发

问题

• redis的分布式锁不能解决超时问题，如果在加锁和释放锁之间的逻辑执行得太长，以至于超出了锁的超时限制，就会出现问题。
• 因为这个时候第一个线程持有的锁过期了，临界区的逻辑还没有执行完，而同时第二个线程就提前持有了这把锁，导致临界区的代码不能得到严格的串行执行
• 还有一种情况是同一时间有 A，B 两个线程在访问代码块

解决

1. 将过期时间设置得足够长，确保代码逻辑能够在锁释放之前能够执行完成
2. 为获取锁的线程增加守护进程，为将要过期但没有是否的锁增加有效时间
   • 让获得锁的线程开启一个守护线程，用来给快要过期的锁“续航”。
   • 当过去了 29 秒，线程 A 还没执行完，这时候守护线程会执行 expire 指令，为这把锁“续命”20 秒。守护线程从第 29 秒开始执行，每 20 秒执行一次。
   • 当线程 A 执行完任务，会显式关掉守护线程。
   • 另一种情况，如果节点 1 忽然断电，由于线程 A 和守护线程在同一个进程，守护线程也会停下。这把锁到了超时的时候，没人给它续命，也就自动释放了。
3. redis分布式不要用于较长时间的任务。如果真的偶尔出现了问题，造成的数据小错乱可能需要人工介入解决

方法嵌套导致的死锁问题：可重入锁

场景分析

就是实际开发过程中，一端代码内部会有嵌套方法，外层方法获取到锁之后，内层再去获取锁时由于锁已经存在了就无法获取了，但内层代码不执行完外层也释放不了锁啊，这就是方法嵌套导致的死锁问题，怎么解决呢？

解决方案

1. 让锁成为可重入锁，也就是外层代码获取到这把锁，内层代码可以获取到该锁
   • 可重入性是指线程在持有锁的情况下再次请求加锁，如果一个锁支持同一个线程的多次加锁，那么这个锁就是可重入的。
   • 不推荐使用可重入锁，它加重了客户端的复杂性，在编写业务方法的时候注意逻辑结构进行调整完全可以不使用可重入锁
2. 获取时判断是不是自己的锁，如果是则继续使用，而且要记录重入的次数
   • redis分布式锁如果要支持可重入，需要对客户端的set方法进行包装，使用线程的ThreadLocal变量存储当前持有锁的计数
   • Redis 可通过对锁进行重入计数，加锁时加 1，解锁时减 1，当计数归 0 时释放锁。
3. 这里的锁不能使用之前的string类型作为lock-key的值了，所的value要使用hash结构

hset lock-key 线程信息, 重入次数(默认1) NX PX 毫秒数
1

key: lock-key
value-key:线程信息
value-value:重入次数

流程图

下面我们假设锁的key为“lock”，hashKey是当前线程的id：“threadId”，锁自动释放时间假设为20

获取锁的步骤：

1、判断lock是否存在 EXISTS lock

• 返回1则存在，说明有人获取锁了，下面判断是不是自己的锁

  • 判断当前线程id作为hashKey是否存在：HEXISTS lock threadId
    • 存在，说明是自己获取的锁，重入次数+1：HINCRBY lock threadId 1，去到步骤2
    • 不存在，说明锁已经有了，且不是自己获取的，锁获取失败，end

• 返回0则不存在，说明可以获取锁，并存入value值HSET key threadId 1

2、设置锁自动释放时间，EXPIRE lock 20、

释放锁的步骤：

1、判断当前线程id作为hashKey是否存在：HEXISTS lock threadId

• 存在，说明锁还在，重入次数减1：HINCRBY lock threadId -1，获取新的重入次数.
• 不存在，说明锁已经失效，不用管了

2、判断重入次数是否为0：

• 为0，说明锁全部释放，删除key：DEL lock
• 大于0，说明锁还在使用，重置有效时间：EXPIRE lock 20

无法等待锁释放

问题

上述命令执行都是立即返回的，如果客户端可以等待锁释放就无法使用。

1. 可以通过客户端轮询的方式解决该问题，当未获取到锁时，等待一段时间重新获取锁，直到成功获取锁或等待超时。这种方式比较消耗服务器资源，当并发量比较大时，会影响服务器的效率。
2. 另一种方式是使用 Redis 的发布订阅功能，当获取锁失败时，订阅锁释放消息，获取锁成功后释放时，发送锁释放消息。

客户端加锁失败怎么办

如果客户端在处理请求时加锁没加成功怎么办？一般有三种策略

• 直接抛出异常，通知用户稍后重试
• sleep一会儿，然后再重试
• 将请求转移到延迟队列，过一会儿再试

（1）直接抛出特定类型的异常

• 这种方式比较适合由用户直接发起的请求。

• 用户看到错误对话框之后，会先阅读对话框的内容，在点击重试，这样就可以起到人工延迟的效果

• 如果考虑到用户体验，可以由前端的代码替代用户来进行延迟控制控制

• 它本质是对当前请求的放弃，有用户决定是否重新发起新的请求

（2）sleep

• sleep会阻塞当前的消息处理线程，会导致队列的后继消息处理出现延迟
• 如果碰撞得比较频繁或者队列里消息比较大，sleep可能并不何时
• 如果因为个别死锁的key导致加锁不成功，线程会彻底堵死，导致后继消息永远得不到及时处理

（3） 延迟队列

• 这种方法比较适合异步消息处理，将当前冲突的请求扔到另一个队列延后处理以避开冲突

集群模式的Redis分布式锁

为什么引入

上面版本的锁有一个很大的缺点，就是它加锁时只能作用在一个redis节点上，即使redis通过sentinel保证高可用，如果这个master由于某些原因发送了主从切换，那么就会出现锁丢失的情况：

• 在redis的master节点上拿到了锁
• 但是这个加锁的key还没有同步到slave节点
• master故障，发生故障转移，slave节点升级为master几点
• 导致锁丢失

正因为如此，Redis作者antirez基于分布式环境下提出了一种更高级的分布式锁的实现方式：Redlock。这种方法比原先单节点的方法更加安全，它可以保证一下特性：

• 安全特性：互斥访问，即永远只能由一个客户端能够拿到锁
• 避免死锁：最终客户端都可以拿到锁，不会出现死锁的情况技术，即使原先锁住某资源的客户端宕机了会在出现了网络分区
• 容错性：只要大部分redis节点存活就可以正常提供服务

Redlock也是Redis所有分布式锁实现方式中唯一能让面试官高潮的方式。

使用场景

多个服务间保证同一时刻同一时间段内同一用户只能有一个请求（防止关键业务出现并发攻击）；

什么是红锁（Redlock）

这里的场景是假设有一个redis cluster，有5个redis master实例。然后执行如下步骤获取一把锁

• 客户端获取服务器当前的时间t0，毫秒数
• 使用相同的key和具有唯一性的value（例如UUID）依次向5个实例获取锁。客户端在获取锁的时候自身设置一个远小于业务锁需要的持续时间的超时时间。举个例子，假设锁需要10s，超时时间可以设置成5-50毫秒。这是为了不要过长时间等待已经关闭的Redis服务。并且试着获取下一个Redis实例。超时之后就直接跳到下一个结点。
• 客户端通过获取所有能获取的锁后的时间减去第一步的时间t2(=t1-t0)。只有t2小于锁的业务有效时间(也就是第二步的10秒)，并且，客户端在至少3(5/2+1)台上获取到锁我们才认为锁获取成功。
• 如果成功获取锁，则锁的真正有效时间是 TTL减去第三步的时间差 的时间；比如：TTL 是5s,获取所有锁用了2s,则真正锁有效时间为3s(其实应该再减去时钟漂移);
• 如果客户端没有获取到锁，可能是没有在大于等于N/2+1个实例上获取锁，也可能是有效时间(10s-t2)为负数，我们就必须去释放所有redis上的锁，否则影响其他client获取锁

假设有5个redis节点，这些节点之间既没有主从，也没有集群关系。客户端用相同的key和随机值在5个节点上请求锁，请求锁的超时时间应小于锁自动释放时间。当在3个（超过半数）redis上请求到锁的时候，才算是真正获取到了锁。如果没有获取到锁，则把部分已锁的redis释放掉。

如果 redis 作为分布式锁的时候，主节点挂掉了，但是数据还没有同步到从节点，这种情况怎么办？

场景

具体发生的过程介绍如下：

1）redis的部署方式为一主多从

2）如果客户端1给redis主节点加了分布式锁，但是redis主节点在异步复制给从节点之前就挂了，导致分布式锁丢失

3）某个从节点自动成为新的主节点后，因为同步锁丢失的原因，并不知道客户端1已经加锁，这时客户端2也来了加锁行为，这时客户端2自然加锁成功！！！

问题的关键是，客户端1没有主动释放锁，也没有因为过期而释放锁，这时客户端2就能得到锁，这不是很大的问题吗？我们的分布式锁，在这种情况下，不是失去意义了吗？？？

解决方法：红锁

关键思想：

• 加锁时，尝试给每个redis节点都加锁
• 必须有超过1半的redis加锁成功，才能算本次加锁成功（加锁成功含义：成功获取到锁，且耗时时间小于锁失效时间）

比如说有1个主节点、4个从节点，那么一共有5个节点，红锁加锁时，会同时往5个节点加锁，至少有3个节点加锁成功，才算最终加锁成功

如果加了红锁机制后，主从情况下，分布式锁刚加完，主节点挂掉，那么新的从节点大概率会持有分布式锁。

为什么会是大概率？因为可能当时红锁加锁也成功了，但是其实5节点中，有一个从节点挂掉了，所以改从节点没有分布式锁，但是之后又被实施人员启动了，这种时候，如果主节点挂掉，而刚启动起来的从节点又刚好被作为新的主节点，那么还有会有问题。

面试题

加锁的时候什么时候选择本地锁，什么时候选择分布式锁？

单机模式下用本地锁

分布式模式下用分布式锁