具有多线程的ConcurrentQueue

debugcn 发表于 Dev

纳文·库玛（Naveen kumar）

我是多线程概念的新手。我需要将一定数量的字符串添加到队列中，并使用多个线程处理它们。使用ConcurrentQueue哪个是线程安全的。

这就是我尝试过的。但是不会处理添加到并发队列中的所有项目。仅处理前4个项目。

class Program
{
    ConcurrentQueue<string> iQ = new ConcurrentQueue<string>();
    static void Main(string[] args)
    {
        new Program().run();
    }

    void run()
    {
        int threadCount = 4;
        Task[] workers = new Task[threadCount];

        for (int i = 0; i < threadCount; ++i)
        {
            int workerId = i;
            Task task = new Task(() => worker(workerId));
            workers[i] = task;
            task.Start();
        }

        for (int i = 0; i < 100; i++)
        {
            iQ.Enqueue("Item" + i);
        }

        Task.WaitAll(workers);
        Console.WriteLine("Done.");

        Console.ReadLine();
    }

    void worker(int workerId)
    {
        Console.WriteLine("Worker {0} is starting.", workerId);
        string op;
        if(iQ.TryDequeue(out op))
        {
            Console.WriteLine("Worker {0} is processing item {1}", workerId, op);
        }

        Console.WriteLine("Worker {0} is stopping.", workerId);
    }


}

什么

您的实现存在两个问题。第一个显而易见的worker方法是该方法仅使零或一个项目出队，然后停止：

    if(iQ.TryDequeue(out op))
    {
        Console.WriteLine("Worker {0} is processing item {1}", workerId, op);
    }

它应该是：

    while(iQ.TryDequeue(out op))
    {
        Console.WriteLine("Worker {0} is processing item {1}", workerId, op);
    }

但是，这还不足以使您的程序正常运行。如果您的工作人员出队的速度快于主线程入队的速度，则他们将在主任务仍在排队时停止。您需要通知工人他们可以停车。您可以定义一个布尔变量，true一旦入队，该变量将被设置为：

for (int i = 0; i < 100; i++)
{
    iQ.Enqueue("Item" + i);
}
Volatile.Write(ref doneEnqueueing, true);

工人将检查该值：

void worker(int workerId)
{
    Console.WriteLine("Worker {0} is starting.", workerId);
    do {
        string op;
        while(iQ.TryDequeue(out op))
        {
            Console.WriteLine("Worker {0} is processing item {1}", workerId, op);
        }
        SpinWait.SpinUntil(() => Volatile.Read(ref doneEnqueueing) || (iQ.Count > 0));
    }
    while (!Volatile.Read(ref doneEnqueueing) || (iQ.Count > 0))
    Console.WriteLine("Worker {0} is stopping.", workerId);
}

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。