-
C#网络编程之分布式缓存（Redis、Memcached）
第56章 分布式锁实战
56.1 分布式锁（Redis锁、ZooKeeper锁）
一、我踩过的分布式锁坑：从“超卖500单”到“锁续期救了我”
做电商618大促时，我用Redis锁做库存扣减，结果因为锁超时时vb.net教程C#教程python教程SQL教程access 2010教程间设短了，服务还没处理完库存，锁就自动释放了，导致两个服务实例同时扣减库存，超卖了500单；后来换成Redisson的看门狗机制自动续期锁，才解决了这个问题。还有一次用ZooKeeper锁做分布式任务调度，结果因为ZooKeeper集群网络抖动，锁被意外释放，任务重复执行——后来加了锁的重入性和异常处理，才稳定下来。这节我把这些血泪经验揉进去，用大白话讲透Redis锁和ZooKeeper锁的核心原理，结合C#实战代码逐行拆解，拓展生产级优化技巧，让你在高并发场景下再也不踩分布式锁的坑！
二、分布式锁核心原理：让多个服务“排队”操作同一个资源
分布式锁是分布式系统中控制多个服务实例并发访问共享资源的工具，核心是“同一时间只有一个服务实例能获得锁，操作共享资源”。
大白话解释：把分布式锁比作“厕所隔间的门”
1.共享资源：厕所隔间（比如电商的库存、数据库的同一行数据）；
2.分布式锁：厕所门的锁（只有拿到钥匙的人才能进去）；
3.服务实例：排队上厕所的人（多个服务实例）；
4.核心要求：
1.互斥性：同一时间只有一个人能进隔间；
2.避免死锁：人进去后不能一直不出来（锁要超时自动释放）；
3.容错性：人突然晕倒了，锁要自动打开（服务挂了锁要释放）；
4.原子性：拿钥匙和锁门必须是一个原子操作（不能拿了钥匙还没锁门，别人就进去了）。
我踩过的坑：一开始用Redis锁时，先判断锁是否存在，不存在再设置锁，结果高并发下两个服务实例同时判断锁不存在，都设置了锁，导致超卖——这就是因为操作不是原子性的，必须用Redis的SETNX命令（原子操作）！
三、Redis锁：高性能分布式锁，适合高并发场景
Redis锁是基于Redis的SETNX命令实现的分布式锁，核心是“原子性设置锁+超时自动释放”，性能高（QPS可达10万+），适合高并发场景（如电商库存扣减、秒杀活动）。
核心原理（大白话+厕所例子）
1.获取锁：用Redis的SET lock_key value NX EX 30命令，意思是“如果lock_key不存在，就设置它的值为value，过期时间30秒”——相当于拿钥匙锁门，原子操作，同一时间只有一个人能成功；
2.释放锁：用Redis的DEL命令删除lock_key——相当于开门把钥匙还给别人；
3.锁超时：30秒后锁自动释放，避免服务挂了锁一直不释放——相当于厕所门30秒后自动打开，防止有人在里面晕倒。
我踩过的坑：释放锁的时候直接用DEL命令，结果服务A的锁超时自动释放了，服务B拿到了锁，服务A处理完后又DEL了服务B的锁，导致服务C又拿到了锁，并发问题又出现了——必须用Lua脚本保证释放锁的原子性，只有锁的value是自己设置的，才能释放！
实战1：C#实现Redis锁（StackExchange.Redis）
步骤1：安装NuGet包
bash
Install-Package StackExchange.Redis
步骤2：Redis锁核心代码（逐行讲解）
csharp
using StackExchange.Redis;
using System;
using System.Threading.Tasks;
namespace DistributedLock.RedisLock;
public class RedisDistributedLock
{
private readonly IConnectionMultiplexer _redis;
private readonly IDatabase _db;
private readonly string _lockKey;
private readonly string _lockValue; // 唯一标识，避免误删别人的锁
private readonly TimeSpan _lockTimeout; // 锁超时时间
private bool _isLockAcquired = false;
public RedisDistributedLock(IConnectionMultiplexer redis, string lockKey, TimeSpan lockTimeout)
{
_redis = redis;
_db = redis.GetDatabase();
_lockKey = lockKey;
_lockTimeout = lockTimeout;
_lockValue = Guid.NewGuid().ToString("N"); // 生成唯一锁值，用于释放锁时的校验
}
/// <summary>
/// 获取分布式锁（原子操作）
/// </summary>
public async Task<bool> AcquireAsync()
{
// 1. 用Redis的SET命令，NX表示只有锁不存在时才设置，EX表示过期时间
// 原子操作：判断锁不存在+设置锁+设置过期时间，一步完成，避免并发问题
_isLockAcquired = await _db.StringSetAsync(
_lockKey,
_lockValue,
_lockTimeout,
When.NotExists); // When.NotExists = NX参数
if (_isLockAcquired)
{
Console.WriteLine($"成功获取锁：{_lockKey}，锁值：{_lockValue}");
}
else
{
string existingValue = await _db.StringGetAsync(_lockKey);
Console.WriteLine($"获取锁失败：{_lockKey}，当前锁值：{existingValue}");
}
return _isLockAcquired;
}
/// <summary>
/// 释放分布式锁（原子操作，避免误删别人的锁）
/// </summary>
public async Task ReleaseAsync()
{
if (!_isLockAcquired)
{
Console.WriteLine("未获取锁，无需释放");
return;
}
try
{
// 2. 用Lua脚本释放锁，保证原子性：只有锁值等于自己设置的，才删除锁
string luaScript = @"
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('DEL', KEYS[1])
else
return 0
end";
// 执行Lua脚本
long result = await _db.ScriptEvaluateAsync(
luaScript,
new RedisKey[] { _lockKey },
new RedisValue[] { _lockValue });
if (result == 1)
{
Console.WriteLine($"成功释放锁：{_lockKey}");
_isLockAcquired = false;
}
else
{
Console.WriteLine($"释放锁失败：{_lockKey}，锁值已变化或已过期");
}
}
catch (Exception ex)
{
Console.WriteLine($"释放锁异常：{ex.Message}");
_isLockAcquired = false;
}
}
/// <summary>
/// 自动续期锁（看门狗机制）
/// </summary>
public async Task StartAutoRenewAsync(CancellationToken cancellationToken)
{
if (!_isLockAcquired)
{
Console.WriteLine("未获取锁，无需续期");
return;
}
// 3. 每10秒续期一次，把锁超时时间重置为原来的时间
while (!cancellationToken.IsCancellationRequested)
{
string luaScript = @"
if redis.call('GET', KEYS[1]) == ARGV[1] then
return redis.call('EXPIRE', KEYS[1], ARGV[2])
else
return 0
end";
long result = await _db.ScriptEvaluateAsync(
luaScript,
new RedisKey[] { _lockKey },
new RedisValue[] { _lockValue, (int)_lockTimeout.TotalSeconds });
if (result == 0)
{
Console.WriteLine($"锁续期失败：{_lockKey}，锁已释放或被其他实例获取");
_isLockAcquired = false;
break;
}
Console.WriteLine($"锁续期成功：{_lockKey}，剩余超时时间：{_lockTimeout.TotalSeconds}秒");
await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken); // 每10秒续期一次
}
}
}
// 测试代码：模拟电商库存扣减
class Program
{
static async Task Main(string[] args)
{
// 1. 创建Redis连接复用器（生产环境用单例）
var redis = ConnectionMultiplexer.Connect("localhost:6379");
string lockKey = "inventory_lock:product_123";
var lockTimeout = TimeSpan.FromSeconds(30);
// 2. 模拟10个并发服务实例
var tasks = new List<Task>();
for (int i = 0; i < 10; i++)
{
int instanceId = i;
tasks.Add(Task.Run(async () =>
{
var distributedLock = new RedisDistributedLock(redis, lockKey, lockTimeout);
try
{
// 3. 获取锁
if (await distributedLock.AcquireAsync())
{
// 4. 启动锁续期（看门狗机制）
using var cts = new CancellationTokenSource();
var renewTask = distributedLock.StartAutoRenewAsync(cts.Token);
// 5. 操作共享资源：扣减库存
Console.WriteLine($"实例 {instanceId} 开始扣减库存...");
await Task.Delay(TimeSpan.FromSeconds(25)); // 模拟处理时间25秒，超过锁超时时间的一半，触发续期
Console.WriteLine($"实例 {instanceId} 库存扣减完成");
// 6. 停止续期
cts.Cancel();
await renewTask;
}
}
catch (Exception ex)
{
Console.WriteLine($"实例 {instanceId} 异常：{ex.Message}");
}
finally
{
// 7. 释放锁
await distributedLock.ReleaseAsync();
}
}));
}
await Task.WhenAll(tasks);
redis.Dispose();
}
}
核心代码拆解：
1.原子获取锁：用SET lock_key value NX EX 30命令，保证判断锁不存在和设置锁是原子操作，避免高并发下多个服务实例同时获取锁；
2.原子释放锁：用Lua脚本释放锁，只有锁值等于自己设置的才删除，避免误删别人的锁；
3.看门狗机制：每10秒续期一次锁，把锁超时时间重置为原来的时间，避免服务处理时间超过锁超时时间，锁被自动释放；
4.锁超时时间：设置合理的锁超时时间（比如30秒），避免服务挂了锁一直不释放。
拓展知识：Redis锁的常见问题与解决方案
| 问题 | 原因 | 解决方案 |
|---|---|---|
| 超卖（多个实例同时获取锁） | 获取锁的操作不是原子性的（先判断再设置） | 用Redis的SETNX命令（原子操作） |
| 误删别人的锁 | 释放锁时直接DEL，锁已经超时被其他实例获取 | 用Lua脚本释放锁，校验锁值是否是自己设置的 |
| 锁提前释放导致并发问题 | 服务处理时间超过锁超时时间，锁被自动释放 | 用看门狗机制自动续期锁（比如Redisson的看门狗） |
| 死锁 | 服务挂了锁没释放 | 设置锁超时时间，自动释放锁 |
| 锁粒度太大 | 锁覆盖了太多资源，导致并发低 | 缩小锁粒度（比如用商品ID作为锁键，而不是整个库存） |
我踩过的坑：一开始用“inventory_lock”作为锁键，所有商品的库存扣减都用同一个锁，导致并发从1000降到10——后来改成“inventory_lock:product_123”，每个商品用自己的锁，并发恢复到1000！
四、ZooKeeper锁：高可靠分布式锁，适合一致性要求高的场景
ZooKeeper锁是基于ZooKeeper的临时有序节点实现的分布式锁，核心是“临时有序节点+监听前一个节点”，可靠性高（不会出现锁丢失），适合对一致性要求高的场景（如分布式任务调度、分布式事务）。
核心原理（大白话+排队买票例子）
1.获取锁：每个服务实例在ZooKeeper的指定节点下创建一个临时有序节点（比如/lock/node-0000000001），然后判断自己的节点是否是最小的，如果是就获得锁；如果不是，就监听前一个节点的删除事件；
2.释放锁：服务实例完成操作后，删除自己的临时节点，或者服务挂了之后，ZooKeeper自动删除临时节点；
3.监听机制：当前一个节点被删除时，ZooKeeper会通知下一个节点，下一个节点判断自己是否是最小的，如果是就获得锁。
我踩过的坑：用ZooKeeper锁时，一开始自己实现监听逻辑，结果因为网络抖动，监听事件丢失，导致锁一直不释放——后来用Curator.NET库，它封装了ZooKeeper的复杂操作，解决了监听事件丢失的问题！
实战2：C#实现ZooKeeper锁（Curator.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.ZooKeeperLock;
public class ZooKeeperDistributedLock
{
private readonly IInterProcessMutex _lock;
private readonly string _lockPath;
public ZooKeeperDistributedLock(string zooKeeperConnectionString, string lockPath)
{
_lockPath = lockPath;
// 1. 创建Curator客户端（生产环境用单例）
var client = CuratorFrameworkFactory.NewClient(
zooKeeperConnectionString,
new ExponentialBackoffRetry(1000, 3)); // 重试策略：指数退避，最多重试3次
client.Start();
client.BlockUntilConnected(TimeSpan.FromSeconds(10)); // 等待连接成功
// 2. 创建分布式锁（InterProcessMutex是可重入的公平锁）
_lock = new InterProcessMutex(client, lockPath);
Console.WriteLine($"ZooKeeper锁已初始化，锁路径：{lockPath}");
}
/// <summary>
/// 获取分布式锁（可重入，公平锁）
/// </summary>
public async Task<bool> AcquireAsync(TimeSpan timeout)
{
try
{
// 3. 获取锁，超时返回false
bool acquired = await _lock.AcquireAsync(timeout);
if (acquired)
{
Console.WriteLine($"成功获取ZooKeeper锁：{_lockPath}");
}
else
{
Console.WriteLine($"获取ZooKeeper锁超时：{_lockPath}");
}
return acquired;
}
catch (Exception ex)
{
Console.WriteLine($"获取ZooKeeper锁异常：{ex.Message}");
return false;
}
}
/// <summary>
/// 释放分布式锁（自动处理重入）
/// </summary>
public async Task ReleaseAsync()
{
try
{
if (_lock.IsAcquiredInThisProcess)
{
// 4. 释放锁，Curator自动处理重入次数
await _lock.ReleaseAsync();
Console.WriteLine($"成功释放ZooKeeper锁：{_lockPath}");
}
else
{
Console.WriteLine("未获取ZooKeeper锁，无需释放");
}
}
catch (Exception ex)
{
Console.WriteLine($"释放ZooKeeper锁异常：{ex.Message}");
}
}
/// <summary>
/// 检查是否已获取锁
/// </summary>
public bool IsAcquired()
{
return _lock.IsAcquiredInThisProcess;
}
}
// 测试代码：模拟分布式任务调度
class Program
{
static async Task Main(string[] args)
{
string zooKeeperConnectionString = "localhost:2181";
string lockPath = "/task_lock:daily_report";
var timeout = TimeSpan.FromSeconds(10);
// 模拟5个并发服务实例竞争锁
var tasks = new List<Task>();
for (int i = 0; i < 5; i++)
{
int instanceId = i;
tasks.Add(Task.Run(async () =>
{
var distributedLock = new ZooKeeperDistributedLock(zooKeeperConnectionString, lockPath);
try
{
if (await distributedLock.AcquireAsync(timeout))
{
// 执行分布式任务：生成日报
Console.WriteLine($"实例 {instanceId} 开始生成日报...");
await Task.Delay(TimeSpan.FromSeconds(15));
Console.WriteLine($"实例 {instanceId} 日报生成完成");
}
}
catch (Exception ex)
{
Console.WriteLine($"实例 {instanceId} 异常：{ex.Message}");
}
finally
{
await distributedLock.ReleaseAsync();
}
}));
}
await Task.WhenAll(tasks);
}
}
核心代码拆解：
1.Curator客户端：封装了ZooKeeper的连接、重试、监听等复杂操作，避免自己实现的各种坑；
2.InterProcessMutex：Curator提供的可重入公平锁，基于ZooKeeper的临时有序节点实现；
3.获取锁：用AcquireAsync方法获取锁，支持超时时间，超时返回false；
4.释放锁：用ReleaseAsync方法释放锁，Curator自动处理重入次数（比如同一个服务实例多次获取锁，只需要释放一次）；
5.临时节点：服务挂了之后，ZooKeeper自动删除临时节点，锁被释放，避免死锁。
拓展知识：Redis锁 vs ZooKeeper锁
| 特性 | Redis锁 | ZooKeeper锁 |
|---|---|---|
| 性能 | 高（QPS 10万+） | 低（QPS 1万左右） |
| 可靠性 | 中（可能出现锁丢失） | 高（不会出现锁丢失） |
| 实现复杂度 | 低（用SETNX命令） | 高（需要处理监听、节点顺序等） |
| 锁续期 | 需要自己实现（看门狗） | 自动续期（临时节点在会话断开时才删除） |
| 死锁风险 | 低（锁超时自动释放） | 无（临时节点自动删除） |
| 适用场景 | 高并发场景（电商库存、秒杀） | 高可靠场景（分布式任务调度、分布式事务） |
我踩过的坑：用Redis锁做分布式任务调度，结果因为Redis集群主从切换，锁丢失，任务重复执行——换成ZooKeeper锁后，任务再也没有重复执行过！
五、生产级分布式锁最佳实践与踩坑总结
-
生产级最佳实践
选择合适的锁：高并发场景用Redis锁，高可靠场景用ZooKeeper锁；
缩小锁粒度：用资源ID作为锁键（比如inventory_lock:product_123），避免锁覆盖太多资源；
设置合理的锁超时时间：根据服务处理时间设置锁超时时间（比如30秒），避免服务挂了锁一直不释放；
实现锁续期：用看门狗机制自动续期锁，避免服务处理时间超过锁超时时间；
原子性操作：获取锁和释放锁的操作必须是原子性的（Redis用SETNX和Lua脚本，ZooKeeper用Curator的InterProcessMutex）；
监控锁的使用：监控锁的获取成功率、等待时间、释放时间，设置告警阈值（比如锁等待时间超过10秒触发告警）；
避免锁的重入：除非必要，否则不要用可重入锁，增加复杂度；
公平锁 vs 非公平锁：高并发场景用非公平锁（Redis锁是非公平锁），提高并发；一致性要求高的场景用公平锁（ZooKeeper锁是公平锁），避免饥饿。 -
我踩过的坑总结
1.锁粒度太大：用“inventory_lock”作为锁键，所有商品的库存扣减都用同一个锁，并发从1000降到10——改成每个商品用自己的锁，并发恢复到1000；
2.锁超时时间设短了：服务处理时间25秒，锁超时时间20秒，锁被自动释放，导致超卖——改成30秒，加看门狗续期；
3.释放锁时没校验锁值：直接DEL锁，锁已经超时被其他实例获取，误删了别人的锁——用Lua脚本释放锁，校验锁值；
4.自己实现ZooKeeper锁：监听事件丢失，导致锁一直不释放——用Curator.NET库，封装了复杂的监听逻辑；
5.Redis集群主从切换导致锁丢失：主节点设置了锁，还没同步到从节点，主节点挂了，从节点变成主节点，锁丢失——用Redis的RedLock算法，或者换成ZooKeeper锁。
六、总结
分布式锁是分布式系统中控制并发访问共享资源的核心工具，Redis锁性能高适合高并发场景，ZooKeeper锁可靠性高适合高可靠场景。实战中要注意原子性操作、锁超时时间、锁续期、锁粒度等问题，避免踩坑。如果不想自己实现分布式锁，可以用成熟的库，比如Redis用Redisson，ZooKeeper用Curator.NET，这些库已经解决了各种边界问题，稳定性高。
下一节我们会学习分布式事务（TCC、Saga、消息最终一致性），解决分布式系统中多个服务的数据一致性问题。
转载请注明出处：https://www.xin3721.com/ArticlecSharp/c49574.html
