什么是实现对象池的好方法？

Question

我有一个第三方课程，我们称之为Analyser。这个类非常擅长分析，但实例化成本高（秒）且不支持多线程。

我的应用程序需要处理涉及调用Analyser 的请求。这些请求将同时发生。

我想我需要创建一个通用类，比如

public class Pool<T>
{
    public Pool(Func<T> instantiator, int size)
    {
        ...
    }

    public async Task<TResult> Invoke<TResult>(
            Func<T, TResult> target,
            CancellationToken cancellationToken)
    {
        // await the first available T,
        // lock the T,
        // invoke the target, return the result
        // release the lock
    }
}

这个类一般会封装池化功能。

我的问题是，实现这个类的正确方法是什么。它是否已经以不同的名称存在？我应该使用TPL.DataFlow 吗？我应该用手卷吗？

好被定义为可靠的线程安全，越容易维护越好。

---

如果通用 Pool 是解决问题的错误方法，请提出正确的替代方法。

---

Pool 类会像这样使用。

private readonly Pool<Analyser> pool = new Pool<Analyser>(
        () => new Analyser(a, b, c),
        100);

public async Task<string> ProcessRequest(
        string raw,
        CancellationToken cancellationToken)
{
    return await this.pool.Invoke(
        analyser => analyser.Analyse(raw),
        cancellationToken);
}

Answer 1

池是一个很好的解决方案。毕竟，池正是用于此目的（维护一组过于昂贵而无法每次实例化的对象：数据库连接、线程等）

但是，如果您想构建一个 通用 池，则必须非常小心：您的代码的用户可能会做“意想不到的”事情，并最终自取其辱。 p>

例如锁定：您应该真正检查这不会导致死锁。如果需要，可以即时扩大池，或者在委托人要求更多对象时抛出...... 异常也应该小心处理。

因此，“等待第一个可用的 T”和“锁定 T”步骤应该完全由池处理，并且应该进行所有必要的检查以避免尴尬的情况。您可以考虑为您的“客户端代码”（目标）提供对池的引用以要求额外的锁定功能，如果您愿意（例如嵌套锁定或类似的东西）

更实际的是：您可以从专门为您的 Analyser 类工作的解决方案开始，然后在需要时从那里开始使用通用池？

Answer 2

我认为构建通用池将是一项相当复杂的任务，因此我会从中获得很多乐趣:-)

注意：我的愿景与您的不同之处最重要的是，我不希望池处理与其管理的对象相关的线程问题。该池有一些与线程安全相关的代码，但仅用于管理它自己的状态（实例列表）。 线程启动、停止/和/或取消是池的客户端和构造对象的问题，而不是池本身。

我会开始：

1. 由池维护的对象的一次性包装器，在处理时会将对象返回到池中
2. 在将实例返回给客户端之前构造或重用可用实例并包装它们的池。

一个超级简化的实现：

class PoolItem<T> : IDisposable
{
    public event EventHandler<EventArgs> Disposed;


    public PoolItem(T wrapped)
    {
        WrappedObject = wrapped;
    }


    public T WrappedObject { get; private set; }


    public void Dispose()
    {
        Disposed(this, EventArgs.Empty);
    }
}

现在是游泳池：

class Pool<T> where T : class
{
    private static readonly object m_SyncRoot = new object();

    private readonly Func<T> m_FactoryMethod;
    private List<T> m_PoolItems = new List<T>();


    public Pool(Func<T> factoryMethod)
    {
        m_FactoryMethod = factoryMethod;
    }


    public PoolItem<T> Get()
    {
        T target = null;

        lock (m_SyncRoot)
        {
            if (m_PoolItems.Count > 0)
            {
                target = m_PoolItems[0];
                m_PoolItems.RemoveAt(0);
            }
        }

        if (target == null)
            target = m_FactoryMethod();

        var wrapper = new PoolItem<T>(target);
        wrapper.Disposed += wrapper_Disposed;

        return wrapper;
    }


    void wrapper_Disposed(object sender, EventArgs e)
    {
        var wrapper = sender as PoolItem<T>;

        lock (m_SyncRoot)
        {
            m_PoolItems.Add(wrapper.WrappedObject);
        }
    }
}

用法：

class ExpensiveConstructionObject
{
    public ExpensiveConstructionObject()
    {
        Console.WriteLine("Executing the expensive constructor...");
    }

    public void Do(string stuff)
    {
        Console.WriteLine("Doing: " + stuff);
    }
}

    class Program
{
    static void Main(string[] args)
    {
        var pool = new Pool<ExpensiveConstructionObject>(() => new ExpensiveConstructionObject());

        var t1 = pool.Get();
        t1.WrappedObject.Do("task 1");

        using (var t2 = pool.Get())
            t2.WrappedObject.Do("task 2");

        using (var t3 = pool.Get())
            t3.WrappedObject.Do("task 3");

        t1.Dispose();

        Console.ReadLine();
    }
}

接下来的步骤是：

1. 经典池功能，例如：初始大小、最大大小
2. 动态代理，允许 Pool::Get 返回类型为 T，而不是 PoolItem
3. 维护包装器列表，如果调用者在 Pool 自行处理时不处理它们，则处理它们

Answer 3

IIUC 你想要实现的是一个通用对象池，当你没有资源可以使用时，你会异步等待直到你使用。

最简单的解决方案是使用TPL Dataflow 的BufferBlock 来保存项目并在它为空时等待。在您的 API 中，您会获得一个委托并运行它，但我建议您从池中返回实际项目并让用户决定如何处理它：

public class ObjectPool<TItem>
{
    private readonly BufferBlock<TItem> _bufferBlock;
    private readonly int _maxSize;
    private readonly Func<TItem> _creator;
    private readonly CancellationToken _cancellationToken;
    private readonly object _lock;
    private int _currentSize;

    public ObjectPool(int maxSize, Func<TItem> creator, CancellationToken cancellationToken)
    {
        _lock = new object();
        _maxSize = maxSize;
        _currentSize = 1;
        _creator = creator;
        _cancellationToken = cancellationToken;
        _bufferBlock = new BufferBlock<TItem>(new DataflowBlockOptions{CancellationToken = cancellationToken});
    }

    public void Push(TItem item)
    {
        if (!_bufferBlock.Post(item) || _bufferBlock.Count > _maxSize)
        {
            throw new Exception();
        }
    }

    public Task<TItem> PopAsync()
    {
        TItem item;
        if (_bufferBlock.TryReceive(out item))
        {
            return Task.FromResult(item);
        }
        if (_currentSize < _maxSize)
        {
            lock (_lock)
            {
                if (_currentSize < _maxSize)
                {
                    _currentSize++;
                    _bufferBlock.Post(_creator());
                }
            }
        }

        return _bufferBlock.ReceiveAsync();
    }
}

说明：

• 我使用锁来确保您一次只创建一个新项目，如果需要很长时间，可以很容易地用AsyncLock 替换。
• 我使用Double Check Locking 来优化所有项目都已创建的常见情况。
• PopAsync 返回 Task 但不是异步方法，因此只要有要返回的项目，它就会同步完成。它仅在池为空且已达到限制时等待。

您可以添加一个返回 IDisposable 的方法，这样您就可以将其放入 using scope 而无需担心：

public async Task<Disposable> GetDisposableAsync()
{
    return new Disposable(this, await PopAsync());
}

public class Disposable : IDisposable
{
    private readonly ObjectPool<TItem> _pool;
    public TItem Item { get; set; }

    public Disposable(ObjectPool<TItem> pool, TItem item)
    {
        Item = item;
        _pool = pool;
    }
    public void Dispose()
    {
        _pool.Push(Item);
    }
}