c# - TPL DataFlow 处理异常的正确方法
问题描述
我在使用 TPL DataFlow 管理队列(数据库)并将工作重定向到网格计算服务的 Windows 服务中遇到问题。并且在某一时刻 BufferBlock 停止释放任务,我不知道为什么。我认为这是因为在某些任务的执行过程中发生了一些异常,但是它们被抑制了,很难理解 BufferBlock 何时停止接受新任务。
我试图在下面的工作示例中简化它。它没有任何异常处理,我想知道如何正确处理 TPL 中的异常。我在 TPL Dataflow 找到了类似的东西,仅在所有源数据块完成时才保证完成。在此示例中,我有 100 个请求,并以 10 个请求批量处理数据。模拟如果 ID % 9 == 0 时发生的一些异常如果我没有捕获此异常,它会工作一点,然后停止接受新请求。如果我处理并返回 Result.Failure 我相信它可以正常工作,但我不确定这是否是在生产环境中使用它的正确方法。
我是 TPL 的新手,如果我没有更清楚地解释我的问题,请忘记我。 GitHub 项目
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Net;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
using System.Timers;
using CSharpFunctionalExtensions;
namespace TestTPL
{
public class ServicePipeline
{
public const int batches = 100;
private int currentBatch = 0;
public ServicePipeline(int maxRequestsInParallel)
{
MaxRequestsInParallel = maxRequestsInParallel;
}
public int MaxRequestsInParallel { get; }
public BufferBlock<MyData> QueueBlock { get; private set; }
public List<TransformBlock<MyData, Result>> ExecutionBlocks
{ get; private set; }
public ActionBlock<Result> ResultBlock { get; private set; }
private void Init()
{
QueueBlock = new BufferBlock<MyData>(new DataflowBlockOptions()
{ BoundedCapacity = MaxRequestsInParallel });
ExecutionBlocks = new List<TransformBlock<MyData, Result>>();
ResultBlock = new ActionBlock<Result>(_ => _.OnFailure(
() => Console.WriteLine($"Error: {_.Error}")));
for (int blockIndex = 0; blockIndex < MaxRequestsInParallel;
blockIndex++)
{
var executionBlock = new TransformBlock<MyData, Result>((d) =>
{
return ExecuteAsync(d);
}, new ExecutionDataflowBlockOptions() { BoundedCapacity = 1 });
executionBlock.LinkTo(ResultBlock, new DataflowLinkOptions()
{ PropagateCompletion = true });
QueueBlock.LinkTo(executionBlock, new DataflowLinkOptions()
{ PropagateCompletion = true });
ExecutionBlocks.Add(executionBlock);
}
}
public static Result ExecuteAsync(MyData myData)
{
//try
//{
WebClient web = new WebClient();
TaskCompletionSource<Result> res = new TaskCompletionSource<Result>();
Task task = Task<Result>.Run(() => web.DownloadStringAsync(
new Uri("http://localhost:49182/Slow.ashx")));
task.Wait();
Console.WriteLine($"Data = {myData}");
if (myData != null && myData.Id % 9 == 0)
throw new Exception("Test");
return Result.Ok();
//}
//catch (Exception ex)
//{
// return Result.Failure($"Exception: {ex.Message}");
//}
}
public async void Start()
{
Init();
while (currentBatch < batches)
{
Thread.Sleep(1000);
await SubmitNextRequests();
}
Console.WriteLine($"Completed: {batches}");
}
private async Task<int> SubmitNextRequests()
{
var emptySlots = MaxRequestsInParallel - QueueBlock.Count;
Console.WriteLine($"Empty slots: {emptySlots}" +
$", left = {batches - currentBatch}");
if (emptySlots > 0)
{
var dataRequests = await GetNextRequests(emptySlots);
foreach (var data in dataRequests)
{
await QueueBlock.SendAsync(data);
}
}
return emptySlots;
}
private async Task<List<MyData>> GetNextRequests(int request)
{
MyData[] myDatas = new MyData[request];
Task<List<MyData>> task = Task<List<MyData>>.Run(() =>
{
for (int i = 0; i < request; i++)
{
myDatas[i++] = new MyData(currentBatch);
currentBatch++;
}
return new List<MyData>(myDatas);
});
return await task;
}
}
public class MyData
{
public int Id { get; set; }
public MyData(int id) => Id = id;
public override string ToString() { return Id.ToString(); }
}
}
编辑:2019年 10 月 30 日当异常被处理并显式调用时,它按预期工作Result.Failure($"Exception: {ex.Message}");
public static Result ExecuteAsync(MyData myData)
{
try
{
WebClient web = new WebClient();
TaskCompletionSource<Result> res = new TaskCompletionSource<Result>();
Task task = Task<Result>.Run(() => Thread.Sleep(2000));
task.Wait();
Console.WriteLine($"Data = {myData}");
if (myData != null && myData.Id % 9 == 0)
throw new Exception("Test");
return Result.Ok();
}
catch (Exception ex)
{
return Result.Failure($"Exception: {ex.Message}");
}
}
解决方案
链接两个块时,可以选择向前传播完成,但不能向后传播。这在使用该选项时会成为问题BoundedCapacity
,并且会发生错误,因为它会阻塞管道的馈线并导致死锁。不过,手动传播完成非常容易。这是您可以使用的方法。
async void OnErrorComplete(IDataflowBlock block1, IDataflowBlock block2)
{
await Task.WhenAny(block1.Completion); // Safe awaiting
if (block1.Completion.IsFaulted) block2.Complete();
}
它异步等待block1
完成,如果失败,它会立即完成block2
. 完成上游块通常就足够了,但如果需要,您也可以传播特定的异常:
async void OnErrorPropagate(IDataflowBlock block1, IDataflowBlock block2)
{
await Task.WhenAny(block1.Completion); // Safe awaiting
if (block1.Completion.IsFaulted)
block2.Fault(block1.Completion.Exception.InnerException);
}
推荐阅读
- java - java中sql字符串中的MYSQL语法错误
- c++ - C ++:访问二维数组的值作为函数中的参数/指针传递
- c# - 如何构建一个处理带有等号“=”号的查询参数的 ASP.Net 核心 Web api Enpoint
- python - 如何构建 Django 项目并将视图实现到 index.html
- c# - XSLT 转换路径中的非法字符
- graphql - 允许订阅返回实体的初始数据
- c++ - 用于模板专业化的虚拟功能或 SFINAE……还是更好的方法?
- azure-devops - 在流程模板中使用 Reviewed By 字段的安全性
- excel - 如何根据其旁边的单元格中的值更改来修改单元格值
- java - 如何在java常量中传递jenkins参数值