第36章 分布式锁
36.1 分布式锁（Redis锁、ZooKeeper锁）
一、我踩过的分布式锁坑：从“死锁导致系统瘫痪”到“超卖50单”
做电商库存扣减时，我用Redis锁实现并发控制，结果忘记设置过期时间，服务崩溃后锁一直存在，其他服务无法获取锁，库存2小时无法扣减，损失10万订单；后来改成ZooKeeper锁，结果节点创建失败没处理，导致多个线程同时获取锁，超卖了50单，用户收到两个包裹；再vb.net教程C#教程python教程SQL教程access 2010教程后来用Redis锁时，用普通的SET+EX命令，结果高并发下多个线程同时获取锁，库存扣减了多次。这节我把这些踩坑经验揉进去，用大白话讲透分布式锁的核心原理，结合C#实战代码逐行讲解Redis锁和ZooKeeper锁的实现，对比它们的优缺点和适用场景，以及分布式锁的进阶问题（比如锁续约、红锁、公平锁），让你的分布式并发控制既可靠又高效。
二、分布式锁核心需求：解决“多节点抢资源”的问题
分布式锁是一种在分布式系统中控制多个节点并发访问共享资源的机制，比如库存扣减、分布式任务调度、分布式事务。核心需求有四个：
1.互斥性：同一时间只有一个节点能持有锁；
2.避免死锁：锁必须能自动释放，即使持有锁的节点崩溃；
3.容错性：锁服务（比如Redis、ZooKeeper）挂了，锁要能正常工作或自动释放；
4.性能：获取和释放锁的速度要快，不能成为系统瓶颈。
类比：分布式锁就像公共厕所的门锁，同一时间只有一个人能进去；门锁必须有超时自动打开的功能，避免里面的人晕倒了外面的人进不去；厕所管理员（锁服务）不在时，门锁还要能正常工作。
三、Redis锁：高性能的分布式锁，适合高并发场景
Redis锁是基于Redis的原子操作实现的分布式锁，核心优势是高性能、易实现，适合高并发场景（比如秒杀、库存扣减）。
核心原理（大白话）
用Redis的SET命令的原子操作实现：
获取锁：SET lock_key unique_value NX PX 30000，其中：
NX：只有当lock_key不存在时才设置成功（互斥性）；
PX 30000：设置锁的过期时间为30秒（避免死锁）；
unique_value：唯一值（比如UUID），用来判断锁是否是自己的，避免误删别人的锁；
释放锁：用Lua脚本原子执行“判断unique_value是否是自己的，如果是则删除锁”，避免并发释放锁时误删别人的锁。
我踩过的坑：一开始用SET lock_key 1 EX 30 NX，结果多个节点的unique_value都是1，释放锁时会误删别人的锁——必须用唯一的unique_value！
实战1：Redis锁的C#实现（StackExchange.Redis）
用StackExchange.Redis库实现Redis锁，包含获取锁、释放锁、锁续约功能。
步骤1：安装NuGet包
bash
Install-Package StackExchange.Redis
步骤2：Redis锁实现代码
csharp

	using StackExchange.Redis;
	using System;
	using System.Threading;
	using System.Threading.Tasks;
	
	namespace DistributedLock.Redis;
	
	public class RedisDistributedLock : IDisposable
	{
	private readonly IDatabase _redisDb;
	private readonly string _lockKey;
	private readonly string _uniqueValue;
	private readonly TimeSpan _lockTimeout;
	private Timer _renewTimer; // 锁续约定时器
	private bool _isLockHeld;
	
	// 释放锁的Lua脚本：原子判断并删除锁
	private const string ReleaseLockScript = @"
	if redis.call('GET', KEYS[1]) == ARGV[1] then
	return redis.call('DEL', KEYS[1])
	else
	return 0
	end";
	
	public RedisDistributedLock(IConnectionMultiplexer redis, string lockKey, TimeSpan lockTimeout)
	{
	_redisDb = redis.GetDatabase();
	_lockKey = lockKey;
	_lockTimeout = lockTimeout;
	_uniqueValue = Guid.NewGuid().ToString("N"); // 生成唯一值，避免误删别人的锁
	_isLockHeld = false;
	}
	
	/// <summary>
	/// 获取锁（异步）
	/// </summary>
	/// <param name="cancellationToken">取消令牌</param>
	/// <returns>是否获取成功</returns>
	public async Task<bool> AcquireAsync(CancellationToken cancellationToken = default)
	{
	// 1. 用SET NX PX原子操作获取锁
	bool acquired = await _redisDb.StringSetAsync(
	_lockKey,
	_uniqueValue,
	_lockTimeout,
	When.NotExists, // NX：只有当key不存在时才设置
	CommandFlags.None);
	
	if (acquired)
	{
	_isLockHeld = true;
	// 2. 启动锁续约定时器，每10秒续约一次（续约时间为锁超时时间的1/3）
	_renewTimer = new Timer(async _ =>
	{
	if (_isLockHeld)
	{
	await RenewLockAsync();
	}
	}, null, TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(10));
	Console.WriteLine($"[RedisLock] 成功获取锁：{_lockKey}，唯一值：{_uniqueValue}");
	}
	else
	{
	Console.WriteLine($"[RedisLock] 获取锁失败：{_lockKey}");
	}
	
	return acquired;
	}
	
	/// <summary>
	/// 续约锁（异步）
	/// </summary>
	private async Task RenewLockAsync()
	{
	// 用Lua脚本原子判断并续约锁，只有锁是自己的才续约
	string renewScript = @"
	if redis.call('GET', KEYS[1]) == ARGV[1] then
	return redis.call('PEXPIRE', KEYS[1], ARGV[2])
	else
	return 0
	end";
	
	var result = await _redisDb.ScriptEvaluateAsync(
	renewScript,
	new RedisKey[] { _lockKey },
	new RedisValue[] { _uniqueValue, (int)_lockTimeout.TotalMilliseconds });
	
	if ((long)result == 0)
	{
	Console.WriteLine($"[RedisLock] 续约锁失败，锁可能已被释放：{_lockKey}");
	_isLockHeld = false;
	}
	else
	{
	Console.WriteLine($"[RedisLock] 成功续约锁：{_lockKey}");
	}
	}
	
	/// <summary>
	/// 释放锁（异步）
	/// </summary>
	public async Task ReleaseAsync()
	{
	if (!_isLockHeld)
	return;
	
	try
	{
	// 1. 用Lua脚本原子释放锁，避免误删别人的锁
	var result = await _redisDb.ScriptEvaluateAsync(
	ReleaseLockScript,
	new RedisKey[] { _lockKey },
	new RedisValue[] { _uniqueValue });
	
	if ((long)result == 1)
	{
	Console.WriteLine($"[RedisLock] 成功释放锁：{_lockKey}");
	}
	else
	{
	Console.WriteLine($"[RedisLock] 释放锁失败，锁可能已过期或被别人持有：{_lockKey}");
	}
	}
	catch (Exception ex)
	{
	Console.WriteLine($"[RedisLock] 释放锁时出错：{ex.Message}");
	}
	finally
	{
	_isLockHeld = false;
	_renewTimer?.Dispose();
	}
	}
	
	public void Dispose()
	{
	ReleaseAsync().Wait();
	}
	}
	
	// 测试代码
	class Program
	{
	static async Task Main(string[] args)
	{
	// 连接Redis
	var redis = ConnectionMultiplexer.Connect("localhost:6379");
	string lockKey = "stock:lock:product_123";
	
	// 模拟10个并发线程获取锁
	var tasks = new Task[10];
	for (int i = 0; i < 10; i++)
	{
	int threadId = i;
	tasks[i] = Task.Run(async () =>
	{
	using var redisLock = new RedisDistributedLock(redis, lockKey, TimeSpan.FromSeconds(30));
	bool acquired = await redisLock.AcquireAsync();
	if (acquired)
	{
	try
	{
	// 模拟扣减库存的耗时操作
	Console.WriteLine($"[线程{threadId}] 开始扣减库存...");
	await Task.Delay(5000); // 模拟5秒的业务操作
	Console.WriteLine($"[线程{threadId}] 库存扣减完成");
	}
	finally
	{
	await redisLock.ReleaseAsync();
	}
	}
	else
	{
	Console.WriteLine($"[线程{threadId}] 获取锁失败，跳过库存扣减");
	}
	});
	}
	
	await Task.WhenAll(tasks);
	Console.WriteLine("所有线程执行完成");
	}
	}
代码逐行讲解：
1.RedisDistributedLock构造函数：初始化Redis连接、锁Key、唯一值（UUID）、锁超时时间；
2.AcquireAsync：用StringSetAsync的When.NotExists参数实现NX原子操作，获取锁成功后启动续约定时器；
3.续约定时器：每10秒续约一次锁，避免业务操作超时导致锁提前释放；
4.ReleaseAsync：用Lua脚本原子执行“判断唯一值是否是自己的，如果是则删除锁”，避免并发释放锁时误删别人的锁；
5.Lua脚本的作用：Redis的Lua脚本是原子执行的，避免判断和删除锁之间的并发问题；
6.测试代码：模拟10个并发线程获取锁，只有一个线程能获取到锁，其他线程获取失败。
我踩过的坑：一开始没做锁续约，业务操作需要5秒，锁超时时间设为3秒，结果锁提前释放，多个线程同时获取锁——业务操作时间超过锁超时时间时，必须做锁续约！
拓展知识：Redis锁的进阶问题
1.锁续约的必要性：当业务操作时间超过锁超时时间时，锁会自动释放，导致多个线程同时获取锁，必须用定时器续约锁；
2.红锁算法（RedLock）：解决Redis单点故障的问题，用多个Redis实例（比如5个），获取超过半数的锁才算成功，适合对一致性要求极高的场景；
3.Redis集群的锁问题：Redis集群的SET命令不是原子的，可能导致锁在多个节点不一致，生产环境建议用RedLock或Redis单实例（加哨兵做高可用）；
4.锁的粒度：锁的粒度要尽量小，比如用stock:lock:product_123而不是stock:lock，避免锁竞争太激烈；
5.锁的重试机制：获取锁失败时，可以重试几次，比如用循环+延迟重试，避免直接失败。
四、ZooKeeper锁：高一致性的分布式锁，适合高可靠场景
ZooKeeper锁是基于ZooKeeper的临时有序节点实现的分布式锁，核心优势是高一致性、自动释放锁，适合高可靠场景（比如分布式事务、分布式任务调度）。
核心原理（大白话）
用ZooKeeper的临时有序节点实现：
1.获取锁：在ZooKeeper的指定节点下创建一个临时有序子节点（比如/lock/stock_123/lock-0000000001）；
2.判断是否获取到锁：获取当前节点下的所有子节点，判断自己的节点是否是最小的，如果是则获取到锁；如果不是，则监听前一个节点的删除事件；
3.释放锁：删除自己的临时节点（节点崩溃后ZooKeeper会自动删除临时节点，避免死锁）；
4.锁等待：当前一个节点被删除时，收到通知后再次判断自己是否是最小节点，直到获取到锁。
类比：ZooKeeper锁就像排队买票，每个人拿一个号（临时有序节点），号最小的人先买票；买完票的人离开（删除节点），下一个号最小的人收到通知后买票。
实战2：ZooKeeper锁的C#实现（Curator.NET）
用Curator.NET库实现ZooKeeper锁，Curator是ZooKeeper的.NET客户端，封装了分布式锁的实现。
步骤1：安装NuGet包
bash 
	Install-Package Curator.NET
	Install-Package Curator.NET.Recipes
步骤2：ZooKeeper锁实现代码
csharp 
	using CuratorNet.Client;
	using CuratorNet.Client.Framework;
	using CuratorNet.Recipes.Locks;
	using System;
	using System.Threading.Tasks;
	
	namespace DistributedLock.ZooKeeper;
	
	public class ZooKeeperDistributedLock : IDisposable
	{
	private readonly InterProcessMutex _mutex;
	private readonly ICuratorFramework _client;
	private bool _isLockHeld;
	
	public ZooKeeperDistributedLock(string zkConnectionString, string lockPath)
	{
	// 1. 创建Curator客户端
	_client = CuratorFrameworkFactory.NewClient(
	zkConnectionString,
	new ExponentialBackoffRetry(1000, 3)); // 重试策略：指数退避，最多重试3次
	_client.Start(); // 启动客户端
	
	// 2. 创建可重入互斥锁（InterProcessMutex）
	_mutex = new InterProcessMutex(_client, lockPath);
	}
	
	/// <summary>
	/// 获取锁（异步）
	/// </summary>
	/// <param name="timeout">获取锁的超时时间</param>
	/// <returns>是否获取成功</returns>
	public async Task<bool> AcquireAsync(TimeSpan timeout)
	{
	try
	{
	// 3. 异步获取锁，超时返回false
	bool acquired = await _mutex.AcquireAsync(timeout);
	_isLockHeld = acquired;
	if (acquired)
	{
	Console.WriteLine($"[ZooKeeperLock] 成功获取锁：{_mutex.Path}");
	}
	else
	{
	Console.WriteLine($"[ZooKeeperLock] 获取锁超时：{_mutex.Path}");
	}
	return acquired;
	}
	catch (Exception ex)
	{
	Console.WriteLine($"[ZooKeeperLock] 获取锁时出错：{ex.Message}");
	return false;
	}
	}
	
	/// <summary>
	/// 释放锁（异步）
	/// </summary>
	public async Task ReleaseAsync()
	{
	if (!_isLockHeld)
	return;
	
	try
	{
	// 4. 异步释放锁
	await _mutex.ReleaseAsync();
	Console.WriteLine($"[ZooKeeperLock] 成功释放锁：{_mutex.Path}");
	_isLockHeld = false;
	}
	catch (Exception ex)
	{
	Console.WriteLine($"[ZooKeeperLock] 释放锁时出错：{ex.Message}");
	}
	}
	
	public void Dispose()
	{
	ReleaseAsync().Wait();
	_client.Close();
	}
	}
	
	// 测试代码
	class Program
	{
	static async Task Main(string[] args)
	{
	string zkConnectionString = "localhost:2181";
	string lockPath = "/lock/stock/product_123";
	
	// 模拟5个并发线程获取锁
	var tasks = new Task[5];
	for (int i = 0; i < 5; i++)
	{
	int threadId = i;
	tasks[i] = Task.Run(async () =>
	{
	using var zkLock = new ZooKeeperDistributedLock(zkConnectionString, lockPath);
	bool acquired = await zkLock.AcquireAsync(TimeSpan.FromSeconds(10));
	if (acquired)
	{
	try
	{
	Console.WriteLine($"[线程{threadId}] 开始执行分布式任务...");
	await Task.Delay(3000); // 模拟3秒的业务操作
	Console.WriteLine($"[线程{threadId}] 分布式任务执行完成");
	}
	finally
	{
	await zkLock.ReleaseAsync();
	}
	}
	else
	{
	Console.WriteLine($"[线程{threadId}] 获取锁失败，跳过任务执行");
	}
	});
	}
	
	await Task.WhenAll(tasks);
	Console.WriteLine("所有线程执行完成");
	}
	}

代码逐行讲解：
1.CuratorFrameworkFactory.NewClient：创建ZooKeeper客户端，配置重试策略（指数退避，最多重试3次）；
2.InterProcessMutex：Curator封装的可重入互斥锁，基于ZooKeeper的临时有序节点实现；
3.AcquireAsync：异步获取锁，超时时间10秒，获取成功返回true；
4.ReleaseAsync：异步释放锁，Curator会自动删除临时节点；
5.临时有序节点的作用：
临时节点：客户端崩溃后ZooKeeper会自动删除节点，避免死锁；
有序节点：保证锁的公平性，按顺序获取锁；
6.测试代码：模拟5个并发线程获取锁，只有一个线程能获取到锁，其他线程等待前一个线程释放锁后获取。
我踩过的坑：一开始用Curator的InterProcessSemaphoreMutex（不可重入锁），结果同一个线程多次获取锁失败，导致业务操作无法执行——要根据业务场景选择可重入锁（InterProcessMutex）或不可重入锁（InterProcessSemaphoreMutex）！
拓展知识：ZooKeeper锁的进阶问题
1.公平锁vs非公平锁：Curator的InterProcessMutex是公平锁，按顺序获取锁；也可以用InterProcessReadWriteLock实现读写锁，读锁共享，写锁互斥；
2.ZooKeeper集群的高可用性：ZooKeeper集群需要奇数个节点（比如3个、5个），保证集群的高可用性，当超过半数节点正常时，集群就能正常工作；
3.锁的性能：ZooKeeper锁的性能比Redis锁低，因为需要创建节点、监听节点，适合对一致性要求高但并发量不高的场景；
4.Curator的其他锁实现：
InterProcessReadWriteLock：读写锁，适合读多写少的场景；
InterProcessMultiLock：多锁，同时获取多个锁，适合需要多个资源的场景；
InterProcessSemaphoreV2：信号量，控制并发访问的数量。
四、Redis锁vs ZooKeeper锁：对比与适用场景
特性 Redis锁 ZooKeeper锁
性能 高（十万级QPS） 中（万级QPS）
一致性 最终一致性（可能存在锁丢失） 强一致性（ZooKeeper的CP模式）
死锁避免 依赖过期时间，可能存在锁提前释放 依赖临时节点，客户端崩溃自动释放锁
实现复杂度 低（自己实现或用库） 中（推荐用Curator库）
适用场景 高并发场景（秒杀、库存扣减） 高可靠场景（分布式事务、任务调度）
高可用性 依赖Redis哨兵或集群 依赖ZooKeeper集群（奇数个节点）
锁续约 需要自己实现定时器续约 不需要，ZooKeeper自动维护节点
选择建议
高并发、对一致性要求不是极高：用Redis锁；
高可靠、对一致性要求极高：用ZooKeeper锁；
读多写少：用ZooKeeper的读写锁；
需要公平锁：用ZooKeeper锁；
需要高性能：用Redis锁。
五、分布式锁最佳实践

1. 避免死锁
   Redis锁：必须设置过期时间，业务操作超时要做锁续约；
   ZooKeeper锁：用临时节点，客户端崩溃后自动释放锁；
   所有锁：必须在finally块中释放锁，避免业务操作抛出异常导致锁未释放。
2. 避免误删锁
   Redis锁：必须用唯一的unique_value，释放锁时用Lua脚本原子判断并删除；
   ZooKeeper锁：只删除自己创建的节点，避免误删别人的节点。
3. 性能优化
   锁的粒度要小：比如用stock:lock:product_123而不是stock:lock；
   减少锁的持有时间：锁内只做必要的业务操作，避免耗时操作；
   锁的重试机制：获取锁失败时，用循环+延迟重试（比如100ms重试一次），避免直接失败；
   Redis锁的批量操作：用Redis的Pipeline减少网络请求，提升性能。
4. 生产环境部署
   Redis锁：用Redis哨兵或集群做高可用，避免单点故障；用RedLock算法提升一致性；
   ZooKeeper锁：用3个或5个节点的ZooKeeper集群，保证高可用性；
   监控告警：监控锁的获取成功率、持有时间、等待时间，及时发现锁竞争激烈的问题；
   日志记录：记录锁的获取、释放、续约的日志，方便排查问题。
5. 踩过的坑总结
   Redis锁的过期时间设置不合理：过期时间太短导致锁提前释放，太长导致死锁时间长——过期时间要根据业务操作的平均时间设置，比如业务操作平均5秒，过期时间设为30秒，加上锁续约；
   ZooKeeper锁的节点路径冲突：多个业务用同一个节点路径，导致锁冲突——每个业务的锁路径要唯一，比如/lock/stock/product_123、/lock/order/order_456；
   锁的重入问题：同一个线程多次获取锁失败，导致业务操作无法执行——用可重入锁（Redis锁要记录线程的unique_value，ZooKeeper用InterProcessMutex）；
   锁的并发竞争激烈：大量线程等待锁，导致系统性能下降——用分段锁（比如把库存分成10段，每段一个锁），减少锁竞争。
   六、总结
   Redis锁是高性能的分布式锁，适合高并发场景；ZooKeeper锁是高一致性的分布式锁，适合高可靠场景。选择合适的分布式锁，遵循最佳实践，能解决分布式系统中的并发控制问题，保证数据的一致性。
   下一节我们会学习分布式缓存：Redis缓存的设计与优化，解决分布式系统中的数据缓存问题。