并行执行多个任务并将结果保存到数据库 UnitOfWork 为空

Question

我最近发布了另一个关于如何处理数据库中的项目列表并在处理失败时重试 3 次的问题。

问题可以在这里找到：C# process large list of items in a loop and retry if fails

我对收到的该问题的答案进行了一些更改，代码如下：

我有一个继承 ApiController 类的抽象类，我所有的 Web Api 控制器都继承了 ApiBaseController：

在ApiBaseController 中，我定义了使用存储库模式的 UnitOfWork，以便对 SQL 数据库进行 CRUD。 UnitOfWork 和存储库模式运行良好，我已经对其进行了测试，没有任何问题。

public abstract class ApiBaseController : ApiController
{
    protected UnitOfWork Uow { get; set; }          

    protected override void Dispose(bool disposing)
    {
        if (Uow != null && Uow is IDisposable)
        {
            ((IDisposable)Uow).Dispose();
            Uow = null;
        }
        base.Dispose(disposing);
    }
}

这里我有一个JobController，它继承了ApiBaseController，它继承了ApiController，使它成为一个Web Api 控制器。

这个控制器有一个端点api/endpoint/sendrequests，当被调用时将从数据库中获取所有Jobs，并以10个为一组进行处理。

ProcessTaskAsync 方法将处理从数据库中检索到的每个单独的任务，如果失败，它将尝试再处理 2 次，直到该任务被忽略。

除了在ProcessTaskAsync 处理任务后我尝试使用UnitOfWork await Uow.Results.AddAsync(result); 将任务结果保存到数据库之外，一切都很好。这里失败了！问题是 Uow 对象为空，我不明白为什么。我在想是因为任务是异步处理的，控制器执行结束，这意味着控制器已被释放，因此 UnitOfWork。

知道为什么Uow 为空，我该如何解决？

[AllowAnonymous]
[RoutePrefix("api/endpoint")]
public class JobController : ApiBaseController
{

private const int PAGED_LIST_SIZE = 10;

private const int MAX_RETRY_COUNT = 3;

public JobController()
{
    Uow = new UnitOfWork();          
}       

[HttpPost]
[AllowAnonymous]
[Route("sendrequests")]
public async Task<IHttpActionResult> SendRequests()
{
    try
    {              

        var items = await Uow.Jobs.GetAllAsync();

        await ProcessTasks(items);                

        return Ok();

    }
    catch (Exception ex)
    {
        return BadRequest();
    }
}


private async Task ProcessTasks(List<Job> items)
{
    var pagedResults = items.Paged<Job>(PAGED_LIST_SIZE);

    if (pagedResults != null && pagedResults.Count() > 0)
    {
        foreach (var collection in pagedResults)
        {
            List<Task> tasks = new List<Task>();

            foreach (var item in collection.ToList())
            {
                var task = Task.Factory.StartNew(() => ProcessTaskAsync(item));
                tasks.Add(task);
            }

            try
            {
                await Task.WhenAll(tasks.ToArray());                   
            }
            catch (Exception ez)
            {
                //log here the failed ones
            }
        }
    }

}

private async void ProcessTaskAsync(Job item)
{
    if (item.Processed >= MAX_RETRY_COUNT)
    {
        Debug.WriteLine(string.Format("Max count {0} reached for task: {1} ", item.Processed.ToString(), item.Name));
        return;
    }

    item.Processed++;

    try
    {
        Result result = await item.Process();

        if (result != null)
        {
           await Uow.Results.AddAsync(result);
        }

        Debug.WriteLine(string.Format("working item name: {0}", item.Name));
    }  
    catch (Exception ex)
    {
        Debug.WriteLine(string.Format("Exception occured: {0}", ex.Message.ToString()));            
        ProcessTaskAsync(item);
    }
}

这用于将列表返回为按 10 分页的列表。它位于静态类中。

public static IEnumerable<IEnumerable<T>> Paged<T>(this IEnumerable<T> enumerable, int chunkSize = 10)
{
    int itemsReturned = 0;
    var list = enumerable.ToList();
    while (itemsReturned < list.Count)
    {
        int currentChunkSize = Math.Min(chunkSize, list.Count - itemsReturned);
        yield return list.GetRange(itemsReturned, currentChunkSize);
        itemsReturned += currentChunkSize;
    }
}

Answer 1

private async void ProcessTaskAsync(Job item)

实际工作方法必须返回您可以等待的Task。 async void 可能只适用于顶级 UI 事件，绝对不是等待工作。

您使用Task.Factory.StartNew 生成这样的工作并等待结果。但是您正在等待的只是开始工作的外部任务（并且没有内部任务，它是void）。我认为您的 Controller 在工作方法开始使用您的 UoW 实例之前被处置是对的 - 因为您实际上并没有在 Controller 方法中等待该工作。

您应该做的是摆脱Task.Factory.StartNew（用于并行工作，而不是异步），并让ProcessTaskAsync 在实际工作完成时返回Task 信号。然后你可以用WhenAll代替那些Tasks，它应该可以按预期工作。

---

async 相关内容的最佳参考/资源可能是 Stephen Cleary 的博客。我强烈推荐阅读 TPL 文章 http://blog.stephencleary.com/