c# - 多线程多个生产者和使用者线程不会同步BlockingCollection争用条件

Question

我有多个生产者和多个消费者。我的共享资源是BlockingCollection。但是，只有在有一个使用者的情况下，我的代码才有效。我知道这是一个竞争条件，因为每次运行代码时输出都是不同的。

我以为BlockingCollection会处理所有同步等问题，但事实并非如此。

那时我该如何在所有生产者和消费者之间同步我的共享资源？

这是我的代码:

/// <summary>
/// PURE PRODUCER TYPE
/// </summary>
class Caller
{

    private BlockingCollection<Call> incommingCalls;

    public Caller(BlockingCollection<Call> calls)
    {
        incommingCalls = calls;
        //start the producer thread
        Thread thread = new Thread(new ThreadStart(placeCall));
        thread.Start();
    }

    public void placeCall()
    {
            incommingCalls.Add(myCall);
    }

}



/// <summary>
/// CONSUMER
/// </summary>
class Fresher : Employee
{

    private BlockingCollection<Call> calls;

    public Fresher(BlockingCollection<Call> incalls)
    {
        calls = incalls;
        Thread thread = new Thread(new ThreadStart(HandleCalls));
        thread.Start();

    }

    /// <summary>
    /// 
    /// </summary>
    public void HandleCalls()
    {
        while (!incommingCalls.IsCompleted)
        {
            Call item;
            if (incommingCalls.TryTake(out item, 100000))
            {
                //do something with the call

            } //else do nothing - just wait

        }


}






/// <summary>
/// 
/// </summary>
class CallCenter
{

    private BlockingCollection<Call> fresherCalls;


    private List<Caller> myCallers;
    private List<Employee> myFreshers;


    public CallCenter() 
    {
        //initial incomming calls to the fresher queue
        fresherCalls = new BlockingCollection<Call>();

        myFreshers = new List<Employee>();
        myCallers = new List<Caller>();

        generate_freshers();

        //generate to start the producer
        generate_callers();

    }

    /// <summary>
    /// 
    /// </summary>
    private void generate_freshers() 
    {
        for (int i = 0; i < 1; i++ )
        {
            myFreshers.Add(new Fresher(fresherCalls, tlCalls, locker2));
        }
    }

    /// <summary>
    /// 
    /// </summary>
    private void generate_callers() 
    {
        for (int i = 0; i < 20; i++ )
        {
            myCallers.Add(new Caller(fresherCalls, locker));
        }

    }
}

Answer 1

I know it is a race condition since the output is different each time I run the code.

这在多线程中很常见，而不一定是由于竞争条件(至少不是一个坏条件)。多线程方案中的订单处理往往不是确定性的，这很可能会改变输出。

话虽这么说，使用 BlockingCollection<T>，通常将您的使用者写为:

public void HandleCalls()
{
    foreach(var item in incommingCalls.GetConsumingEnumerable())
    {
        //do something with the call
    }
}

这将为 BlockingCollection<T>上的任何数量的使用者处理所有同步并为您检查。

编辑:如果需要控制调度并实现某种形式的 Round-Robin Scheduling，则可能需要查看TPL示例中的 Parallel Extension Extras。它们提供了一个 RoundRobinTaskScheduler，可用于安排以可预测方式工作的 Task<T>实例。