时间戳

首页 > 解决方案 > ChannelReader 完成任务在 OperationCanceledException 之后永远不会完成

c# - ChannelReader 完成任务在 OperationCanceledException 之后永远不会完成

问题描述

如果我打电话Stop()，OperationCanceledException是发生了，_writer.TryComplete(exp)是真的。但是_reader.Completion Task还没有完成。

这是频道的期望行为吗？如果是的话，有人可以告诉我如何在Channel不等到它为空并使其Completion Task处于Completed状态的情况下停止它吗？

public interface IItem
{
    Uri SourceUri { get; }

    string TargetPath { get; }
}

public class Item : IItem
{
    public Item(Uri sourceUri, string targetPath)
    {
        SourceUri = sourceUri;
        TargetPath = targetPath;
    }

    public Uri SourceUri { get; }

    public string TargetPath { get; }
}

public class TestService
{
    private readonly ChannelWriter<IItem> _writer;
    private readonly ChannelReader<IItem> _reader;

    private readonly CancellationTokenSource _cts;

    public TestService()
    {
        _cts = new CancellationTokenSource();
        Channel<IItem> channel = Channel.CreateUnbounded<IItem>();
        _reader = channel.Reader;
        _writer = channel.Writer;
    }

    public async Task QueueDownload(IItem information)
    {
        await _writer.WriteAsync(information);
    }

    public void StartDownload()
    {
        Task.Factory.StartNew(async () =>
        {
            await ProcessDownloadAsync();
        }, TaskCreationOptions.LongRunning);
    }

    public void Stop()
    {
        _cts.Cancel();
        //_writer.Complete();
        //_writer = null;
        Console.WriteLine("Stop");
    }

    public async Task Wait()
    {
        await _reader.Completion;
    }

    private async Task ProcessDownloadAsync()
    {
        try
        {
            while (await _reader.WaitToReadAsync(_cts.Token))
            {
                IItem information = await _reader.ReadAsync(_cts.Token);
                using (WebClient webClient = new WebClient())
                {
                    Console.WriteLine(information.TargetPath);
                    await webClient.DownloadFileTaskAsync(information.SourceUri,
                        information.TargetPath);
                }
            }
        }
        catch (OperationCanceledException exp)
        {
            bool res = _writer.TryComplete(exp);
        }

    }
}

static class Program
{
    static async Task Main(string[] args)
    {
        TestService tSvc = new TestService();
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));
        await tSvc.QueueDownload(new Item(new Uri(@"https://images.pexels.com/" +
            @"photos/753626/pexels-photo-753626.jpeg"), @"D:\\Temp\1.png"));

        tSvc.StartDownload();
        Task t = tSvc.Wait();
        tSvc.Stop();

        await t;

        Console.WriteLine("Finished");
    }
}

标签： c#.netsystem.threading.channels

解决方案

该ChannelWriter.Complete方法的行为与人们预期的略有不同。它不会立即使频道的内容无效。相反，它只是阻止在频道中添加更多项目。现有物品仍然有效消费，并且ChannelReader.Completion在所有存储的物品被消费之前，该属性不会完成。

下面的示例演示了这种行为：

var channel = Channel.CreateUnbounded<int>();
channel.Writer.TryWrite(1);
channel.Writer.Complete(new FileNotFoundException());
//channel.Reader.TryRead(out var data);
var completed = channel.Reader.Completion.Wait(500);
Console.WriteLine($"Completion: {(completed ? "OK" : "Timed-out")}");

输出：

Completion: Timed-out

您可以取消注释该channel.Reader.TryRead行，以查看FileNotFoundException要出现的内容。

推荐阅读