.NET中的多线程处理 [英] Multi-Thread Processing in .NET

问题描述

我已经有了一些想法，但是如果可能的话，我想听听大家不同的意见和选择。

我有一个Windows控制台应用程序，它使用Exchange Web服务连接到Exchange并下载电子邮件。目标是获取每个单独的消息对象，提取元数据，解析附件等。应用程序每60秒检查一次收件箱。我没有问题连接到收件箱和获取消息对象。这很好。

这里是我接受你的输入的地方：当我收到一个消息对象时，我立即想处理消息，做所有繁忙的工作解释。我正在考虑几种不同的方法：

• 将电子邮件对象在表中排队并逐个处理它们。


• 将电子邮件对象传递到本地Windows服务以执行繁忙的工作。

我不认为db排队是一个好的方法，因为有时，需要处理多个电子邮件对象。如果在处理具有5个附件的高优先级电子邮件之前处理具有30个附件的低优先级电子邮件，这是不公平的。换句话说，在堆栈中较低的电子邮件不应该需要在行中等待被处理。这就像在商店排队等候，在你面前有一个单一的注册记录，以便扫描100件物品。这只是不公平。我的电子邮件对象的相同概念。

我有点不确定Windows服务方法。但是，我非常有信心，我可以安装服务侦听，等待处理新电子邮件的指令的需求。如果我有5个单独的电子邮件对象，我可以对Windows服务和进程5个单独的调用没有冲突？

我可以接受建议或替代方法。但是，解决方案必须使用.NET技术栈来呈现。

解决方案

一个选项是在控制台应用程序中进行处理。你看起来像一个标准的生产者 - 消费者问题与一个生产者（获取电子邮件的线程）和多个消费者。这可以通过阻止收集轻松处理。

我假设你的邮件类型（你从邮件服务器得到的）叫做 MailMessage 。 / p>

所以你在类范围创建一个 BlockingCollection< MailMessage> 。我还假设您有一个计时器，每60秒钟，以收集消息和入列的队列：

  private BlockingCollection< MailMessage> ; MailMessageQueue = 
 new BlockingCollection< MailMessage>（）; 
 
 //定时器被创建为一次性，并在每次打勾时重新初始化。 
 //如果需要
 //运行超过60秒，这将阻止定时器proc被重新输入。 
 System.Threading.Timer ProducerTimer = new System.Threading.Timer（
 TimerProc，null，TimeSpan.FromSeconds（60），TimeSpan.FromMilliseconds（-1））; 
 
 
 void TimerProc（object state）
 {
 var newMessages = GetMessagesFromServer（）; 
 foreach（var msg in newMessages）
 {
 MailMessageQueue.Add（msg）; 
} 
 ProducerTimer.Change（TimeSpan.FromSeconds（60），TimeSpan.FromMilliseconds（-1））; 
} 
  

您的用户线程只是读取队列：

void MessageProcessor（）
{
foreach（MailMessageQueue.GetConsumingEnumerable（）中的var msg）
{
ProcessMessage （）;
}
}


计时器将使生产者每分钟运行一次。要启动消费者（假设你想要两个消费者）：

  var t1 = Task.Factory.StartNew（MessageProcessor，TaskCreationOptions。 LongRunning）; 
 var t2 = Task.Factory.StartNew（MessageProcessor，TaskCreationOptions.LongRunning）; 
  

因此，您将有两个线程处理邮件。

拥有更多的处理线程与拥有可用的CPU内核相比没有任何意义。生产者线程可能不需要大量的CPU资源，所以你不必为其专用一个线程。

我已经跳过了上面的一些细节，特别是取消了线程。当你想停止程序，但让消费者完成处理消息，只是杀死生产者计时器，并设置队列完成添加：

  MailMessageQueue.CompleteAdding（）; 
  

消费者将清空队列并退出。你当然希望等待任务完成（参见 Task.Wait ）。

如果你想要在不清空队列的情况下杀死消费者的能力，您需要查看 BlockingCollection 的默认后备商店

是 ConcurrentQueue ，这是一个严格的FIFO。如果您希望优先处理事情，您需要提出一个并发优先级队列，以实现 IProducerConsumerCollection 界面。 .NET没有这样的东西（或者甚至优先级队列类），但是一个简单的二进制堆使用锁来防止并发访问就足够了;你不是在说这个东西很难。

当然，你需要一些方法来优先处理消息。可能按附件数量排序，以便更快地处理没有附件的邮件。另一个选项是具有两个单独的队列：一个用于具有0或1个附件的消息，以及用于具有大量附件的队列的单独队列。您可以让您的一个消费者专用于0或1队列，以便易于处理的邮件始终有良好的先被处理的机会，其他消费者从0或1队列中取出，除非它是空的，然后从另一个队列。

如果你选择将消息处理移动到一个单独的程序，你将需要一些方法来使消息处理更加复杂，将数据从生产者持久存储到消费者。有很多可能的方法，但我只是看不到它的优势。

I already have a few ideas, but I'd like to hear some differing opinions and alternatives from everyone if possible.

I have a Windows console app that uses Exchange web services to connect to Exchange and download e-mail messages. The goal is to take each individual message object, extract metadata, parse attachments, etc. The app is checking the inbox every 60 seconds. I have no problems connecting to the inbox and getting the message objects. This is all good.

Here's where I am accepting input from you: When I get a message object, I immediately want to process the message and do all of the busy work explained above. I was considering a few different approaches to this:

• Queuing the e-mail objects up in a table and processing them one-by-one.
• Passing the e-mail object off to a local Windows service to do the busy work.

I don't think db queuing would be a good approach because, at times, multiple e-mail objects need to be processed. It's not fair if a low-priority e-mail with 30 attachments is processed before a high-priority e-mail with 5 attachments is processed. In other words, e-mails lower in the stack shouldn't need to wait in line to be processed. It's like waiting in line at the store with a single register for the bonehead in front of you to scan 100 items. It's just not fair. Same concept for my e-mail objects.

I'm somewhat unsure about the Windows service approach. However, I'm pretty confident that I could have an installed service listening, waiting on demand for an instruction to process a new e-mail. If I have 5 separate e-mail objects, can I make 5 separate calls to the Windows service and process without collisions?

I'm open to suggestions or alternative approaches. However, the solution must be presented using .NET technology stack.

解决方案

One option is to do the processing in the console application. What you have looks like a standard producer-consumer problem with one producer (the thread that gets the emails) and multiple consumers. This is easily handled with BlockingCollection.

I'll assume that your message type (what you get from the mail server) is called MailMessage.

So you create a BlockingCollection<MailMessage> at class scope. I'll also assume that you have a timer that ticks every 60 seconds to gather messages and enqueue them:

private BlockingCollection<MailMessage> MailMessageQueue =
    new BlockingCollection<MailMessage>();

// Timer is created as a one-shot and re-initialized at each tick.
// This prevents the timer proc from being re-entered if it takes
// longer than 60 seconds to run.
System.Threading.Timer ProducerTimer = new System.Threading.Timer(
    TimerProc, null, TimeSpan.FromSeconds(60), TimeSpan.FromMilliseconds(-1));


void TimerProc(object state)
{
    var newMessages = GetMessagesFromServer();
    foreach (var msg in newMessages)
    {
        MailMessageQueue.Add(msg);
    }
    ProducerTimer.Change(TimeSpan.FromSeconds(60), TimeSpan.FromMilliseconds(-1));
}

Your consumer threads just read the queue:

void MessageProcessor()
{
    foreach (var msg in MailMessageQueue.GetConsumingEnumerable())
    {
        ProcessMessage();
    }
}

The timer will cause the producer to run once per minute. To start the consumers (say you want two of them):

var t1 = Task.Factory.StartNew(MessageProcessor, TaskCreationOptions.LongRunning);
var t2 = Task.Factory.StartNew(MessageProcessor, TaskCreationOptions.LongRunning);

So you'll have two threads processing messages.

It makes no sense to have more processing threads than you have available CPU cores. The producer thread presumably won't require a lot of CPU resources, so you don't have to dedicate a thread to it. It'll just slow down message processing briefly whenever it's doing its thing.

I've skipped over some detail in the description above, particularly cancellation of the threads. When you want to stop the program, but let the consumers finish processing messages, just kill the producer timer and set the queue as complete for adding:

MailMessageQueue.CompleteAdding();

The consumers will empty the queue and exit. You'll of course want to wait for the tasks to complete (see Task.Wait).

If you want the ability to kill the consumers without emptying the queue, you'll need to look into Cancellation.

The default backing store for BlockingCollection is a ConcurrentQueue, which is a strict FIFO. If you want to prioritize things, you'll need to come up with a concurrent priority queue that implements the IProducerConsumerCollection interface. .NET doesn't have such a thing (or even a priority queue class), but a simple binary heap that uses locks to prevent concurrent access would suffice in your situation; you're not talking about hitting this thing very hard.

Of course you'd need some way to prioritize the messages. Probably sort by number of attachments so that messages with no attachments are processed quicker. Another option would be to have two separate queues: one for messages with 0 or 1 attachments, and a separate queue for those with lots of attachments. You could have one of your consumers dedicated to the 0 or 1 queue so that easy messages always have a good chance of being processed first, and the other consumers take from the 0 or 1 queue unless it's empty, and then take from the other queue. It would make your consumers a little more complicated, but not hugely so.

If you choose to move the message processing to a separate program, you'll need some way to persist the data from the producer to the consumer. There are many possible ways to do that, but I just don't see the advantage of it.

这篇关于.NET中的多线程处理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！