using System.Collections.Concurrent;
using Aging.Infrastructure.Contracts;
using log4net;
namespace Aging.Infrastructure.Core
{
///
/// 资源管理实现。
/// 提供线程安全的资源预约、租约获取和释放功能。
/// 使用 ConcurrentDictionary 存储活跃预约,使用 SemaphoreSlim 实现 per-resource 锁。
///
public sealed class ResourceManager : IResourceManager
{
private static readonly ILog Log = LogManager.GetLogger(typeof(ResourceManager));
///
/// 活跃预约字典(key = ResourceKey.ToString())
///
private readonly ConcurrentDictionary _reservations = new();
///
/// 每资源信号量字典,用于实现细粒度锁
///
private readonly ConcurrentDictionary _resourceLocks = new();
///
/// 全局世代号,每次新预约递增
///
private long _generation;
///
/// 默认租约有效期(秒)
///
private const int DefaultLeaseDurationSeconds = 30;
///
public async Task 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;
}
}
///
public async Task 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;
}
}
///
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;
}
}
///
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;
}
///
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;
}
}
}