基于TPL DataFlow将类中函数拆分为TransformBlocks的实现问题

首先需要解决不同块之间的数据类型传递问题——步骤间输入输出类型会变化（Calculation → SolverHandler → (Calculation, OptimizationResults)），因此用值元组携带上下文信息，保证后续步骤能获取原始Calculation实例。

1. 调整辅助方法（优化上下文传递） #

修改OptimizationBuilder，确保每个步骤能正确传递所需上下文：

public class OptimizationBuilder
{
    // 步骤1：初始化Calculation
    public Calculation InitCalculation(Calculation calc)
    {
        calc.CreateResultsPath();
        calc.SetupLogger();
        return calc;
    }

    // 步骤2：创建SolverHandler并绑定进度事件，保留Calculation上下文
    public (Calculation Calc, SolverHandler Handler) PrepareSolver(Calculation calc)
    {
        var handler = new SolverHandler(calc);
        handler.ProgressReporter += ProgressChanged;
        return (calc, handler);
    }

    // 步骤3：运行求解器并返回结果，保留Calculation上下文
    public (Calculation Calc, OptimizationResults? Results) RunAndGetResults((Calculation Calc, SolverHandler Handler) input)
    {
        // 假设RunSolver返回bool标识是否成功，根据实际逻辑调整
        var success = input.Handler.RunSolver();
        if (!success)
        {
            throw new InvalidOperationException("求解器运行失败");
            // 也可返回(input.Calc, null)，后续块判断是否跳过处理
        }
        var results = input.Handler.GetOptimizationResults();
        return (input.Calc, results);
    }

    // 步骤4：写入结果文件
    public void WriteResultsFile((Calculation Calc, OptimizationResults? Results) input)
    {
        if (input.Results != null)
        {
            input.Calc.WriteResultsFile(input.Results);
        }
    }

    // 步骤5：写入Excel
    public void WriteResultsToExcel((Calculation Calc, OptimizationResults? Results) input)
    {
        if (input.Results != null)
        {
            input.Calc.WriteResultsToExcel(input.Results);
        }
    }

    // 进度处理逻辑，根据实际需求实现
    private void ProgressChanged(object? sender, ProgressEventArgs e)
    {
        // 处理进度更新
    }
}

2. 定义并连接DataFlow块 #

创建对应步骤的TransformBlock和ActionBlock，通过LinkTo串联，同时配置并行度与取消令牌：

var builder = new OptimizationBuilder();
var cts = _cancelationCancellationTokenSource;
var baseOptions = new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 4,
    CancellationToken = cts.Token
};

// 块1：初始化Calculation
var initBlock = new TransformBlock<Calculation, Calculation>(
    calc => builder.InitCalculation(calc),
    baseOptions);

// 块2：创建SolverHandler并携带上下文
var prepareSolverBlock = new TransformBlock<Calculation, (Calculation, SolverHandler)>(
    calc => builder.PrepareSolver(calc),
    baseOptions);

// 块3：运行求解器并获取结果
var runSolverBlock = new TransformBlock<(Calculation, SolverHandler), (Calculation, OptimizationResults?)>(
    input => builder.RunAndGetResults(input),
    baseOptions);

// 块4：写入结果文件
var writeFileBlock = new ActionBlock<(Calculation, OptimizationResults?)>(
    input => builder.WriteResultsFile(input),
    baseOptions);

// 块5：写入Excel
var writeExcelBlock = new ActionBlock<(Calculation, OptimizationResults?)>(
    input => builder.WriteResultsToExcel(input),
    baseOptions);

// 连接各个块，设置完成状态自动传播
initBlock.LinkTo(prepareSolverBlock, new DataflowLinkOptions { PropagateCompletion = true });
prepareSolverBlock.LinkTo(runSolverBlock, new DataflowLinkOptions { PropagateCompletion = true });
runSolverBlock.LinkTo(writeFileBlock, new DataflowLinkOptions { PropagateCompletion = true });
runSolverBlock.LinkTo(writeExcelBlock, new DataflowLinkOptions { PropagateCompletion = true });

// 提交所有Calculation任务
foreach (var calc in Calculations)
{
    await initBlock.SendAsync(calc, cts.Token);
}

// 标记输入完成
initBlock.Complete();

// 等待所有处理步骤完成
await Task.WhenAll(writeFileBlock.Completion, writeExcelBlock.Completion);

关键细节说明 #

• 上下文传递：用值元组(Calculation, T)保留原始实例，避免后续写结果时丢失上下文。
• 并行控制：全局设置MaxDegreeOfParallelism，若需单独控制某步骤并行度（如求解器），可单独为该块配置选项。
• 错误处理：求解失败时可选择抛出异常终止数据流，或返回null结果让后续块跳过处理，根据业务需求调整。
• 完成传播：PropagateCompletion = true确保上游块完成后，下游块自动标记完成，无需手动处理。
• 取消支持：所有块共享同一个CancellationToken，保证取消信号能传递到所有步骤。

内容的提问来源于stack exchange，提问作者Diddlik