使用 Task.Factory 和实体框架的多线程？答案

【问题标题】：Multithreading using Task.Factory and Entity Framework?使用 Task.Factory 和实体框架的多线程？
【发布时间】：2014-03-13 20:21:14
【问题描述】：

我有一个应用程序通知我们 SMS 数据库中的每个订阅者我们系统的更新。它使用实体框架来选择每条记录，然后创建一个新任务来向该人发送消息。从理论上讲，这应该是一个快速的过程。我做错了什么，因为它每两秒只能得到一个完整的响应。

我认为问题与我在Task.Factory.StartNew() 中设置任务的方式有关。它的行为就像是同步运行，但我希望它异步运行。

如果我完全不了解我使用任务的方式，请告诉我。我的灵感来自this post。

这是我的代码：

class Program
{
static List<MessageToSend> Messages = new List<MessageToSend>();
static Entities oDatabase = new Entities();
static SMS.API oAPI = new SMS.API();

const string sAuthToken = "*****";
const string sNotificationMessage = "*****";

static void Main(string[] args)
{
    foreach (var subscriber in oDatabase.SMS_Subscribers.Where(x => x.GlobalOptOut == false))
    {
        MessageToSend oMessage = new MessageToSend();
        oMessage.ID = subscriber.ID;
        oMessage.MobileNumber = subscriber.MobileNumber;

        var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
        if (recentlySentMessage != null)
        {
            oMessage.Completed = true;
            continue;
        }

        Task t = Task.Factory.StartNew(() =>
        {
            try{
                var keywordID = oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;
                var keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
                oMessage.DemographicID = keyword.DemographicID;
                oMessage.Keyword = keyword.Keyword;

                SendNotificationMessage(oMessage);
            }
            catch (Exception oEx){ //Write exception to console}
        });

        Thread.Sleep(15);
    }

    while (Messages.ToList().Any(x => !x.Completed)){ //wait till all are completed}
}

public static void SendNotificationMessage(object message)
{
    MessageToSend oMessage = (MessageToSend)message;
    try
    {
        SMS.APIResponse oResponse = oAPI.SendMessage(sAuthToken, oMessage.DemographicID, oMessage.Keyword, oMessage.MobileNumber, sNotificationMessage);

        if (oResponse.Success){ //Write success to console }
        else{ //Write failure to console }
    }
    catch (Exception oEx){ //Write Exception to console }

    oMessage.Completed = true;
}
}

class MessageToSend
{
public long ID { get; set; }
public long DemographicID {get;set;}
public string MobileNumber { get; set; }
public bool Completed { get; set; }
public string Keyword { get; set; }

public MessageToSend(){ Completed = false; }
}

编辑：foreach 块的内部现在看起来像这样：

        MessageToSend oMessage = new MessageToSend();
        oMessage.ID = subscriber.ID;
        oMessage.MobileNumber = subscriber.MobileNumber;

        int keywordID = 0;
        SMSShortcodeMover.SMS_Keywords keyword;

        var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
        if (recentlySentMessage != null)
        {
            oMessage.Completed = true;
            continue;
        }

        try
        {
            keywordID = (int)oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;
            keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
        } catch (Exception oEx){ //write exception to console, then continue; }

        Task t = Task.Factory.StartNew(() =>
        {
            oMessage.DemographicID = keyword.DemographicID;
            oMessage.Keyword = keyword.Keyword;

            SendNotificationMessage(oMessage);
        });

        Thread.Sleep(15);
    }

编辑 2：我再次更新了我的代码，现在我在发送之前收集了所有数据。它仍然挂在某个地方，但现在它在大约 5 秒内获取了所有 52,000 行数据。代码如下所示：

var query =
(from subscriber in oDatabase.SMS_Subscribers
where subscriber.GlobalOptOut == false
where !(from x in oDatabase.SMS_OutgoingMessages
        where x.Message == sNotificationMessage
        where x.MobileNumber == subscriber.MobileNumber
        where x.Sent > new DateTime(2014, 3, 12)
        select x).Any()
join sk in oDatabase.SMS_SubscribersKeywords
    on subscriber.ID equals sk.SubscriberID
join k in oDatabase.SMS_Keywords on sk.KeywordID equals k.ID into ks
from k2 in ks.Take(1)
select new MessageToSend()
 {
     ID = subscriber.ID,
     MobileNumber = subscriber.MobileNumber,
     DemographicID = k2.DemographicID,
     Keyword = k2.Keyword
 }).ToList();

foreach( var q in query){
    Task t = Task.Factory.StartNew(() => SendNotificationMessage(q));
    Tasks.Add(t);
    Thread.Sleep(80);
}

Task.WaitAll(Tasks.ToArray());

【问题讨论】：

首先，尝试在任务之外使用oDatabase 查询。还有为什么你的代码有Thread.Sleep(15)？
我有一个Thread.Sleep(15) 来限制循环的速度。我们被允许向我们的 SMS 提供商产生多少流量是有限制的。我最初在任务之外有 oDatabase 查询，但我将它们放在其中以防止它们阻塞其余代码。不过，这似乎没有帮助。

标签： c# entity-framework asynchronous task

【解决方案1】：

如果我是你，我会尝试一次执行所有数据库调用，然后再尝试发送你的消息。

尝试这样做：

var query =
    from subscriber in oDatabase.SMS_Subscribers
    where subscriber.GlobalOptOut == false
    where !(from x in oDatabase.SMS_OutgoingMessages
        where x.Message == sNotificationMessage
        where x.MobileNumber == subscriber.MobileNumber
        where x.Sent > new DateTime(2014, 3, 12)
        select x
    ).Any()
    join sk in oDatabase.SMS_SubscribersKeywords
        on subscriber.ID equals sk.SubscriberID
    join k in oDatabase.SMS_Keywords on sk.KeywordID equals k.ID into ks
    from k2 in ks.Take(1)
    select new
    {
            ID = subscriber.ID,
            MobileNumber = subscriber.MobileNumber,
            DemographicID = k2.DemographicID,
            Keyword = k2.Keyword
    };

var tasks =
    from x in query.ToArray()
    let message = new MessageToSend()
    {
        ID = x.ID,
        MobileNumber = x.MobileNumber,
        DemographicID = x.DemographicID,
        Keyword = x.Keyword
    }
    select Task.Factory.StartNew(() => SendNotificationMessage(message));

Task.WaitAll(tasks.ToArray());

我没有你的数据库，所以我无法测试这个，但如果这不完全正确，这样的东西应该可以工作。

【讨论】：

这里有一些错误的条目，这就是为什么我有所有的 try/catch 块。这个查询会避免这些吗？另外，非常感谢。我从来没有想过这样做。
这是一个很好的方向，我还在努力。我在这篇文章中遇到了一个看起来像这样的错误，我试图弄清楚如何在最后调整选择：stackoverflow.com/questions/7259567/…
@ijb109 - 我已修改查询以消除您提到的错误问题。基本上，我将数据库查询与您需要的对象的构建分开。请注意强制执行数据库查询的query.ToArray() 调用。现在应该更接近工作了。
以更有效的方式获取数据后，似乎问题出在任务的某个地方。感觉好像它只使用了两个或三个线程，或者所有线程都挂在 Web 请求上。感谢您帮助整理数据！
@ijb109 - 你的SMS.API 可能不是线程安全的。此外，TPL 确实将自身限制为几个线程。

【解决方案2】：

for each 循环的每次迭代需要 1-2 秒，这并不让我感到惊讶，因为您有 3 个同步执行的独立数据库调用。总体而言，数据库调用非常慢。解决此问题的一种方法是使用一种方法，该方法在 foreach 块中包含除 Task 代码之外的所有内容，然后使用 task 调用它。只需要注意任务方法中的任何内容都不会阻塞。

即

var tasks = new List<Task>();
foreach (var subscriber in oDatabase.SMS_Subscribers.Where(x => x.GlobalOptOut == false))
{
   tasks.Add(Task.Factory.StartNew(() => SendNotificationTask(subscriber));
   Thread.Sleep(15);
}
//Might want to to use Task.WhenAll instead of WaitAll.  Just need to debug it and see what happens.
Task.WaitAll(tasks.ToArray());



public void SendNotificationTask(SomeType subscriber)
{
    MessageToSend oMessage = new MessageToSend();
    oMessage.ID = subscriber.ID;
    oMessage.MobileNumber = subscriber.MobileNumber;

    int keywordID = 0;
    SMSShortcodeMover.SMS_Keywords keyword;

    ////Database call 1
    var recentlySentMessage = oDatabase.SMS_OutgoingMessages.Where(x => x.Message == sNotificationMessage && x.MobileNumber == oMessage.MobileNumber && x.Sent > new DateTime(2014, 3, 12)).FirstOrDefault();
    if (recentlySentMessage != null)
    {
        oMessage.Completed = true;
    }
    else
    {
      try
      {
        ////Database call 2
        keywordID = (int)oDatabase.SMS_SubscribersKeywords.Where(x => x.SubscriberID == oMessage.ID).First().KeywordID;

        ////Database call 3
        keyword = oDatabase.SMS_Keywords.Where(x => x.ID == keywordID).First();
    } catch (Exception oEx){ //write exception to console, then continue; }

    oMessage.DemographicID = keyword.DemographicID;
    oMessage.Keyword = keyword.Keyword;

    SendNotificationMessage(oMessage);

   }

}

【讨论】：