MVP架构桌面应用多并发任务取消与异常一次性清理方案咨询

问题 #

我正在开发一款控制小型XY平台的桌面应用，采用MVP架构。UI线程向Operation线程发送ViewEvent，由后者消费并执行实际控制逻辑，且Operation线程会并发处理新事件。存在以下场景：

• 收到ViewEvent.Stop事件时终止当前任务
• 新事件（如ViewEvent.MoveRight）覆盖当前任务（如终止ViewEvent.MoveLeft并执行新任务）

我需要实现：取消任务、执行ViewEvent专属清理操作，并利用Operation线程的最新上下文处理LostConnectionException等故障。

我尝试通过异步循环读取BlockingCollection<ViewEvent>，不使用await执行async Task，借助Task.ContinueWith处理取消/故障清理，业务逻辑在Presenter中实现，代码如下：

public async Task HandleEventsBlocking()
{
    foreach (var viewEvent in _viewEventsQueue.GetConsumingEnumerable())
    {
        switch (viewEvent)
        {
            case ViewEvent.Start:
                Presenter.HandleStartAsync(_cts.Token).ContinueWith((task) =>
                {
                    if (task.IsFaulted)
                    {
                        // Clean up and cancel all other tasks.
                        foreach (var innerException in task.Exception.InnerExceptions)
                        {
                            Console.WriteLine("Caught exception: {innerException.Message}");
                        }
                        _cts.Cancel();
                    }
                    else if (task.IsCanceled)
                    {
                        // Clean up.
                        await Presenter.CleanUp();
                    }
                });
                break;
            case ViewEvent.Faulty:
                Presenter.HandleFaultyAsync(_cts.Token).ContinueWith((task) =>
                {
                    if (task.IsFaulted)
                    {
                        // Clean up and cancel all other tasks.
                        foreach (var innerException in task.Exception.InnerExceptions)
                        {
                            Console.WriteLine("Caught exception: {innerException.Message}");
                        }
                        _cts.Cancel();
                    }
                    else if (task.IsCanceled)
                    {
                        // Clean up.
                        await Presenter.CleanUp();
                    }

                });
                break;
            case ViewEvent.Stop:
                _cts.Cancel();
                break;
            case ViewEvent.Reset:
                _cts.Dispose();
                _cts = new CancellationTokenSource();
                break;
        }
    }
}

public async Task HandleStartAsync(CancellationToken token)
{
    await Task.Run(async () =>
    {
        token.ThrowIfCancellationRequested();
        await Task.Delay(2000); // Simulated work. Would be passed token.
        token.ThrowIfCancellationRequested();
    });
}

public async Task HandleFaultyAsync(CancellationToken token)
{
    await Task.Run(async () =>
    {
        token.ThrowIfCancellationRequested();
        await Task.Delay(2000); // Simulated work that failed. Would be passed token.
        throw new InvalidOperationException("FAILED ON PURPOSE");
    });
}

但当前方案存在问题：每个async Task可能会执行重复的清理工作。请问是否可以添加一个PerformCleanUp布尔标志，在switch前执行，对所有运行中的async Task调用Task.WaitAll等待其退出后再执行清理？该标志由失败的Task在ContinueWith中设置一次，后续任务会跳过设置。

回答 #

可以通过添加清理标志结合统一任务跟踪的方式解决重复清理问题，同时优化任务管理逻辑：

核心思路 #

1. 用线程安全集合跟踪所有正在运行的任务，确保能统一等待任务退出
2. 借助原子操作（如Interlocked类）设置清理标志，确保仅触发一次清理流程
3. 处理新事件前先检查清理状态，等待所有任务终止后再执行清理或处理新事件

优化后的代码示例 #

// 新增线程安全的任务集合和原子性清理标志
private readonly ConcurrentBag<Task> _runningTasks = new ConcurrentBag<Task>();
private int _cleanupTriggered = 0; // 0=未触发,1=已触发

public async Task HandleEventsBlocking()
{
    foreach (var viewEvent in _viewEventsQueue.GetConsumingEnumerable())
    {
        // 先检查是否需要执行清理
        if (Interlocked.CompareExchange(ref _cleanupTriggered, 0, 1) == 1)
        {
            // 等待所有运行中任务完成终止
            var tasks = _runningTasks.ToArray();
            await Task.WhenAll(tasks);
            // 执行统一清理操作
            await Presenter.CleanUp();
            // 重置取消令牌
            _cts.Dispose();
            _cts = new CancellationTokenSource();
        }

        switch (viewEvent)
        {
            case ViewEvent.Start:
                var startTask = Presenter.HandleStartAsync(_cts.Token);
                TrackTask(startTask);
                break;
            case ViewEvent.Faulty:
                var faultyTask = Presenter.HandleFaultyAsync(_cts.Token);
                TrackTask(faultyTask);
                break;
            case ViewEvent.Stop:
                TriggerCleanup();
                break;
            case ViewEvent.Reset:
                TriggerCleanup();
                break;
        }
    }
}

private void TrackTask(Task task)
{
    _runningTasks.Add(task);
    task.ContinueWith(t =>
    {
        _runningTasks.TryTake(out _); // 任务完成后从集合移除
        if (t.IsFaulted)
        {
            foreach (var ex in t.Exception.InnerExceptions)
            {
                Console.WriteLine($"Caught exception: {ex.Message}");
            }
            TriggerCleanup();
        }
        // 取消状态不单独执行清理，统一由TriggerCleanup触发
    }, TaskContinuationOptions.ExecuteSynchronously);
}

private void TriggerCleanup()
{
    // 原子操作确保仅触发一次清理
    if (Interlocked.Exchange(ref _cleanupTriggered, 1) == 0)
    {
        _cts.Cancel();
    }
}

// Presenter方法优化：将取消令牌传递给支持取消的异步操作
public async Task HandleStartAsync(CancellationToken token)
{
    await Task.Run(async () =>
    {
        token.ThrowIfCancellationRequested();
        await Task.Delay(2000, token); // 传递token让Delay支持取消
        token.ThrowIfCancellationRequested();
    });
}

public async Task HandleFaultyAsync(CancellationToken token)
{
    await Task.Run(async () =>
    {
        token.ThrowIfCancellationRequested();
        await Task.Delay(2000, token);
        throw new InvalidOperationException("FAILED ON PURPOSE");
    });
}

关键改进点 #

• 避免重复清理：通过Interlocked.Exchange原子操作设置标志，确保清理逻辑只执行一次
• 统一任务管理：用ConcurrentBag<Task>跟踪任务，清理时通过Task.WhenAll等待全部任务终止
• 集中触发清理：任务故障、Stop、Reset事件都通过TriggerCleanup统一触发取消和清理流程
• 及时响应取消：将取消令牌传递给所有支持取消的异步操作，保证任务能快速终止

这样既解决了重复清理的问题，又能保证切换任务或处理故障时，先完成现有任务的终止和清理，再处理新事件。

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