using System.Collections.Concurrent;
|
using Aging.Infrastructure.Contracts;
|
using log4net;
|
|
namespace Aging.Infrastructure.Core
|
{
|
/// <summary>
|
/// 资源管理实现。
|
/// 提供线程安全的资源预约、租约获取和释放功能。
|
/// 使用 ConcurrentDictionary 存储活跃预约，使用 SemaphoreSlim 实现 per-resource 锁。
|
/// </summary>
|
public sealed class ResourceManager : IResourceManager
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(ResourceManager));
|
|
/// <summary>
|
/// 活跃预约字典（key = ResourceKey.ToString()）
|
/// </summary>
|
private readonly ConcurrentDictionary<string, ResourceReservation> _reservations = new();
|
|
/// <summary>
|
/// 每资源信号量字典，用于实现细粒度锁
|
/// </summary>
|
private readonly ConcurrentDictionary<string, SemaphoreSlim> _resourceLocks = new();
|
|
/// <summary>
|
/// 全局世代号，每次新预约递增
|
/// </summary>
|
private long _generation;
|
|
/// <summary>
|
/// 默认租约有效期（秒）
|
/// </summary>
|
private const int DefaultLeaseDurationSeconds = 30;
|
|
/// <inheritdoc/>
|
public async Task<ResourceReservation> ReserveAsync(
|
ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct)
|
{
|
if (owner == null) throw new ArgumentNullException(nameof(owner));
|
if (resource == null) throw new ArgumentNullException(nameof(resource));
|
|
var resourceKey = resource.ToString();
|
var semaphore = _resourceLocks.GetOrAdd(resourceKey, _ => new SemaphoreSlim(1, 1));
|
|
try
|
{
|
await semaphore.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
{
|
// 检查是否已有 Active 预约
|
if (_reservations.TryGetValue(resourceKey, out var existingReservation))
|
{
|
if (existingReservation.State == ReservationState.Active)
|
{
|
// 检查是否为同一 Owner（续期场景）
|
if (existingReservation.Owner.TokenId == owner.TokenId)
|
{
|
// 同一所有者续期：更新世代号
|
var renewedGeneration = Interlocked.Increment(ref _generation);
|
owner.Generation = renewedGeneration;
|
existingReservation.Owner.Generation = renewedGeneration;
|
|
Log.Info($"[ResourceManager] 资源 {resourceKey} 续期成功，新世代号: {renewedGeneration}");
|
return existingReservation;
|
}
|
|
// 不同所有者冲突
|
var conflictMsg = $"资源 {resourceKey} 已被所有者 {existingReservation.Owner.OwnerId} 预约，" +
|
$"当前请求者: {owner.OwnerId}";
|
Log.Warn($"[ResourceManager] 资源冲突: {conflictMsg}");
|
throw new ResourceConflictException(conflictMsg);
|
}
|
}
|
|
// 创建新预约
|
var newGeneration = Interlocked.Increment(ref _generation);
|
owner.Generation = newGeneration;
|
|
var reservation = new ResourceReservation(resource, owner)
|
{
|
State = ReservationState.Active
|
};
|
|
_reservations[resourceKey] = reservation;
|
|
Log.Info($"[ResourceManager] 资源 {resourceKey} 预约成功，所有者: {owner.OwnerId}，世代号: {newGeneration}");
|
return reservation;
|
}
|
finally
|
{
|
semaphore.Release();
|
}
|
}
|
catch (OperationCanceledException)
|
{
|
Log.Warn($"[ResourceManager] 资源 {resourceKey} 预约被取消");
|
throw;
|
}
|
catch (ResourceConflictException)
|
{
|
throw;
|
}
|
catch (Exception ex)
|
{
|
Log.Error($"[ResourceManager] 资源 {resourceKey} 预约失败: {ex.Message}", ex);
|
throw;
|
}
|
}
|
|
/// <inheritdoc/>
|
public async Task<IOperationLease> AcquireOperationAsync(
|
ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct)
|
{
|
if (owner == null) throw new ArgumentNullException(nameof(owner));
|
if (resource == null) throw new ArgumentNullException(nameof(resource));
|
|
var resourceKey = resource.ToString();
|
|
try
|
{
|
// 检查资源是否有有效预约
|
if (!_reservations.TryGetValue(resourceKey, out var reservation))
|
{
|
throw new InvalidOperationException($"资源 {resourceKey} 未被预约，无法获取操作租约");
|
}
|
|
if (reservation.State != ReservationState.Active)
|
{
|
throw new InvalidOperationException($"资源 {resourceKey} 预约状态为 {reservation.State}，无法获取操作租约");
|
}
|
|
// 验证 Owner 匹配（TokenId + Generation）
|
if (reservation.Owner.TokenId != owner.TokenId || reservation.Owner.Generation != owner.Generation)
|
{
|
throw new InvalidOperationException(
|
$"资源 {resourceKey} 所有者不匹配，期望 TokenId: {reservation.Owner.TokenId}，Generation: {reservation.Owner.Generation}");
|
}
|
|
// 创建 OperationLease（短期，默认30秒过期）
|
var lease = new OperationLease(
|
resource,
|
owner,
|
onRelease: () => Log.Debug($"[ResourceManager] 租约释放: {resourceKey}"),
|
leaseDurationSeconds: DefaultLeaseDurationSeconds);
|
|
Log.Info($"[ResourceManager] 获取操作租约: {resourceKey}，所有者: {owner.OwnerId}，有效期: {DefaultLeaseDurationSeconds}秒");
|
|
await Task.CompletedTask; // 保持异步签名
|
return lease;
|
}
|
catch (InvalidOperationException)
|
{
|
throw;
|
}
|
catch (Exception ex)
|
{
|
Log.Error($"[ResourceManager] 获取操作租约失败: {ex.Message}", ex);
|
throw;
|
}
|
}
|
|
/// <inheritdoc/>
|
public async Task ReleaseReservationAsync(
|
ResourceOwnerToken owner, ResourceKey resource, CancellationToken ct)
|
{
|
if (owner == null) throw new ArgumentNullException(nameof(owner));
|
if (resource == null) throw new ArgumentNullException(nameof(resource));
|
|
var resourceKey = resource.ToString();
|
var semaphore = _resourceLocks.GetOrAdd(resourceKey, _ => new SemaphoreSlim(1, 1));
|
|
try
|
{
|
await semaphore.WaitAsync(ct).ConfigureAwait(false);
|
|
try
|
{
|
if (!_reservations.TryGetValue(resourceKey, out var reservation))
|
{
|
Log.Warn($"[ResourceManager] 尝试释放不存在的资源预约: {resourceKey}");
|
return;
|
}
|
|
// 验证 Owner 匹配
|
if (reservation.Owner.TokenId != owner.TokenId)
|
{
|
throw new InvalidOperationException(
|
$"资源 {resourceKey} 释放失败：所有者不匹配，期望 TokenId: {reservation.Owner.TokenId}");
|
}
|
|
// 标记预约为 Released
|
reservation.State = ReservationState.Released;
|
_reservations.TryRemove(resourceKey, out _);
|
|
Log.Info($"[ResourceManager] 资源 {resourceKey} 预约已释放，所有者: {owner.OwnerId}");
|
}
|
finally
|
{
|
semaphore.Release();
|
}
|
|
await Task.CompletedTask; // 保持异步签名
|
}
|
catch (InvalidOperationException)
|
{
|
throw;
|
}
|
catch (OperationCanceledException)
|
{
|
Log.Warn($"[ResourceManager] 资源 {resourceKey} 释放操作被取消");
|
throw;
|
}
|
catch (Exception ex)
|
{
|
Log.Error($"[ResourceManager] 释放资源 {resourceKey} 失败: {ex.Message}", ex);
|
throw;
|
}
|
}
|
|
/// <inheritdoc/>
|
public bool IsReserved(ResourceKey resource, out ResourceReservation? reservation)
|
{
|
if (resource == null) throw new ArgumentNullException(nameof(resource));
|
|
var resourceKey = resource.ToString();
|
var result = _reservations.TryGetValue(resourceKey, out var res);
|
|
if (result && res != null && res.State == ReservationState.Active)
|
{
|
reservation = res;
|
return true;
|
}
|
|
reservation = null;
|
return false;
|
}
|
|
/// <inheritdoc/>
|
public bool ValidateOwner(ResourceOwnerToken owner, ResourceKey resource)
|
{
|
if (owner == null) throw new ArgumentNullException(nameof(owner));
|
if (resource == null) throw new ArgumentNullException(nameof(resource));
|
|
var resourceKey = resource.ToString();
|
|
if (!_reservations.TryGetValue(resourceKey, out var reservation))
|
return false;
|
|
if (reservation.State != ReservationState.Active)
|
return false;
|
|
return reservation.Owner.TokenId == owner.TokenId &&
|
reservation.Owner.Generation == owner.Generation;
|
}
|
}
|
}
|
