“固定”/“负载平衡”C# 线程池？

Question

我有一个“昂贵”的第 3 方组件。该组件不是线程安全的。所述组件托管在 WCF 服务内部（目前），所以......每次调用进入服务时，我都必须更新组件。

我想做的是拥有一个由 16 个线程组成的池，每个线程启动自己的组件副本，并有一种机制来调用该方法并将其分发到 16 个线程之一并拥有返回值。

所以很简单，比如：

var response = threadPool.CallMethod(param1, param2);

调用阻塞直到它得到响应是可以的，因为我需要响应才能继续。

有什么建议吗？也许我想多了，由 16 个线程服务的 ConcurrentQueue 可以完成这项工作，但现在确定方法返回值将如何返回给调用者？

Answer 1

WCF 已经使用线程池来管理它的资源，所以如果你在上面添加一层线程管理，它只会变得很糟糕。尽可能避免这样做，因为您会在服务调用中发生争用。

在您的情况下，我会做的只是使用 single ThreadLocal 或线程静态，它会被您的昂贵对象初始化一次。此后它将可用于线程池线程。

这是假设您的对象在 MTA 线程上正常；我猜它来自您的帖子，因为听起来目前一切正常，但速度很慢。

担心会创建太多对象，并且随着池变得太大，您会使用太多内存。但是，在执行其他任何操作之前，请先查看实际情况是否如此。这是一个非常简单的策略，实施起来很容易试用。仅在您确实需要时才变得更复杂。

Answer 2

首先，我同意@briantyler：ThreadLocal<T> 或线程静态字段可能是您想要的。您应该以此为起点，如果它不能满足您的需求，请考虑其他选择。

一个复杂但灵活的替代方案是单例对象池。在最简单的形式中，您的池类型将如下所示：

public sealed class ObjectPool<T>
{
    private readonly ConcurrentQueue<T> __objects = new ConcurrentQueue<T>();
    private readonly Func<T> __factory;

    public ObjectPool(Func<T> factory)
    {
        __factory = factory;
    }

    public T Get()
    {
        T obj;
        return __objects.TryDequeue(out obj) ? obj : __factory();
    }

    public void Return(T obj)
    {
        __objects.Enqueue(obj);
    }
}

如果您在原始类或结构（即ObjectPool<MyComponent>）方面考虑类型T，这似乎不是非常有用，因为池没有内置任何线程控件。但您可以替换你的类型 T 为 Lazy<T> 或 Task<T> monad，得到你想要的。

池初始化：

Func<Task<MyComponent>> factory = () => Task.Run(() => new MyComponent());
ObjectPool<Task<MyComponent>> pool = new ObjectPool<Task<MyComponent>>(factory);

// "Pre-warm up" the pool with 16 concurrent tasks.
// This starts the tasks on the thread pool and
// returns immediately without blocking.
for (int i = 0; i < 16; i++) {
    pool.Return(pool.Get());
}

用法：

// Get a pooled task or create a new one. The task may
// have already completed, in which case Result will
// be available immediately. If the task is still
// in flight, accessing its Result will block.
Task<MyComponent> task = pool.Get();

try
{
    MyComponent component = task.Result; // Alternatively you can "await task"

    // Do something with component.
}
finally
{
    pool.Return(task);
}

这种方法比在ThreadLocal 或线程静态字段中维护您的组件更复杂，但如果您需要做一些花哨的事情，例如限制池实例的数量，池抽象会非常有用。

编辑

带有Get 的基本“固定X 实例集”池实现，一旦池耗尽就会阻塞：

public sealed class ObjectPool<T>
{
    private readonly Queue<T> __objects;

    public ObjectPool(IEnumerable<T> items)
    {
        __objects = new Queue<T>(items);
    }

    public T Get()
    {
        lock (__objects)
        {
            while (__objects.Count == 0) {
                Monitor.Wait(__objects);
            }

            return __objects.Dequeue();
        }
    }

    public void Return(T obj)
    {
        lock (__objects)
        {
            __objects.Enqueue(obj);

            Monitor.Pulse(__objects);
        }
    }
}