如何为LINQ管道计算完成预估剩余时间（ETA）？

绝对可以用System.Reactive（Rx）搞定这个需求！Rx天生擅长处理异步序列的追踪和反馈，刚好能帮我们在逐个处理Process(x)的时候动态估算剩余时间。下面是一个简洁的实现思路和代码示例：

• 把原始的items集合转换成Rx的可观察序列，让Rx帮我们管理异步处理流程
• 给每个处理项打上进度标记（比如索引），同时记录单个项的处理耗时
• 基于已完成项的平均耗时，结合剩余项数，实时计算预估剩余时间
• 通过Rx的订阅机制，在每个项处理完成后立即更新剩余时间

using System;
using System.Collections.Generic;
using System.Linq;
using System.Reactive.Linq;
using System.Diagnostics;
using System.Threading;

// 模拟你的Process方法，实际替换成你的业务逻辑
T Process<T>(T item)
{
    // 模拟数秒的耗时操作
    Thread.Sleep(2000);
    return item;
}

void Main()
{
    // 替换成你的实际数据集合
    var items = Enumerable.Range(1, 10);
    var totalItemCount = items.Count();
    
    // 用于追踪耗时的辅助变量
    var completedItems = 0;
    var totalElapsedTime = TimeSpan.Zero;
    var stopwatch = new Stopwatch();

    // 将LINQ序列转为Rx可观察序列，并添加进度追踪逻辑
    var processedSequence = items
        .ToObservable()
        .Select((item, itemIndex) => 
        {
            stopwatch.Restart();
            var processedResult = Process(item);
            stopwatch.Stop();

            // 线程安全地更新统计数据（如果用多线程处理的话）
            lock (stopwatch)
            {
                completedItems++;
                totalElapsedTime += stopwatch.Elapsed;
            }

            return (Result: processedResult, CompletedCount: itemIndex + 1);
        });

    // 订阅序列，实时输出剩余时间
    processedSequence.Subscribe(
        onNext: progressData => 
        {
            var averageTimePerItem = totalElapsedTime.TotalSeconds / completedItems;
            var remainingItems = totalItemCount - progressData.CompletedCount;
            var estimatedRemaining = TimeSpan.FromSeconds(averageTimePerItem * remainingItems);

            Console.WriteLine($"已完成 {progressData.CompletedCount}/{totalItemCount} | 预估剩余时间：{estimatedRemaining:hh\\:mm\\:ss}");
        },
        onCompleted: () => 
        {
            Console.WriteLine("✅ 所有项处理完成！");
        }
    );

    // 控制台程序需要这行等待处理完成，UI环境可忽略
    Console.ReadLine();
}

如果你的Process(x)耗时波动较大，想要更精准的预估，可以改用滑动平均，只取最近N个项的耗时来计算平均值：

// 替换之前的completedItems和totalElapsedTime
var recentElapsedTimes = new Queue<TimeSpan>(5); // 保存最近5个项的耗时

// 在Select方法内更新队列
recentElapsedTimes.Enqueue(stopwatch.Elapsed);
if (recentElapsedTimes.Count > 5) recentElapsedTimes.Dequeue();

// 在onNext里计算平均耗时
var averageTimePerItem = recentElapsedTimes.Average(t => t.TotalSeconds);

另外，如果希望Process(x)在后台线程执行，避免阻塞主线程，可以给序列加上SubscribeOn：

var processedSequence = items
    .ToObservable()
    .SubscribeOn(System.Reactive.Concurrency.Scheduler.Default) // 用线程池处理
    .Select(/* 原有逻辑 */);

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