具有多线程或任务的进程队列

Question

我有一个电话消息应用程序，其中有很多消息要处理。由于电话端口有限，所以消息将被处理先进先出。每条消息都有一个“确认”标志，指示处理的是哪个消息。它当然被初始化为假。

我想将所有消息放入一个队列，然后用多个线程或任务处理它们。

    public class MessageQueue
    {
        public Queue MessageWorkItem { get; set; }
        public Messages Message { get; set; }
        public MessageQueue()
        {
            MessageWorkItem = new Queue();
            Message = new Messages();
        }
        public void GetMessageMetaData()
        {
            try
            {
                // It is just a test, add only one item into the queue
                Message.MessageID = Guid.NewGuid();
                Message.NumberToCall = "1111111111";
                Message.FacilityID = "3333";
                Message.NumberToDial = "2222222222";
                Message.CountryCode = "1";
                Message.Acknowledge = false;
            }
            catch (Exception ex)
            {
            }
        }

        public void AddingItemToQueue()
        {
            GetMessageMetaData();
            if (!Message.Acknowledge)
            {
                lock (MessageWorkItem)
                {
                    MessageWorkItem.Enqueue(Message);
                }
            }
        }
    }

    public class Messages
    {
        public Guid MessageID { get; set; }
        public string NumberToCall { get; set; }
        public string FacilityID { get; set; }
        public string NumberToDial { get; set; }
        public string CountryCode { get; set; }
        public bool Acknowledge { get; set; }
    }

现在我的问题是如何使用多线程从队列中取出项目。 对于队列中的每一项，我都想运行一个脚本。

        public void RunScript(Message item)
        {
            try
            {
                PlayMessage(item); 
                return;
            }
            catch (HangupException hex)
            {
                Log.WriteWithId("Caller Hungup!", hex.Message);
            }
            catch (Exception ex)
            {
                Log.WriteException(ex, "Unexpected exception: {0}");
            }
        }

我的想法是看看

if(MessageWorkItem.Count >= 1) 然后做点什么，但我确实需要代码帮助。

Answer 1

Parallel.ForEach 来自TPL。它是并行的。

示例（将 MessageWorkItem 更改为通用队列）：

    public class MessageQueue
{
    public Queue<Message> MessageWorkItem { get; set; }

    public MessageQueue()
    {
        MessageWorkItem = new Queue<Message>();
    }

    public Message GetMessageMetaData()
    {
        try
        {
            // It is just a test, add only one item into the queue
            return new Message()
            {
                MessageID = Guid.NewGuid(),
                NumberToCall = "1111111111",
                FacilityID = "3333",
                NumberToDial = "2222222222",
                CountryCode = "1",
                Acknowledge = false
            };
        }
        catch (Exception ex)
        {
            return null;
        }
    }

    public void AddingItemToQueue()
    {
        var message = GetMessageMetaData();
        if (!message.Acknowledge)
        {
            lock (MessageWorkItem)
            {
                MessageWorkItem.Enqueue(message);
            }
        }
    }
}

public class Message
{
    public Guid MessageID { get; set; }
    public string NumberToCall { get; set; }
    public string FacilityID { get; set; }
    public string NumberToDial { get; set; }
    public string CountryCode { get; set; }
    public bool Acknowledge { get; set; }
}

class Program
{
    static void Main(string[] args)
    {
        MessageQueue me = new MessageQueue();
        for (int i = 0; i < 10000; i++)
            me.AddingItemToQueue();

        Console.WriteLine(me.MessageWorkItem.Count);

        Parallel.ForEach(me.MessageWorkItem, RunScript);
    }

    static void RunScript(Message item)
    {
        // todo: ...
        Console.WriteLine(item.MessageID);
        Thread.Sleep(300);
    }
}

Answer 2

如果您可以使用 .Net 4.5，我建议您查看 Dataflow from the the Task Parallel Library (TPL)。

该页面提供了很多示例演练，例如 How to: Implement a Producer-Consumer Dataflow Pattern 和 Walkthrough: Using Dataflow in a Windows Forms Application。

查看该文档，看看它是否对您有帮助。需要考虑的内容很多，但我认为这可能是您最好的方法。

或者，您可以考虑使用BlockingCollection 及其GetConsumingEnumerable() 方法来访问队列中的项目。

您所做的是将工作拆分为您希望以某种方式处理的对象，并使用 BlockingCollection 来管理队列。

一些使用ints而不是对象作为工作项的示例代码将有助于证明这一点：

当工作线程完成其当前项时，它将从工作队列中删除一个新项，处理该项，然后将其添加到输出队列中。

一个单独的消费者线程从输出队列中移除已完成的项目并对它们进行处理。

最后，我们必须等待所有工作人员完成 (Task.WaitAll(workers))，然后才能将输出队列标记为已完成 (outputQueue.CompleteAdding())。

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace Demo
{
    class Program
    {
        static void Main(string[] args)
        {
            new Program().run();
        }

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

            Task.Factory.StartNew(consumer);

            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)
            {
                Console.WriteLine("Queueing work item {0}", i);
                inputQueue.Add(i);
                Thread.Sleep(50);
            }

            Console.WriteLine("Stopping adding.");
            inputQueue.CompleteAdding();
            Task.WaitAll(workers);
            outputQueue.CompleteAdding();
            Console.WriteLine("Done.");

            Console.ReadLine();
        }

        void worker(int workerId)
        {
            Console.WriteLine("Worker {0} is starting.", workerId);

            foreach (var workItem in inputQueue.GetConsumingEnumerable())
            {
                Console.WriteLine("Worker {0} is processing item {1}", workerId, workItem);
                Thread.Sleep(100);          // Simulate work.
                outputQueue.Add(workItem);  // Output completed item.
            }

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

        void consumer()
        {
            Console.WriteLine("Consumer is starting.");

            foreach (var workItem in outputQueue.GetConsumingEnumerable())
            {
                Console.WriteLine("Consumer is using item {0}", workItem);
                Thread.Sleep(25);
            }

            Console.WriteLine("Consumer is finished.");
        }

        BlockingCollection<int> inputQueue = new BlockingCollection<int>();
        BlockingCollection<int> outputQueue = new BlockingCollection<int>();
    }
}