如何使用Channel改写数组拆分函数并处理传入的数据流

问题背景 #

已有同步函数可根据序列中的数字拆分数组（例如将3, 2, 1, 1, 8, 2, 2, 9, 9拆分为[2, 1, 1]、[8]、[9, 9]），现需要改写为接收Channel<int>作为输入源，异步处理并返回拆分后的子数组。由于Channel是流式结构，无法直接获取长度或通过索引访问元素，原同步逻辑无法直接复用。

解决方案 #

核心思路是通过Channel的Reader异步逐个读取元素：先读取子数组的长度值，再读取对应数量的元素组成子数组，直到Channel的Writer关闭、没有更多元素可读为止。

完整实现代码 #

using System;
using System.Collections.Generic;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Linq;

class Program
{
    static async Task Main(string[] args)
    {
        List<int> list = new List<int>() { 1, 2, 2, 5, 6, 3, 9, 9, 9 };
        var inputChannel = Channel.CreateUnbounded<int>();

        // 模拟向Channel写入流式数据
        _ = Task.Run(async () =>
        {
            foreach (var num in list)
            {
                await inputChannel.Writer.WriteAsync(num);
                await Task.Delay(500); // 模拟延迟，体现流式处理
            }
            inputChannel.Writer.Complete(); // 写入完成后标记Channel完成
        });

        // 读取并处理拆分后的子数组
        await foreach (var subArray in SplitChannelSequence(inputChannel))
        {
            Console.WriteLine($"拆分结果: {string.Join(" ", subArray)}");
        }
    }

    static async IAsyncEnumerable<int[]> SplitChannelSequence(Channel<int> inputChannel)
    {
        var reader = inputChannel.Reader;
        while (await reader.WaitToReadAsync()) // 等待有元素可读
        {
            // 1. 读取子数组的长度
            if (!await reader.TryReadAsync(out int d))
                break;

            // 修正无效长度（≤0的长度视为0，即空数组）
            d = d <= 0 ? 0 : d;

            // 2. 读取d个元素组成子数组
            var subArray = new int[d];
            for (int i = 0; i < d; i++)
            {
                // 如果Channel已无更多元素，提前终止
                if (!await reader.TryReadAsync(out int element))
                {
                    // 截断数组到已读取的长度
                    Array.Resize(ref subArray, i);
                    break;
                }
                subArray[i] = element;
            }

            yield return subArray;
        }
    }
}

关键细节说明 #

• 流式读取：使用reader.WaitToReadAsync()和reader.TryReadAsync()异步读取元素，避免阻塞，适配Channel的流式特性。
• 处理边界情况：
  • 当读取到的长度d≤0时，返回空数组。
  • 如果读取子数组元素时Channel已关闭（无更多元素），则截断数组到已读取的实际长度。
• Channel完成标记：写入端完成数据写入后，必须调用Writer.Complete()，否则读取端会一直等待新元素。
• 异步枚举返回：使用IAsyncEnumerable<int[]>让调用方可以通过await foreach异步遍历拆分结果，符合异步编程范式。

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