c# – 我应该如何使用DataflowBlockOptions.CancellationToken？

发布时间：2020-12-15 22:13:17 所属栏目：百科来源：网络整理

导读：我怎么能使用DataflowBlockOptions.CancellationToken？如果我像这样创建BufferBlock的实例： var queue = new BufferBlock int(new DataflowBlockOptions {BoundedCapacity = 5,CancellationToken = _cts.Token}); 然后使用使用队列的消费者/生产者方法,我

我怎么能使用DataflowBlockOptions.CancellationToken？

如果我像这样创建BufferBlock的实例：

var queue = new BufferBlock< int>(new DataflowBlockOptions {BoundedCapacity = 5,CancellationToken = _cts.Token});

然后使用使用队列的消费者/生产者方法,我如何使用其CancellationToken来处理取消？

例如.在生产者方法中,我如何检查取消令牌 – 我没有找到任何属性来访问令牌..

编辑：
生产/消费方法的样本：

private static async Task Produce(BufferBlock<int> queue,IEnumerable<int> values)
{
    foreach (var value in values)
    {
        await queue.SendAsync(value);
    }

    queue.Complete();
}

private static async Task<IEnumerable<int>> Consume(BufferBlock<int> queue)
{
    var ret = new List<int>();
    while (await queue.OutputAvailableAsync())
    {
        ret.Add(await queue.ReceiveAsync());
    }

    return ret;
}

调用它的代码：

var queue = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 5,CancellationToken = _cts.Token });

// Start the producer and consumer.
var values = Enumerable.Range(0,10);
Produce(queue,values);
var consumer = Consume(queue);

// Wait for everything to complete.
await Task.WhenAll(consumer,queue.Completion);

EDIT2：

如果我调用_cts.Cancel(),则Produce方法不会取消并完成而不会中断.

解决方法

如果你想取消生产过程,你应该在其中传递令牌,如下所示：

private static async Task Produce(
        BufferBlock<int> queue,IEnumerable<int> values,CancellationToken token
        )
    {
        foreach (var value in values)
        {
            await queue.SendAsync(value,token);
            Console.WriteLine(value);
        }

        queue.Complete();
    }

    private static async Task<IEnumerable<int>> Consume(BufferBlock<int> queue)
    {
        var ret = new List<int>();
        while (await queue.OutputAvailableAsync())
        {
            ret.Add(await queue.ReceiveAsync());
        }

        return ret;
    }

    static void Main(string[] args)
    {
        var cts = new CancellationTokenSource();

        var queue = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 5,CancellationToken = cts.Token });

        // Start the producer and consumer.
        var values = Enumerable.Range(0,100);
        Produce(queue,values,cts.Token);
        var consumer = Consume(queue);

        cts.Cancel();

        try
        {
            Task.WaitAll(consumer,queue.Completion);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }

        foreach (var i in consumer.Result)
        {
            Console.WriteLine(i);
        }

        Console.ReadKey();

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!