0

我对电子邮件发送窗口服务有疑问。该服务在每三分钟延迟后启动，并获取要从数据库发送的消息，然后开始发送。代码如下所示：

        MessageFilesHandler MFHObj = new MessageFilesHandler();
        List<Broadcostmsg> imidiateMsgs = Manager.GetImidiateBroadCastMsgs(conString);
        if (imidiateMsgs.Count > 0)
        {

           // WriteToFileImi(strLog);

            Thread imMsgThread = new Thread(new             ParameterizedThreadStart(MFHObj.SendImidiatBroadcast));              
            imMsgThread.IsBackground = true;
            imMsgThread.Start(imidiateMsgs);
        }

这会将消息发送到大型列表，并且需要很长时间才能完成发送到更大的列表。现在问题发生在消息仍在发送并且服务获取要发送的新消息时，先前的发送被拖拽并且新消息发送开始，虽然我正在使用线程，但每次服务获取消息发送它都会启动一个新线程。你能帮我在代码中出错的地方吗？

4

2 回答 2

0

看起来要求是建立一个消费者生产者队列。其中生产者将继续将消息添加到列表中，而消费者将从该列表中选择项目并对其进行一些工作 唯一让我担心的是，您每次都创建一个新的线程来发送电子邮件，而不是从线程池中选择线程。如果您继续创建越来越多的线程，您的应用程序的性能将由于上下文切换产生的开销而降低。

如果您使用的是 .Net framwe work 4.0，那么灵魂将变得非常容易。您可以使用System.Collections.Concurrent.ConcurrentQueue对您的项目进行排队和出队。它的线程安全，因此不需要锁定对象。使用任务来处理您的消息。

BlockingCollection 在其构造函数中采用 IProducerConsumerCollection，或者如果您调用其空构造函数，它将默认使用 ConcurrentQueue。

因此，将您的消息排入队列。

//define a blocking collectiom
var blockingCollection = new BlockingCollection<string>();

//Producer
Task.Factory.StartNew(() =>
{
    while (true)
    {
        blockingCollection.Add("value" + count);
        count++;                    
    }
});

//consumer
//GetConsumingEnumerable would wait until it find some item for work
// its similar to while(true) loop that we put inside consumer queue
Task.Factory.StartNew(() =>
{
    foreach (string value in blockingCollection.GetConsumingEnumerable())
    {
        Console.WriteLine("Worker 1: " + value);
    }                
});

更新

由于您使用的是 FrameWork 3.5。我建议你看看 Joseph Albahari 的Consumer/Producer Queue实现。它是你会发现的最好的之一。

直接从上面的链接获取代码

public class PCQueue
{
  readonly object _locker = new object();
  Thread[] _workers;
  Queue<Action> _itemQ = new Queue<Action>();

  public PCQueue (int workerCount)
  {
    _workers = new Thread [workerCount];

    // Create and start a separate thread for each worker
    for (int i = 0; i < workerCount; i++)
      (_workers [i] = new Thread (Consume)).Start();
  }

  public void Shutdown (bool waitForWorkers)
  {
    // Enqueue one null item per worker to make each exit.
    foreach (Thread worker in _workers)
      EnqueueItem (null);

    // Wait for workers to finish
    if (waitForWorkers)
      foreach (Thread worker in _workers)
        worker.Join();
  }

  public void EnqueueItem (Action item)
  {
    lock (_locker)
    {
      _itemQ.Enqueue (item);           // We must pulse because we're
      Monitor.Pulse (_locker);         // changing a blocking condition.
    }
  }

  void Consume()
  {
    while (true)                        // Keep consuming until
    {                                   // told otherwise.
      Action item;
      lock (_locker)
      {
        while (_itemQ.Count == 0) Monitor.Wait (_locker);
        item = _itemQ.Dequeue();
      }
      if (item == null) return;         // This signals our exit.
      item();                           // Execute item.
    }
  }
}

这种方法的优点是您可以控制为优化性能而需要创建的线程数。使用线程池方法，虽然它是安全的，但您无法控制可以同时创建的线程数。

于 2012-07-26T05:39:17.603 回答
0

我认为您在等待新消息的循环中使用您的代码，您是否管理了这些等待？让我们来看看：

while(imidiateMsgs.Count == 0)
{
    //Wait for new Message
}

//Now you have a new message Here

//Make a new thread to process message

该等待有不同的方法，我建议使用 BlockingQueues：

公共区域：

BlockingCollection<Broadcostmsg> imidiateMsgs = new BlockingCollection<Broadcostmsg>();

在您的消费者（生成消息的线程）中：

SendImidiatBroadcast = imidiateMsgs.Take();//this will wait for new message
//Now you have a new message Here

//Make a new thread to process message

在生产者（回答消息的线程）中：

imidiateMsgs.Add(SendImidiatBroadcast);

而且每次回复消息都必须使用线程池创建新线程，不要每次都初始化新线程。

于 2012-07-26T05:43:41.670 回答