-
Notifications
You must be signed in to change notification settings - Fork 43
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #126 from netcorepal/command-lock
feat:commandlock
- Loading branch information
Showing
30 changed files
with
927 additions
and
105 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,67 @@ | ||
| # 命令锁 | ||
| 顾名思义，命令锁是为了解决命令并发执行的问题。在一些场景下，我们需要保证某个命令在同一时间只能被一个实例执行，这时候就可以使用命令锁。 | ||
| 本质上命令锁是一种分布式锁，它的实现方式有很多种，我们默认提供了基于Redis来实现的命令锁。 | ||
|
|
||
| ## 注册命令锁 | ||
|
|
||
| 在Program.cs注册`CommandLocks`： | ||
| ```csharp | ||
| builder.Services.AddMediatR(cfg => | ||
| cfg.RegisterServicesFromAssemblies(Assembly.GetExecutingAssembly()) | ||
| .AddCommandLockBehavior() //注册命令锁行为 | ||
| .AddUnitOfWorkBehaviors() | ||
| .AddKnownExceptionValidationBehavior()); | ||
|
|
||
| builder.Services.AddCommandLocks(typeof(Program).Assembly); //注册所有的命令锁类型 | ||
| ``` | ||
|
|
||
| 注意： 命令锁应该在事务开启前执行，所以需要在`AddUnitOfWorkBehaviors`之前添加`AddCommandLockBehavior`。 | ||
|
|
||
|
|
||
| ## 使用命令锁 | ||
|
|
||
| 定义一个命令锁，实现`ICommandLock<TCommand>`接口，其中`TCommand`是命令类型，例如： | ||
|
|
||
| ```csharp | ||
| public record PayOrderCommand(OrderId Id) : ICommand<OrderId>; | ||
|
|
||
| public class PayOrderCommandLock : ICommandLock<PayOrderCommand> | ||
| { | ||
| public Task<CommandLockSettings> GetLockKeysAsync(PayOrderCommand command, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| return Task.FromResult(command.Id.ToCommandLockSettings()); | ||
| } | ||
| } | ||
| ``` | ||
|
|
||
| 其中`command.Id.ToCommandLockSettings()`是将`OrderId`转换为`CommandLockSettings`，`CommandLockSettings`是命令锁的配置，包含了锁的Key、获取锁之前可以等待的过期时间。 | ||
|
|
||
| 设计上，命令锁与命令是一对一关系，建议将命令锁与命令、命令处理器放在同一个类文件中，便于维护。 | ||
|
|
||
| ## 多key命令锁 | ||
|
|
||
| 命令锁支持多Key机制，即一个命令可以对应多个Key，例如： | ||
|
|
||
| ```csharp | ||
|
|
||
| public class PayOrderCommandLock : ICommandLock<PayOrderCommand> | ||
| { | ||
| public Task<CommandLockSettings> GetLockKeysAsync(PayOrderCommand command, | ||
| CancellationToken cancellationToken = default) | ||
| { | ||
| var ids = new List<OrderId> { new OrderId(1), new OrderId(2) }; | ||
| return Task.FromResult(ids.ToCommandLockSettings()); | ||
| } | ||
| } | ||
| ``` | ||
|
|
||
| 在这个例子中，`PayOrderCommand`对应两个Key，分别是`OrderId(1)`和`OrderId(2)`。 | ||
|
|
||
| 当需要锁定多个Key时，CommandLockSettings会对多个Key进行排序，然后逐个锁定，如果其中一个Key锁定失败，则会释放已经锁定的Key。 | ||
|
|
||
|
|
||
| ## 可重入机制 | ||
|
|
||
| 命令锁实现了可重入机制，即在同一个请求上下文中，相同的Key可以重复获取锁，不会造成死锁。 | ||
| 例如上面示例的命令执行后序的事件处理过程中再次执行携带相同Key的命令锁，不会死锁。 |
File renamed without changes.
File renamed without changes.
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,103 @@ | ||
| using MediatR; | ||
| using NetCorePal.Extensions.DistributedLocks; | ||
| using NetCorePal.Extensions.Primitives; | ||
|
|
||
| namespace NetCorePal.Extensions.AspNetCore.CommandLocks; | ||
|
|
||
| public class CommandLockBehavior<TRequest, TResponse>( | ||
| IEnumerable<ICommandLock<TRequest>> commandLocks, | ||
| IDistributedLock distributedLock) | ||
| : IPipelineBehavior<TRequest, TResponse> | ||
| where TRequest : IBaseCommand | ||
| { | ||
| #pragma warning disable S3604 | ||
| private readonly CommandLockedKeysHolder _lockedKeys = new(); | ||
| #pragma warning restore S3604 | ||
|
|
||
| public async Task<TResponse> Handle(TRequest request, RequestHandlerDelegate<TResponse> next, | ||
| CancellationToken cancellationToken) | ||
| { | ||
| var count = commandLocks.Count(); | ||
| if (count == 0) | ||
| { | ||
| return await next(); | ||
| } | ||
|
|
||
| if (count > 1) | ||
| { | ||
| throw new InvalidOperationException("Only one ICommandLock<TRequest> is allowed"); | ||
| } | ||
|
|
||
| var commandLock = commandLocks.First(); | ||
| var options = await commandLock.GetLockKeysAsync(request, cancellationToken); | ||
| if (!string.IsNullOrEmpty(options.LockKey)) | ||
| { | ||
| if (!_lockedKeys.LockedKeys.Keys.Contains(options.LockKey)) | ||
| { | ||
| await using var lockHandler = | ||
| await TryAcquireAsync(options.LockKey, timeout: options.AcquireTimeout, | ||
| cancellationToken: cancellationToken); | ||
| if (lockHandler == null) | ||
| { | ||
| throw new CommandLockFailedException("Acquire Lock Faild", options.LockKey); | ||
| } | ||
|
|
||
| _lockedKeys.LockedKeys.Keys.Add(options.LockKey); | ||
| // 确保在执行next后，释放锁 | ||
| return await next(); | ||
| } | ||
| else | ||
| { | ||
| return await next(); | ||
| } | ||
| } | ||
| else | ||
| { | ||
| return await LockAndRelease(options, 0, next, cancellationToken); | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private async Task<TResponse> LockAndRelease(CommandLockSettings settings, int lockIndex, | ||
| RequestHandlerDelegate<TResponse> next, CancellationToken cancellationToken) | ||
| { | ||
| if (lockIndex >= settings.LockKeys!.Count) | ||
| { | ||
| return await next(); | ||
| } | ||
|
|
||
| var key = settings.LockKeys[lockIndex]; | ||
|
|
||
| if (!_lockedKeys.LockedKeys.Keys.Contains(key)) | ||
| { | ||
| await using var lockHandler = | ||
| await TryAcquireAsync(key, timeout: settings.AcquireTimeout, | ||
| cancellationToken: cancellationToken); | ||
| if (lockHandler == null) | ||
| { | ||
| throw new CommandLockFailedException("Acquire Lock Faild", key); | ||
| } | ||
|
|
||
| _lockedKeys.LockedKeys.Keys.Add(key); | ||
| // 确保在执行LockAndRelease后，释放锁 | ||
| return await LockAndRelease(settings, lockIndex + 1, next, cancellationToken); | ||
| } | ||
| else | ||
| { | ||
| return await LockAndRelease(settings, lockIndex + 1, next, cancellationToken); | ||
| } | ||
| } | ||
|
|
||
| private async Task<LockSynchronizationHandlerWarpper?> TryAcquireAsync(string key, TimeSpan timeout, | ||
| CancellationToken cancellationToken) | ||
| { | ||
| var handler = await distributedLock.TryAcquireAsync(key, timeout: timeout, | ||
| cancellationToken: cancellationToken); | ||
| if (handler == null) | ||
| { | ||
| return null; | ||
| } | ||
|
|
||
| return new LockSynchronizationHandlerWarpper(key, _lockedKeys.LockedKeys.Keys, handler); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| namespace NetCorePal.Extensions.AspNetCore.CommandLocks; | ||
|
|
||
| /// <inheritdoc /> | ||
| #pragma warning disable S3925 | ||
| public sealed class CommandLockFailedException : Exception | ||
| #pragma warning restore S3925 | ||
| { | ||
| public CommandLockFailedException(string message, string failedKey) : base(message) | ||
| { | ||
| FailedKey = failedKey; | ||
| } | ||
|
|
||
| public string FailedKey { get; private set; } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| namespace NetCorePal.Extensions.AspNetCore.CommandLocks; | ||
|
|
||
| class CommandLockedKeysHolder | ||
| { | ||
| private static readonly AsyncLocal<KeysHolder> KeysHolderCurrent = new AsyncLocal<KeysHolder>(); | ||
|
|
||
| public KeysHolder LockedKeys => KeysHolderCurrent.Value ??= new KeysHolder(); | ||
| } | ||
|
|
||
| class KeysHolder | ||
| { | ||
| public HashSet<string> Keys { get; set; } = new HashSet<string>(); | ||
| } |
19 changes: 19 additions & 0 deletions
19
src/AspNetCore/CommandLocks/LockSynchronizationHandlerWarpper.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| using NetCorePal.Extensions.DistributedLocks; | ||
|
|
||
| namespace NetCorePal.Extensions.AspNetCore.CommandLocks; | ||
|
|
||
| /// <summary> | ||
| /// 在释放锁后，从持有者集合中移除key | ||
| /// </summary> | ||
| /// <param name="key"></param> | ||
| /// <param name="holder"></param> | ||
| /// <param name="handler"></param> | ||
| class LockSynchronizationHandlerWarpper(string key, HashSet<string> holder, ILockSynchronizationHandler handler) | ||
| : IAsyncDisposable | ||
| { | ||
| public async ValueTask DisposeAsync() | ||
| { | ||
| await handler.DisposeAsync(); | ||
| holder.Remove(key); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| using NetCorePal.Extensions.Domain; | ||
| using NetCorePal.Extensions.Primitives; | ||
|
|
||
| namespace NetCorePal.Extensions.Primitives; | ||
|
|
||
| public static class EntityIdExtensions | ||
| { | ||
| public static CommandLockSettings ToCommandLockSettings<TId>(this TId id, | ||
| int acquireSeconds = 10) | ||
| where TId : IEntityId | ||
| { | ||
| return new CommandLockSettings(typeof(TId).Name + "-" + id.ToString()!, acquireSeconds: acquireSeconds); | ||
| } | ||
|
|
||
| public static CommandLockSettings ToCommandLockSettings<TId>(this IEnumerable<TId> ids, | ||
| int acquireSeconds = 10) | ||
| where TId : IEntityId | ||
| { | ||
| var typeName = typeof(TId).Name; | ||
| return new CommandLockSettings(ids.Select(id => typeName + "-" + id.ToString()), | ||
| acquireSeconds: acquireSeconds); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.