استفاده از Pool در .Net برای Channel RabbitMQ و Database Connection

در صورتی که در برنامه خود نیاز دارید تا از Pool برای دریافت موارد مختلف مانند RabbitMQ Channel به دلیل جلوگیری از سربار ایجاد هردفعه آن استفاده کنید، می‌توانید از کد زیر کمک بگیرید.
این کد بصورت جنریک است و علاوه بر مورد گفته شده می‌توانید از آن در جاهای مختلف مانند کانکشن دیتابیس نیز استفاده کنید.

Interface:

namespace Pool;

public interface IPool<T> : IDisposable where T : class
{
	/// <summary>
	/// Retrieves an item from the pool.
	/// </summary>
	/// <returns>An item from the pool.</returns>
	/// <exception cref="InvalidOperationException">Thrown when the pool fails to create a new resource.</exception>
	T GetFromPool();

	/// <summary>
	/// Retrieves an item from the pool.
	/// </summary>
	/// <returns>An item from the pool.</returns>
	/// <exception cref="InvalidOperationException">Thrown when the pool fails to create a new resource.</exception>
	Task<T> GetFromPoolAsync();

	/// <summary>
	/// Stop Channel pool and Dispose all items
	/// </summary>
	void Stop();

	/// <summary>
	/// Returns an item back to the pool.
	/// </summary>
	/// <param name="item">The item to return to the pool.</param>
	/// <exception cref="ArgumentNullException">Thrown when <paramref name="item"/> is null.</exception>
	void ReturnToPool(T item);

	/// <summary>
	/// Get the number of remaining threads that can enter
	/// </summary>
	/// <returns></returns>
	int GetCurrentCount();

	/// <summary>
	/// Get the current size of the pool.
	/// </summary>
	/// <returns></returns>
	int GetCurrentSize();

	/// <summary>
	/// Get the maximum size of the pool.
	/// </summary>
	/// <returns></returns>
	int GetMaxSize();

	/// <summary>
	/// Get the available size of the pool.
	/// </summary>
	/// <returns></returns>
	int GetAvailableSize();
}

کد:

using System.Collections.Concurrent;

namespace Pool;

/// <summary>
/// Easy Pool
/// </summary>
/// <typeparam name="T">The type of objects to be pooled.</typeparam>
public class Pool<T> : IPool<T> where T : class
{
	private readonly TimeSpan _defaultShrinkInterval = TimeSpan.FromMinutes(30);
	private readonly System.Timers.Timer _shrinkTimer;
	private readonly Action<T> _cleanupAction;
	private readonly SemaphoreSlim _semaphore;
	private readonly ConcurrentBag<T> _items;
	private readonly object _lock = new();
	private readonly Func<T> _factory;
	private readonly int _maxPoolSize;
	private int _currentSize;
	private bool _disposed;

	/// <summary>
	/// Initializes a new instance of the <see cref="Pool{T}"/> class.
	/// </summary>
	/// <param name="factory">A function to create new instances of <typeparamref name="T"/>.</param>
	/// <param name="cleanupAction">Call on Shrink and Dispose to clean <typeparamref name="T"/>. default is call .Dispose if item is Disposable</param>
	/// <param name="shrinkInterval">Time interval to shrink unused pools and disposed them, then reset to initPoolSize, default value is 30 min on null param</param>
	/// <param name="initPoolSize">The initial number of objects to be created and added to the pool. Default is 100.</param>
	/// <param name="maxPoolSize">The maximum number of objects that can be in the pool. Default is <see cref="int.MaxValue"/>.</param>
	/// <exception cref="ArgumentNullException">Thrown when <paramref name="factory"/> is null.</exception>
	/// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="initPoolSize"/> is negative or <paramref name="maxPoolSize"/> is less than or equal to zero.</exception>
	/// <exception cref="ArgumentOutOfRangeException">Thrown when <paramref name="shrinkInterval"/> is less than 30 min (just if not null)</exception>
	/// <exception cref="ArgumentException">Thrown when <paramref name="maxPoolSize"/> is less than <paramref name="initPoolSize"/>.</exception>
	public Pool(Func<T> factory, Action<T>? cleanupAction = null, TimeSpan? shrinkInterval = null, int initPoolSize = 100, int maxPoolSize = int.MaxValue)
	{
#if NET8_0
		ArgumentNullException.ThrowIfNull(factory);
		ArgumentOutOfRangeException.ThrowIfNegativeOrZero(initPoolSize);
#else
		if (factory is null)
		{
			throw new ArgumentNullException(nameof(factory), Resources.Object_Can_Not_Be_Null);
		}

		if (initPoolSize < 1)
		{
			throw new ArgumentOutOfRangeException(nameof(initPoolSize), Resources.Can_Not_Be_Zero);
		}
#endif

		if (maxPoolSize <= 0)
		{
			throw new ArgumentOutOfRangeException(nameof(maxPoolSize), Resources.Max_pool_Size_Min_Value);
		}

		if (maxPoolSize < initPoolSize)
		{
			throw new ArgumentOutOfRangeException(nameof(maxPoolSize), Resources.Max_Pool_Size_More_Than_Init);
		}

		if (shrinkInterval != null && shrinkInterval < TimeSpan.FromMinutes(30))
		{
			throw new ArgumentOutOfRangeException(nameof(shrinkInterval), Resources.Min_Shrink_Interval);
		}

		_items = [];
		_factory = factory;
		_maxPoolSize = maxPoolSize;
		_currentSize = initPoolSize;
		_semaphore = new SemaphoreSlim(maxPoolSize, maxPoolSize);
		_cleanupAction = cleanupAction ?? (item =>
		{
			if (item is IDisposable disposable)
			{
				disposable.Dispose();
			}
		});

		for (var i = 0; i < initPoolSize; i++)
		{
			_items.Add(Create());
		}

#if NET8_0
		_shrinkTimer = new System.Timers.Timer(shrinkInterval ?? _defaultShrinkInterval) { AutoReset = true, Enabled = true };
		_shrinkTimer.Elapsed += async (_, _) => await Task.Run(() => ShrinkPool(initPoolSize)).ConfigureAwait(false);
#else
		var interval = shrinkInterval ?? _defaultShrinkInterval;
		_shrinkTimer = new System.Timers.Timer(interval.TotalMilliseconds) { AutoReset = true, Enabled = true };
		_shrinkTimer.Elapsed += async (_, _) => await Task.Run(() => ShrinkPool(initPoolSize)).ConfigureAwait(false);
#endif
	}

	/// <summary>
	/// Retrieves an item from the pool.
	/// </summary>
	/// <returns>An item from the pool.</returns>
	/// <exception cref="InvalidOperationException">Thrown when the pool fails to create a new resource.</exception>
	public T GetFromPool()
	{
		_semaphore.Wait();

		try
		{
			return _items.TryTake(out var result) ? result : TryCreate();
		}
		catch
		{
			_ = _semaphore.Release();
			throw;
		}
	}

	/// <summary>
	/// Retrieves an item from the pool.
	/// </summary>
	/// <returns>An item from the pool.</returns>
	/// <exception cref="InvalidOperationException">Thrown when the pool fails to create a new resource.</exception>
	public async Task<T> GetFromPoolAsync()
	{
		await _semaphore.WaitAsync().ConfigureAwait(false);

		try
		{
			return _items.TryTake(out var result) ? result : TryCreate();
		}
		catch
		{
			_ = _semaphore.Release();
			throw;
		}
	}

	/// <summary>
	/// Returns an item back to the pool.
	/// </summary>
	/// <param name="item">The item to return to the pool.</param>
	/// <exception cref="ArgumentNullException">Thrown when <paramref name="item"/> is null.</exception>
	public void ReturnToPool(T item)
	{
#if NET8_0
		ArgumentNullException.ThrowIfNull(item);
#else
		if (item is null)
		{
			throw new ArgumentNullException(nameof(item));
		}
#endif

		_items.Add(item);
		_semaphore.Release();
	}

	/// <summary>
	/// Get the number of remaining threads that can enter
	/// </summary>
	/// <returns></returns>
	public int GetCurrentCount() => _semaphore.CurrentCount;

	/// <summary>
	/// Get the current size of the pool.
	/// </summary>
	/// <returns></returns>
	public int GetCurrentSize() => _currentSize;

	/// <summary>
	/// Get the maximum size of the pool.
	/// </summary>
	/// <returns></returns>
	public int GetMaxSize() => _maxPoolSize;

	/// <summary>
	/// Get the available size of the pool.
	/// </summary>
	/// <returns></returns>
	public int GetAvailableSize() => _maxPoolSize - _currentSize;

	/// <summary>
	/// Stop Channel pool and Dispose all items
	/// </summary>
	public void Stop() => Dispose();

	/// <summary>
	/// Disposes the pool and releases all resources.
	/// </summary>
	public void Dispose()
	{
		Dispose(true);
		GC.SuppressFinalize(this);
	}

	/// <summary>
	/// Disposes the pool and releases all resources.
	/// </summary>
	/// <param name="disposing">A boolean value indicating whether the method is called from the Dispose method.</param>
	protected virtual void Dispose(bool disposing)
	{
		lock (_lock)
		{
			if (_disposed)
			{
				return;
			}

			if (disposing)
			{
				_semaphore.Dispose();
				_shrinkTimer.Dispose();

				while (_items.TryTake(out var item))
				{
					_cleanupAction(item);
				}
			}

			_disposed = true;
		}
	}

	private T TryCreate()
	{
		try
		{
			var newSize = Interlocked.Increment(ref _currentSize);

			if (newSize > _maxPoolSize)
			{
				_ = Interlocked.Decrement(ref _currentSize);
				throw new InvalidOperationException(Resources.Poo_Maximum_Capacity);
			}

			return Create();
		}
		catch
		{
			_ = Interlocked.Decrement(ref _currentSize);
			throw;
		}
	}

	private T Create()
	{
		try
		{
			var item = _factory();

			if (item != null)
			{
				return item;
			}

			throw new InvalidOperationException(Resources.Factory_Produced_Null_Item);
		}
		catch (InvalidOperationException)
		{
			throw;
		}
		catch (Exception ex)
		{
			throw new InvalidOperationException(Resources.Erro_Creation, ex);
		}
	}

	private void ShrinkPool(int initPoolSize)
	{
		lock (_lock)
		{
			if (_disposed)
			{
				return;
			}

			var itemsToRemove = Math.Max(0, _currentSize - initPoolSize);

			for (var i = 0; i < itemsToRemove; i++)
			{
				try
				{
					if (_items.TryTake(out var item))
					{
						_cleanupAction(item);

						_ = Interlocked.Decrement(ref _currentSize);
					}
					else
					{
						break;
					}
				}
				catch (Exception e)
				{
					if (Environment.UserInteractive)
					{
						Console.WriteLine(e);
					}
				}
			}
		}
	}
}

نمونه استفاده:

using Pool;

var pool = new Pool<object>(() => new object(), initPoolSize: 5, maxPoolSize: 10);

try
{
    var obj = pool.Get();
    pool.Return(obj);
}
finally
{
    pool.Dispose();
}

نمونه کد برای ربیت:

using Pool;

_connection = factory.CreateConnection(endpoints);

var pool = new Pool<IModel>(() => _connection.CreateModel(), initPoolSize: 5, maxPoolSize: 10);

try
{
    var obj = pool.Get();
    pool.Return(obj);
}
finally
{
    pool.Dispose();
}

مورد بالا برای راحتی بصورت Nuget Package نیز درآمده است که می‌توانید از آن استفاده کنید.

EasyPool github
EasyPool nuget