C#线程间通信

Question

我想要两个线程协作；生产者和消费者。消费者相当慢，而生产者非常快并且突发性地工作。

例如，消费者可以每 20 秒处理一条消息，而生产者可以在一秒钟内产生 10 条消息，但在很长一段时间内大约会这样做一次，以便消费者能够 catch 。

我想要这样的东西:

Stream commonStream;
AutoResetEvent commonLock;

void Producer()
{
    while (true)
    {
        magic.BlockUntilMagicAvalible();
        byte[] buffer = magic.Produce();
        commonStream.Write(buffer);
        commonLock.Set();
    }
}

void Consumer()
{
    while(true)
    { 
        commonLock.WaitOne();
        MagicalObject o = binarySerializer.Deserialize(commonStream);
        DoSomething(o);
    }
}

Answer 1

如果您有 .Net 4.0 或更高版本，您可以使用 BlockingCollection

int maxBufferCap = 500;
BlockingCollection<MagicalObject> Collection 
                           = new BlockingCollection<MagicalObject>(maxBufferCap);
void Producer()
{
    while (magic.HasMoreMagic)
    {
        this.Collection.Add(magic.ProduceMagic());
    }
    this.Collection.CompleteAdding();
}

void Consumer()
{
    foreach (MagicalObject magicalObject in this.Collection.GetConsumingEnumerable())
    {
        DoSomthing(magicalObject);
    }
}

如果缓冲区中没有数据，foreach 行将休眠，当有内容添加到集合中时，它也会自动唤醒。

我设置最大缓冲区的原因是，如果您的生产者比消费者快得多，那么随着越来越多的对象被放入集合中，您最终可能会消耗大量内存。通过在创建阻塞集合时设置最大缓冲区大小，当达到缓冲区大小时，对生产者的 Add 调用将阻塞，直到消费者从集合中删除一个项目。

BlockingCollection 类的另一个好处是它可以拥有任意数量的生产者和消费者，而不需要是 1:1 的比例。如果 DoSomthing 支持它，您可以为计算机的每个核心设置一个 foreach 循环(或者甚至使用 Parallel.ForEach 并使用消耗枚举作为数据源)

void ConsumersInParalell()
{
    //This assumes the method signature of DoSomthing is one of the following:
    //    Action<MagicalObject>
    //    Action<MagicalObject, ParallelLoopState>
    //    Action<MagicalObject, ParallelLoopState, long>
    Paralell.ForEach(this.Collection.GetConsumingEnumerable(), DoSomthing);
}