目录

一、引言

上一篇博客
分享了使用
Channel
来实现针对大量数据的多线程异步处理，感谢大哥们在评论中提出的宝贵的问题和建议！本篇将分享使用
ActionBlock
如何实现，欢迎在评论区留言讨论。

二、ActionBlock介绍

什么是 ActionBlock？

ActionBlock
是 .NET 中 TPL Dataflow 库的一部分，用于处理数据流和并行任务。它提供了一种简单而强大的方式来处理并行任务，并且可以轻松地实现生产者-消费者模式。

ActionBlock 的特点

  • 并行处理：
    ActionBlock
    可以配置为并行处理多个任务，从而提高处理效率
  • 异步编程：
    支持异步编程模型，可以避免阻塞线程，提高应用程序的响应速度和吞吐量
  • 数据流控制：
    可以通过设置最大并行度和其他选项来控制数据流的处理方式
  • 任务调度：
    可以用于调度和管理并行任务，确保任务按预期执行

ActionBlock 的使用场景

  • 生产者-消费者模式
    ：可以用于实现生产者-消费者模式，其中生产者将数据发送到
    ActionBlock
    ，消费者从
    ActionBlock
    中读取数据并进行处理
  • 数据流处理
    ：适用于需要处理大量数据并且需要并行处理的场景，例如日志处理、数据转换等
  • 任务调度
    ：可以用于调度和管理并行任务，确保任务按预期执行

ActionBlock 的基本用法

使用
ActionBlock
非常简单，主要步骤如下：

  1. 创建 ActionBlock：定义一个 ActionBlock，指定要执行的操作和并行选项
  2. 发送数据到 ActionBlock：使用
    SendAsync
    方法将数据发送到 ActionBlock
  3. 完成 ActionBlock：在所有数据发送完成后，调用
    Complete
    方法通知 ActionBlock 不再接收新的数据
  4. 等待处理完成：使用
    Completion
    属性等待所有数据处理完成

以下是一个简单的示例代码，展示了如何使用 ActionBlock：

using System.Threading.Tasks.Dataflow;

var actionBlock = new ActionBlock<int>(async item =>
{
    // 模拟异步处理
    await Task.Delay(100);
    Console.WriteLine($"Processed item: {item}");
}, new ExecutionDataflowBlockOptions
{
    MaxDegreeOfParallelism = 4 // 设置最大并行度
});

// 发送数据到 ActionBlock
for (int i = 0; i < 10; i++)
{
    await actionBlock.SendAsync(i);
}

// 完成 ActionBlock
actionBlock.Complete();
// 等待处理完成
await actionBlock.Completion;

Console.WriteLine("All items processed.");

三、假设场景

假设我们有一组数据需要经过两个步骤的处理。每个数据项都需要进行初步处理，然后进行进一步处理。希望步骤2可以在步骤1产生结果数据后立即开始处理，而不是等待步骤1完全处理完毕。

四、解决方案

使用
TransformBlock
和
ActionBlock
来实现生产者-消费者模式。生产者负责读取数据并将其发送到
TransformBlock
中，消费者从
TransformBlock
中读取数据并进行处理。
以下是一个简单的示例代码，演示如何使用
TransformBlock
和
ActionBlock
实现生产者-消费者模式来处理数据：

using System.Threading.Tasks.Dataflow;

var cts = new CancellationTokenSource();
// 假设有一组数据
var dataItems = Enumerable.Range(0, 1000).Select(x => $"data_{x}").ToList();

var processor = new DataProcessor(10, cts.Token);
await processor.ProcessAsync(dataItems);

Console.ReadKey();

/// <summary>
/// 数据处理器
/// </summary>
public class DataProcessor(int maxDegreeOfParallelism, CancellationToken cancellationToken)
{
    public async Task ProcessAsync(List<string> dataItems)
    {
        // 创建一个 TransformBlock 用于步骤1的处理，并将结果发送到步骤2的 ActionBlock
        var step1Block = new TransformBlock<string, string>(async dataItem => await Step1(dataItem), new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = maxDegreeOfParallelism,
            CancellationToken = cancellationToken
        });
         
        // 创建一个 ActionBlock 用于步骤2的处理
        var step2Block = new ActionBlock<string>(async dataItem =>
        {
            await Step2(dataItem);
        }, new ExecutionDataflowBlockOptions
        {
            MaxDegreeOfParallelism = maxDegreeOfParallelism,
            CancellationToken = cancellationToken
        });

        // 将 TransformBlock 链接到 ActionBlock
        step1Block.LinkTo(step2Block, new DataflowLinkOptions { PropagateCompletion = true });

        // 启动多个步骤1的任务（生产者）
        foreach (var dataItem in dataItems)
        {
            await step1Block.SendAsync(dataItem, cancellationToken);
        }

        // 完成步骤1的 TransformBlock 的写入
        step1Block.Complete();
        // 等待步骤1的 TransformBlock 处理完成
        await step1Block.Completion;

        // 完成步骤2的 ActionBlock 的写入
        step2Block.Complete();
        // 等待步骤2的 ActionBlock 处理完成
        await step2Block.Completion;
    }

    private async Task<string> Step1(string dataItem)
    {
        // 模拟步骤1的处理（如初步处理数据）
        await Task.Delay(10, cancellationToken);
        Console.WriteLine($"Step1 processed data item: {dataItem}");
        return dataItem;
    }

    private async Task Step2(string dataItem)
    {
        // 模拟步骤2的处理（如进一步处理数据）
        await Task.Delay(10, cancellationToken);
        Console.WriteLine($"Step2 processed data item: {dataItem}");
    }
}

代码解释:

  1. 创建Step1的 TransformBlock
    ：在
    ProcessAsync
    方法中，我们首先创建了一个 TransformBlock，用于Step1的处理，TransformBlock 接受一个输入数据项，进行处理后返回一个输出数据项，
    TransformBlock<string, string>
    表示输入和输出都是
    string
    类型
  2. 创建Step2的 ActionBlock
    ：创建一个 ActionBlock 用于Step2的处理，ActionBlock 接受一个输入数据项并进行处理，但不返回输出数据项。
    ActionBlock<string>
    表示输入是
    string
    类型
  3. 链接 TransformBlock 和 ActionBlock
    ：将 TransformBlock 链接到 ActionBlock ，以便将Step1的处理结果发送到Step2进行处理，使用
    LinkTo
    方法将两个块连接起来，并设置
    PropagateCompletion
    为 true，表示当 TransformBlock 完成时，ActionBlock 也会完成
  4. 启动Step1的任务
    ：逐个将数据项发送到 TransformBlock，并等待所有数据处理完成，使用
    SendAsync
    方法将数据项发送到 TransformBlock
  5. 等待任务完成
    ：使用
    Complete
    方法通知 TransformBlock 不再接收新的数据，并使用
    Completion
    属性等待所有数据处理完成。然后完成Step2的 ActionBlock 的写入，并等待Step2的 ActionBlock 处理完成

标签: none

添加新评论