线程安全可复用资源池实现求助：多Worker资源访问管理

针对你的需求，我们可以通过ConcurrentQueue（线程安全队列）存储可用资源，结合SemaphoreSlim实现异步等待的资源抢占逻辑，确保同一时间资源仅被一个任务占用，同时支持资源复用和异步场景。

核心实现思路 #

• 用ConcurrentQueue<T>管理空闲资源，它本身就是线程安全的，无需额外锁操作。
• 用SemaphoreSlim控制并发访问数量：当有空闲资源时，允许任务获取；无空闲资源时，任务会异步等待直到有资源被释放（或取消）。
• 通过try-finally块确保资源无论回调执行成功与否，都能被正确回收复用。
• 支持CancellationToken，允许任务在等待资源时主动取消。

完整实现代码 #

using System.Collections.Concurrent;

public interface IResource
{
}

public class TestResource : IResource
{
    public void Print()
    {
        Console.WriteLine($"Resource {GetHashCode()} printing on thread {Environment.CurrentManagedThreadId}");
    }
}

public class ResourcePool
{
    // 存储不同类型资源的字典，支持多种资源类型
    private readonly ConcurrentDictionary<Type, object> _resourceQueues = new();
    // 存储不同类型资源的信号量，控制并发
    private readonly ConcurrentDictionary<Type, SemaphoreSlim> _semaphores = new();
    // 资源最大数量限制，可按需调整
    private readonly int _maxResourceCount;

    public ResourcePool(int maxResourceCount = 5)
    {
        _maxResourceCount = maxResourceCount;
    }

    public async Task TryUseResource<T>(Action<T> callback, CancellationToken token = default) where T : IResource, new()
    {
        var resourceType = typeof(T);
        // 获取或创建对应类型的信号量
        var semaphore = _semaphores.GetOrAdd(resourceType, _ => new SemaphoreSlim(_maxResourceCount));
        // 获取或创建对应类型的资源队列
        var queue = (ConcurrentQueue<T>)_resourceQueues.GetOrAdd(resourceType, _ => new ConcurrentQueue<T>());

        try
        {
            // 等待信号量，支持取消
            await semaphore.WaitAsync(token);
            
            // 尝试从队列获取空闲资源，没有则创建新的
            if (!queue.TryDequeue(out var resource))
            {
                resource = new T();
            }

            try
            {
                // 执行回调逻辑
                callback(resource);
            }
            finally
            {
                // 将资源放回队列，供后续任务复用
                queue.Enqueue(resource);
            }
        }
        finally
        {
            // 释放信号量，允许其他任务获取资源
            semaphore.Release();
        }
    }

    // 重载版本：支持异步回调
    public async Task TryUseResourceAsync<T>(Func<T, Task> callback, CancellationToken token = default) where T : IResource, new()
    {
        var resourceType = typeof(T);
        var semaphore = _semaphores.GetOrAdd(resourceType, _ => new SemaphoreSlim(_maxResourceCount));
        var queue = (ConcurrentQueue<T>)_resourceQueues.GetOrAdd(resourceType, _ => new ConcurrentQueue<T>());

        try
        {
            await semaphore.WaitAsync(token);
            
            if (!queue.TryDequeue(out var resource))
            {
                resource = new T();
            }

            try
            {
                await callback(resource);
            }
            finally
            {
                queue.Enqueue(resource);
            }
        }
        finally
        {
            semaphore.Release();
        }
    }
}

public class Worker
{
    public async Task ExecuteWork(ResourcePool pool, CancellationToken token)
    {
        await pool.TryUseResource<TestResource>(resource =>
        {
            resource.Print();
            // 模拟任务执行耗时
            Thread.Sleep(100);
        }, token);
    }

    // 异步任务示例
    public async Task ExecuteAsyncWork(ResourcePool pool, CancellationToken token)
    {
        await pool.TryUseResourceAsync<TestResource>(async resource =>
        {
            resource.Print();
            await Task.Delay(100, token);
        }, token);
    }
}

// 使用示例
class Program
{
    static async Task Main(string[] args)
    {
        var pool = new ResourcePool(maxResourceCount: 2);
        var cancellationToken = CancellationToken.None;
        var workers = new List<Task>();

        // 启动10个Worker任务，验证资源池的并发控制
        for (int i = 0; i < 10; i++)
        {
            workers.Add(new Worker().ExecuteWork(pool, cancellationToken));
        }

        await Task.WhenAll(workers);
    }
}

关键细节说明 #

• 多类型资源支持：通过ConcurrentDictionary存储不同类型的资源队列和信号量，一个资源池可以管理多种IResource实现类。
• 资源复用：任务执行完成后，资源会被重新放回队列，避免频繁创建销毁资源的开销。
• 并发控制：SemaphoreSlim的初始计数设置为资源最大数量，确保同一时间最多有指定数量的任务占用资源。
• 异常安全：嵌套的try-finally块保证即使回调抛出异常，资源也能被正确回收，信号量也能被释放，不会导致死锁。
• 异步支持：提供了TryUseResourceAsync重载方法，支持异步回调逻辑，更适合现代异步编程场景。

内容的提问来源于stack exchange，提问作者Bas Visscher