c#线程间通信

Question

您好我想要线程来协作生产者和消费者.消费者相当缓慢,生产者非常快,并且突然工作.

例如,消费者可以每20秒处理一个消息,并且生产者可以在一秒内产生10个消息,但是在很长一段时间内消息可以产生一次消息,因此消费者可以赶上.

我想要的东西:

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);
}

Answer 2

我会阅读以下文章，它们描述了您的问题。基本上，您没有为您的工作单元获得正确的隔离。

链接 链接