企业官网建设流程全解析

分布式锁 — 概念、原理与实践

一、为什么需要分布式锁

1.1 单机锁的局限

单机应用（1个JVM进程）： 线程A ─┐ 线程B ─┼→ synchronized / ReentrantLock → 资源 线程C ─┘ ✅ 有效：所有线程在同一个JVM内，共享同一把锁 分布式应用（多个JVM进程/多台服务器）： 服务器1（JVM-1）：线程A → synchronized → 本地锁1 服务器2（JVM-2）：线程B → synchronized → 本地锁2 服务器3（JVM-3）：线程C → synchronized → 本地锁3 ❌ 无效：三台服务器各自加的是各自的本地锁，互相不可见 → 三个线程可以同时访问共享资源（数据库同一行）

1.2 类比理解

本地锁 = 家门锁（只能锁住自己家门，邻居有自己的锁） 分布式锁 = 小区门禁（所有住户共享同一个门禁系统，一次只能一个人通过）

1.3 核心问题

多个服务实例同时操作同一条数据时，如何保证同一时刻只有一个实例能执行关键操作？

注：

博客：

https://blog.csdn.net/badao_liumang_qizhi

二、核心概念

2.1 什么是分布式锁

分布式锁是一种跨进程/跨服务器的互斥机制，通过一个所有实例都能访问的共享存储（Redis/ZooKeeper/数据库）来协调并发访问。

2.2 关键术语

2.3 分布式锁必须满足的条件

三、与本地锁的对比

3.1 Java 本地锁（JVM 内有效）

3.2 分布式锁（跨 JVM 有效）

四、底层原理

4.1 基于 Redis 的分布式锁

加锁原理

客户端A → Redis：SET lock_key requestId_A NX PX 30000 NX = Only set if Not eXists（key不存在时才设置成功） PX 30000 = 30秒后自动过期 如果返回 OK → 加锁成功 如果返回 nil → 加锁失败（锁被别人持有）

为什么用 NX？

时刻T1：客户端A执行 SET lock NX → 成功（key不存在） 时刻T2：客户端B执行 SET lock NX → 失败（key已存在） → 只有A获得锁，B被拒绝

为什么需要 PX（过期时间）？

客户端A加锁成功后崩溃了（没来得及释放锁）： 没有过期时间 → 锁永远不释放 → 死锁（其他人永远无法获取） 有过期时间 → 30秒后Redis自动删除key → 其他人可以获取锁

为什么需要 requestId？

场景：A加的锁，B来释放 时刻T1：A 加锁成功（lock = requestId_A） 时刻T2：A 处理业务超时，锁过期自动释放 时刻T3：B 加锁成功（lock = requestId_B） 时刻T4：A 处理完毕，执行释放锁操作 如果不校验 requestId： A 直接 DEL lock → 把B的锁删了！B以为自己还持有锁，实际已经没了 如果校验 requestId： A 释放时检查 lock 的值是否是 requestId_A → 不是 → 不释放 → B 的锁不受影响

加锁 Lua 脚本（原子操作）

-- KEYS[1] = 锁的key-- KEYS[2] = requestId的key-- ARGV[1] = requestId值-- ARGV[2] = 过期时间（毫秒）if(redis.call('exists',KEYS[1])==0)thenredis.call('hset',KEYS[1],KEYS[2],ARGV[1])redis.call('pexpire',KEYS[1],ARGV[2])return1-- 加锁成功elsereturn0-- 锁已存在，加锁失败end

为什么用 Lua 脚本？

• exists+hset+pexpire三个命令需要原子执行
• 如果分开执行，两条命令之间可能被其他客户端插入操作
• Redis 保证 Lua 脚本原子性执行

解锁 Lua 脚本（原子操作）

-- KEYS[1] = 锁的key-- KEYS[2] = requestId的key-- ARGV[1] = requestId值ifredis.call('hget',KEYS[1],KEYS[2])==ARGV[1]thenredis.call('del',KEYS[1])return1-- 解锁成功elsereturn0-- 不是自己的锁，拒绝释放end

4.2 基于数据库的分布式锁

原理

利用数据库的唯一约束或排他锁实现互斥。

方式1：唯一索引（INSERT 方式）

-- 建表CREATETABLEdistributed_lock(idBIGINTAUTO_INCREMENTPRIMARYKEY,lock_keyVARCHAR(100)NOTNULLUNIQUE,-- 唯一约束request_idVARCHAR(64)NOTNULL,expire_timeDATETIMENOTNULL,create_timeDATETIMENOTNULL);-- 加锁：INSERT（唯一约束保证互斥）INSERTINTOdistributed_lock(lock_key,request_id,expire_time,create_time)VALUES('order_process_123','uuid-xxx',NOW()+INTERVAL30SECOND,NOW());-- 成功 → 获得锁-- 失败（Duplicate entry） → 锁被占用-- 解锁：DELETE（校验 request_id）DELETEFROMdistributed_lockWHERElock_key='order_process_123'ANDrequest_id='uuid-xxx';

方式2：SELECT FOR UPDATE（悲观锁）

-- 加锁BEGIN;SELECT*FROMdistributed_lockWHERElock_key='order_process_123'FORUPDATE;-- 其他事务对同一行的 SELECT FOR UPDATE 会阻塞等待-- 执行业务逻辑...-- 解锁COMMIT;-- 事务提交后锁自动释放

4.3 基于 ZooKeeper 的分布式锁

原理

利用 ZooKeeper 的临时有序节点实现。

/locks/order_process_123/ ├── node_0000000001 (客户端A创建) ← 序号最小，获得锁 ├── node_0000000002 (客户端B创建) ← 监听前一个节点 └── node_0000000003 (客户端C创建) ← 监听前一个节点 客户端A处理完毕 → 删除 node_0000000001 → 触发 node_0000000002 的 Watch 通知 → 客户端B检查自己是否最小 → 是 → 获得锁

优点：临时节点在客户端断连时自动删除（防死锁），Watch 机制避免轮询。

五、JPA/数据库中的锁实现

5.1 乐观锁（Optimistic Locking）

思想：假设冲突很少发生，不加锁，提交时检查是否被修改过。

@Entity@Table(name="stock")publicclassStock{@IdprivateIntegerid;privateIntegeritemSkuId;privateIntegerqty;@Version// JPA 乐观锁注解privateIntegerversion;}

执行流程：

线程A：SELECT * FROM stock WHERE id=1 → version=5, qty=100 线程B：SELECT * FROM stock WHERE id=1 → version=5, qty=100 线程A：UPDATE stock SET qty=90, version=6 WHERE id=1 AND version=5 → 成功（影响1行） 线程B：UPDATE stock SET qty=80, version=6 WHERE id=1 AND version=5 → 失败（影响0行，version已变为6） → JPA 抛出 OptimisticLockException → 线程B可以重试

适用场景：并发冲突概率低，读多写少。

5.2 悲观锁（Pessimistic Locking）

思想：假设冲突频繁，操作前先加锁。

// JPA 悲观锁查询@RepositorypublicinterfaceStockRepositoryextendsJpaRepository<Stock,Integer>{@Lock(LockModeType.PESSIMISTIC_WRITE)@Query("SELECT s FROM Stock s WHERE s.itemSkuId = :itemSkuId")StockfindByItemSkuIdForUpdate(@Param("itemSkuId")IntegeritemSkuId);}

执行的 SQL：

SELECT*FROMstockWHEREitem_sku_id=?FORUPDATE;-- 其他事务对同一行的写操作会阻塞，直到当前事务提交

适用场景：并发冲突概率高，写多读少。

5.3 乐观锁 vs 悲观锁 vs 分布式锁

六、各技术框架中的分布式锁实现

6.1 Redisson（最流行的 Redis 分布式锁框架）

// 依赖// <dependency>// <groupId>org.redisson</groupId>// <artifactId>redisson-spring-boot-starter</artifactId>// <version>3.27.0</version>// </dependency>@ServicepublicclassOrderService{@ResourceprivateRedissonClientredissonClient;publicvoidprocessOrder(StringorderId){RLocklock=redissonClient.getLock("order_lock_"+orderId);try{// 等待10秒，锁自动释放时间30秒booleanacquired=lock.tryLock(10,30,TimeUnit.SECONDS);if(acquired){doProcess(orderId);}else{thrownewRuntimeException("获取锁超时");}}catch(InterruptedExceptione){Thread.currentThread().interrupt();}finally{if(lock.isHeldByCurrentThread()){lock.unlock();}}}}

Redisson 的高级特性：

• 看门狗机制（Watchdog）：自动续期，防止业务未完成锁就过期
• 可重入锁：同一线程可以重复获取同一把锁
• 红锁（RedLock）：多 Redis 节点加锁，防单点故障

6.2 Spring Integration（数据库锁）

// 依赖// <dependency>// <groupId>org.springframework.integration</groupId>// <artifactId>spring-integration-jdbc</artifactId>// </dependency>@ConfigurationpublicclassLockConfig{@BeanpublicDefaultLockRepositorylockRepository(DataSourcedataSource){DefaultLockRepositoryrepository=newDefaultLockRepository(dataSource);repository.setPrefix("APP_LOCK_");repository.setTimeToLive(30000);// 30秒过期returnrepository;}@BeanpublicJdbcLockRegistrylockRegistry(LockRepositorylockRepository){returnnewJdbcLockRegistry(lockRepository);}}@ServicepublicclassOrderService{@ResourceprivateLockRegistrylockRegistry;publicvoidprocessOrder(StringorderId){Locklock=lockRegistry.obtain("order_"+orderId);try{if(lock.tryLock(10,TimeUnit.SECONDS)){try{doProcess(orderId);}finally{lock.unlock();}}}catch(InterruptedExceptione){Thread.currentThread().interrupt();}}}

6.3 Curator（ZooKeeper 锁）

// 依赖// <dependency>// <groupId>org.apache.curator</groupId>// <artifactId>curator-recipes</artifactId>// <version>5.5.0</version>// </dependency>@ServicepublicclassOrderService{@ResourceprivateCuratorFrameworkcuratorClient;publicvoidprocessOrder(StringorderId){InterProcessMutexlock=newInterProcessMutex(curatorClient,"/locks/order_"+orderId);try{if(lock.acquire(10,TimeUnit.SECONDS)){try{doProcess(orderId);}finally{lock.release();}}}catch(Exceptione){thrownewRuntimeException("获取锁失败",e);}}}

七、通用示例代码

7.1 基于 Redis 的分布式锁

/** * 分布式锁实现（基于Redis + Lua脚本）. * * 设计要点： * 1. NX 保证互斥 * 2. PX 防死锁（自动过期） * 3. requestId 防误释放 * 4. Lua 脚本保证原子性 * 5. 自旋等待 + 超时退出 */publicclassRedisDistributedLockimplementsAutoCloseable{privatestaticfinalLoggerlog=LoggerFactory.getLogger(RedisDistributedLock.class);privatefinalStringRedisTemplateredisTemplate;privatefinalStringlockKey;privatefinalStringrequestId;privatefinallongleaseTimeMillis;privatevolatilebooleanlocked=false;privatestaticfinalStringLOCK_PREFIX="distributed_lock:";// 加锁Lua脚本privatestaticfinalStringLOCK_SCRIPT="if redis.call('setnx', KEYS[1], ARGV[1]) == 1 then "+" redis.call('pexpire', KEYS[1], ARGV[2]); "+" return 1; "+"else "+" return 0; "+"end";// 解锁Lua脚本privatestaticfinalStringUNLOCK_SCRIPT="if redis.call('get', KEYS[1]) == ARGV[1] then "+" redis.call('del', KEYS[1]); "+" return 1; "+"else "+" return 0; "+"end";publicRedisDistributedLock(StringRedisTemplateredisTemplate,StringbusinessKey,longleaseTimeMillis){this.redisTemplate=redisTemplate;this.lockKey=LOCK_PREFIX+businessKey;this.requestId=UUID.randomUUID().toString().replace("-","");this.leaseTimeMillis=leaseTimeMillis;}/** * 尝试获取锁（支持超时等待）. * * @param waitTimeMillis 最大等待时间（毫秒），0表示不等待立即返回 * @return true-获取成功，false-超时失败 */publicbooleantryLock(longwaitTimeMillis){longdeadline=System.currentTimeMillis()+waitTimeMillis;// 第一次尝试if(doLock()){this.locked=true;log.debug("获取锁成功: key={}, requestId={}",lockKey,requestId);returntrue;}// 自旋等待while(System.currentTimeMillis()<deadline){try{Thread.sleep(100);// 每100ms重试一次}catch(InterruptedExceptione){Thread.currentThread().interrupt();returnfalse;}if(doLock()){this.locked=true;log.debug("获取锁成功(重试): key={}, requestId={}",lockKey,requestId);returntrue;}}log.warn("获取锁超时: key={}",lockKey);returnfalse;}/** * 释放锁. */publicbooleanunlock(){if(!locked){returntrue;}DefaultRedisScript<Long>script=newDefaultRedisScript<>(UNLOCK_SCRIPT,Long.class);Longresult=redisTemplate.execute(script,Collections.singletonList(lockKey),requestId);booleansuccess=result!=null&&result==1;if(success){this.locked=false;log.debug("释放锁成功: key={}, requestId={}",lockKey,requestId);}else{log.warn("释放锁失败(非持有者): key={}, requestId={}",lockKey,requestId);}returnsuccess;}@Overridepublicvoidclose(){unlock();}privatebooleandoLock(){DefaultRedisScript<Long>script=newDefaultRedisScript<>(LOCK_SCRIPT,Long.class);Longresult=redisTemplate.execute(script,Collections.singletonList(lockKey),requestId,String.valueOf(leaseTimeMillis));returnresult!=null&&result==1;}}

7.2 锁工厂

/** * 分布式锁工厂. * 注入后直接使用，无需关心底层 Redis 操作. */@ComponentpublicclassDistributedLockFactory{@ResourceprivateStringRedisTemplatestringRedisTemplate;/** * 获取分布式锁实例. * * @param businessKey 业务锁标识（如 "order_123"） * @param leaseTime 锁过期时间 * @param unit 时间单位 */publicRedisDistributedLockgetLock(StringbusinessKey,longleaseTime,TimeUnitunit){returnnewRedisDistributedLock(stringRedisTemplate,businessKey,unit.toMillis(leaseTime));}/** * 获取分布式锁（默认过期10分钟）. */publicRedisDistributedLockgetLock(StringbusinessKey){returngetLock(businessKey,10,TimeUnit.MINUTES);}}

7.3 使用示例

@ServicepublicclassStockDeductService{@ResourceprivateDistributedLockFactorylockFactory;@ResourceprivateStockRepositorystockRepository;/** * 扣减库存（分布式锁保证并发安全）. */publicvoiddeductStock(IntegeritemId,Integerqty){StringlockKey="stock_deduct_"+itemId;// try-with-resources 自动释放锁try(RedisDistributedLocklock=lockFactory.getLock(lockKey,30,TimeUnit.SECONDS)){// 尝试获取锁，最多等待5秒if(!lock.tryLock(5000)){thrownewRuntimeException("系统繁忙，请稍后重试");}// 获取锁成功，安全执行业务Stockstock=stockRepository.findByItemId(itemId);if(stock.getQty()<qty){thrownewRuntimeException("库存不足");}stock.setQty(stock.getQty()-qty);stockRepository.save(stock);}// 离开 try 块自动调用 close() → unlock()}}

八、常见问题与解决方案

九、关键设计总结