基于Redis的分布式锁和Redlock算法-662P.COM

在单进程的系统中，当存在多个线程可以同时改变某个变量(可变共享变量)时，就需要对变量或代码块做同步，使其在修改这种变量时能够线性执行消除并发修改变量。

本文转载自微信公众号「UP技术控」，作者conan5566 。转载本文请联系UP技术控公众号。

而同步的本质是通过锁来实现的。为了实现多个线程在一个时刻同一个代码块只能有一个线程可执行，那么需要在某个地方做个标记，这个标记必须每个线程都能看到，当标记不存在时可以设置该标记，其余后续线程发现已经有标记了则等待拥有标记的线程结束同步代码块取消标记后再去尝试设置标记。这个标记可以理解为锁。

不同地方实现锁的方式也不一样，只要能满足所有线程都能看得到标记即可。如 Java 中 synchronize 是在对象头设置标记，Lock 接口的实现类基本上都只是某一个 volitile 修饰的 int 型变量其保证每个线程都能拥有对该 int 的可见性和原子修改，linux 内核中也是利用互斥量或信号量等内存数据做标记。

除了利用内存数据做锁其实任何互斥的都能做锁(只考虑互斥情况)，如流水表中流水号与时间结合做幂等校验可以看作是一个不会释放的锁，或者使用某个文件是否存在作为锁等。只需要满足在对标记进行修改能保证原子性和内存可见性即可。

1 什么是分布式?

分布式的 CAP 理论告诉我们:

任何一个分布式系统都无法同时满足一致性(Consistency)、可用性(Availability)和分区容错性(Partition tolerance)，最多只能同时满足两项。

目前很多大型网站及应用都是分布式部署的，分布式场景中的数据一致性问题一直是一个比较重要的话题。基于 CAP理论，很多系统在设计之初就要对这三者做出取舍。在互联网领域的绝大多数的场景中，都需要牺牲强一致性来换取系统的高可用性，系统往往只需要保证最终一致性。

分布式场景

此处主要指集群模式下，多个相同服务同时开启.

在许多的场景中，我们为了保证数据的最终一致性，需要很多的技术方案来支持，比如分布式事务、分布式锁等。很多时候我们需要保证一个方法在同一时间内只能被同一个线程执行。在单机环境中，通过 Java 提供的并发 API 我们可以解决，但是在分布式环境下，就没有那么简单啦。

分布式与单机情况下最大的不同在于其不是多线程而是多进程。
多线程由于可以共享堆内存，因此可以简单的采取内存作为标记存储位置。而进程之间甚至可能都不在同一台物理机上，因此需要将标记存储在一个所有进程都能看到的地方。

什么是分布式锁?

当在分布式模型下，数据只有一份(或有限制)，此时需要利用锁的技术控制某一时刻修改数据的进程数。
与单机模式下的锁不仅需要保证进程可见，还需要考虑进程与锁之间的网络问题。(我觉得分布式情况下之所以问题变得复杂，主要就是需要考虑到网络的延时和不可靠。。。一个大坑)
分布式锁还是可以将标记存在内存，只是该内存不是某个进程分配的内存而是公共内存如 Redis、Memcache。至于利用数据库、文件等做锁与单机的实现是一样的，只要保证标记能互斥就行。

2 我们需要怎样的分布式锁?

可以保证在分布式部署的应用集群中，同一个方法在同一时间只能被一台机器上的一个线程执行。

这把锁要是一把可重入锁(避免死锁)

这把锁最好是一把阻塞锁(根据业务需求考虑要不要这条)

这把锁最好是一把公平锁(根据业务需求考虑要不要这条)

有高可用的获取锁和释放锁功能

获取锁和释放锁的性能要好

代码实现

public interface IDistributedLock
{
ILockResult Lock(string resourceKey);
ILockResult Lock(string resourceKey, TimeSpan expiryTime);
ILockResult Lock(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime);
ILockResult Lock(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime, CancellationToken cancellationToken);
Task<ILockResult> LockAsync(string resourceKey);
Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime);
Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime);
Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime, CancellationToken cancellationToken);
}
public interface ILockResult : IDisposable
{
string LockId { get; }
bool IsAcquired { get; }
int ExtendCount { get; }
}

class EndPoint:RedLock.RedisLockEndPoint
{
private readonly string _connectionString;
public EndPoint(string connectionString)
{
_connectionString = connectionString;
//139.196.40.252,password=xstudio,defaultDatabase=9
var connection = connectionString.Split(',');
var dict = new Dictionary<string, string>();
foreach (var item in connection)
{
var keypar = item.Split('=');
if (keypar.Length>1)
{
dict[keypar[0]] = keypar[1];
}
}
this.EndPoint = new System.Net.DnsEndPoint(connection[0], 6379);
if (dict.TryGetValue("password", out string password))
{
this.Password = password;
}
if (dict.TryGetValue("defaultDatabase", out string defaultDatabase) && int.TryParse(defaultDatabase,out int database))
{
RedisDatabase = database;
}
}
}

[Export(typeof(IDistributedLock))]
class InnerLock : IDistributedLock
{
private static Lazy<RedLock.RedisLockFactory> _factory;
static InnerLock()
{
_factory = new Lazy<RedisLockFactory>(() => new RedisLockFactory(new EndPoint(ConfigurationManager.AppSettings["Redis"])), System.Threading.LazyThreadSafetyMode.ExecutionAndPublication);
}
public ILockResult Lock(string resourceKey)
{
return new LockResult(_factory.Value.Create(resourceKey, TimeSpan.FromDays(1)));
}
public ILockResult Lock(string resourceKey, TimeSpan expiryTime)
{
return new LockResult(_factory.Value.Create(resourceKey, expiryTime));
}
public ILockResult Lock(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime)
{
return new LockResult(_factory.Value.Create(resourceKey, expiryTime, waitTime, retryTime));
}
public ILockResult Lock(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime, CancellationToken cancellationToken)
{
return new LockResult(_factory.Value.Create(resourceKey, expiryTime, waitTime, retryTime, cancellationToken));
}
public async Task<ILockResult> LockAsync(string resourceKey)
{
var result = await _factory.Value.CreateAsync(resourceKey, TimeSpan.FromDays(1));
return new LockResult(result);
}
public async Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime)
{
var result = await _factory.Value.CreateAsync(resourceKey, expiryTime);
return new LockResult(result);
}
public async Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime)
{
var result = await _factory.Value.CreateAsync(resourceKey, expiryTime, waitTime, retryTime);
return new LockResult(result);
}
public async Task<ILockResult> LockAsync(string resourceKey, TimeSpan expiryTime, TimeSpan waitTime, TimeSpan retryTime, CancellationToken cancellationToken)
{
var result = await _factory.Value.CreateAsync(resourceKey, expiryTime, waitTime, retryTime, cancellationToken);
return new LockResult(result);
}
}
class LockResult : ILockResult
{
private IRedisLock _lock;
public LockResult(IRedisLock redisLock)
{
_lock = redisLock;
}
public string LockId => _lock.LockId;
public bool IsAcquired => _lock.IsAcquired;
public int ExtendCount => _lock.ExtendCount;
public void Dispose()
{
_lock.Dispose();
}
}

算法

1 什么是分布式?

2 我们需要怎样的分布式锁?

0评论