如何合并多个IProgress<float>对象并计算进度平均值？

需求描述 #

需要将多个IProgress<T>（T为数值类型）对象合并为一个IProgress<T>对象，合并后的对象每次报告的进度为所有源进度的平均值。例如：当progress1报告50%、progress2报告70%时，合并后的mergedProgress应报告60%。

理想实现示例 #

期望的调用方式如下：

IProgress<float> p1 = new Progress(value => Console.WriteLine(value));
IProgress<float> p2 = new Progress(value => Console.WriteLine(value));
IProgress<float> pM = Progress.Merge(p1, p2, value => Console.WriteLine("Full progress: " + value));

当前变通方案 #

目前的实现通过数组存储各进度值并手动计算平均值，但存在多线程访问的潜在风险：

float[] progress = new float[2];

IProgress<float> pM = new Progress(value =>
{  
   double p = 0;
   for (int i = 0; i < progress.Length; i++)
   {
      p += progress[i];
   }
   Console.WriteLine("Full progress: " + p / progress.Length);
});

IProgress<float> p1 = new Progress(value => { progress[0] = value; pM.Report(0); });
IProgress<float> p2 = new Progress(value => { progress[1] = value; pM.Report(0); });

更优实现方案 #

可以通过自定义MergedProgress<T>类封装合并逻辑，同时解决多线程安全问题，且支持任意数量的源进度对象。

实现代码 #

using System;
using System.Collections.Generic;
using System.Threading;

public class MergedProgress<T> : IProgress<T> where T : struct, IConvertible
{
    private readonly List<IProgress<T>> _sourceProgresses;
    private readonly Action<T> _mergedHandler;
    private readonly Dictionary<IProgress<T>, T> _latestProgresses;
    private readonly object _lockObj = new object();

    public MergedProgress(Action<T> mergedHandler, params IProgress<T>[] sourceProgresses)
    {
        _mergedHandler = mergedHandler ?? throw new ArgumentNullException(nameof(mergedHandler));
        _sourceProgresses = new List<IProgress<T>>(sourceProgresses ?? throw new ArgumentNullException(nameof(sourceProgresses)));
        _latestProgresses = new Dictionary<IProgress<T>, T>();

        // 为每个源Progress包装代理，捕获进度更新
        for (int i = 0; i < _sourceProgresses.Count; i++)
        {
            var originalProgress = _sourceProgresses[i];
            _latestProgresses[originalProgress] = default;
            _sourceProgresses[i] = new Progress<T>(value =>
            {
                lock (_lockObj)
                {
                    _latestProgresses[originalProgress] = value;
                }
                ReportMergedProgress();
            });
        }
    }

    public void Report(T value)
    {
        ReportMergedProgress();
    }

    private void ReportMergedProgress()
    {
        lock (_lockObj)
        {
            if (_latestProgresses.Count == 0)
            {
                _mergedHandler(default);
                return;
            }

            // 转换为double计算平均值，再转回T类型
            double sum = 0;
            foreach (var progress in _latestProgresses.Values)
            {
                sum += progress.ToDouble(null);
            }
            double average = sum / _latestProgresses.Count;
            T mergedValue = (T)Convert.ChangeType(average, typeof(T));
            _mergedHandler(mergedValue);
        }
    }

    // 提供静态Merge方法，匹配理想调用方式
    public static IProgress<T> Merge(Action<T> mergedHandler, params IProgress<T>[] sourceProgresses)
    {
        return new MergedProgress<T>(mergedHandler, sourceProgresses);
    }
}

使用示例 #

// 创建原始进度对象
IProgress<float> p1 = new Progress<float>(value => Console.WriteLine($"Progress 1: {value:P0}"));
IProgress<float> p2 = new Progress<float>(value => Console.WriteLine($"Progress 2: {value:P0}"));

// 合并进度
IProgress<float> mergedProgress = MergedProgress<float>.Merge(
    value => Console.WriteLine($"Merged Progress: {value:P0}"),
    p1, p2
);

// 模拟进度更新
p1.Report(0.5f); // 输出 Progress 1: 50%，随后 Merged Progress: 25%
p2.Report(0.7f); // 输出 Progress 2: 70%，随后 Merged Progress: 60%

方案优势 #

• 多线程安全：通过lock保证进度字典的线程安全访问，避免竞争问题。
• 灵活性：支持任意数量的源IProgress<T>对象，无需修改底层存储结构。
• 封装性：合并逻辑完全封装在类内部，调用方式简洁符合预期。
• 类型兼容：基于泛型和IConvertible约束，支持所有数值类型（float、int、double等）。

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