ASP.NET Core WebApi能否将IAsyncEnumerable<T>作为输入参数及实现方式？

可以将IAsyncEnumerable<T>作为ASP.NET Core WebApi控制器的输入参数，但默认模型绑定器不直接支持该类型，需要通过流式数据解析的方式实现，最常用的是基于NDJSON（每行一个JSON对象）的流式处理方案。

实现方式 #

方式一：直接读取请求流（简单快速） #

无需额外配置，直接在控制器中读取请求体的流，逐行反序列化数据：

[Route("Operation")]
[HttpPost]
public async Task Operation()
{
    using var reader = new StreamReader(Request.Body);
    while (!reader.EndOfStream)
    {
        var line = await reader.ReadLineAsync();
        if (!string.IsNullOrWhiteSpace(line))
        {
            var command = JsonSerializer.Deserialize<Entity>(line);
            // 执行单个实体的业务逻辑
            await ProcessSingleCommand(command);
        }
    }
}

private async Task ProcessSingleCommand(Entity command)
{
    // 你的业务处理代码
}

方式二：自定义输入格式化器（支持直接注入IAsyncEnumerable） #

如果希望直接将IAsyncEnumerable<T>作为方法参数，可以自定义NDJSON输入格式化器，让框架自动解析流式数据：

1. 实现NDJSON输入格式化器

public class NdJsonInputFormatter : TextInputFormatter
{
    public NdJsonInputFormatter()
    {
        SupportedMediaTypes.Add(MediaTypeHeaderValue.Parse("application/x-ndjson"));
        SupportedEncodings.Add(Encoding.UTF8);
        SupportedEncodings.Add(Encoding.Unicode);
    }

    protected override bool CanReadType(Type type)
    {
        // 判断是否为IAsyncEnumerable<T>类型
        return typeof(IAsyncEnumerable<>).IsAssignableFrom(type) || 
               type.GetInterfaces().Any(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IAsyncEnumerable<>));
    }

    public override async Task<InputFormatterResult> ReadRequestBodyAsync(InputFormatterContext context, Encoding encoding)
    {
        var request = context.HttpContext.Request;
        var itemType = context.ModelType.GetGenericArguments()[0];
        // 创建泛型异步枚举器实例
        var asyncEnumerable = Activator.CreateInstance(
            typeof(AsyncEnumerable<>).MakeGenericType(itemType),
            request.Body, encoding) as IAsyncEnumerable<object>;

        return await InputFormatterResult.SuccessAsync(asyncEnumerable);
    }

    // 内部泛型异步枚举器实现
    private class AsyncEnumerable<T> : IAsyncEnumerable<T>
    {
        private readonly Stream _stream;
        private readonly Encoding _encoding;

        public AsyncEnumerable(Stream stream, Encoding encoding)
        {
            _stream = stream;
            _encoding = encoding;
        }

        public async IAsyncEnumerator<T> GetAsyncEnumerator(CancellationToken cancellationToken = default)
        {
            using var reader = new StreamReader(_stream, _encoding);
            while (!reader.EndOfStream)
            {
                var line = await reader.ReadLineAsync(cancellationToken);
                if (!string.IsNullOrWhiteSpace(line))
                {
                    var item = JsonSerializer.Deserialize<T>(line);
                    if (item != null)
                    {
                        yield return item;
                    }
                }
            }
        }
    }
}

2. 注册格式化器

在Program.cs中添加自定义格式化器：

builder.Services.AddControllers(options =>
{
    // 将NDJSON格式化器插入到格式化器列表的首位
    options.InputFormatters.Insert(0, new NdJsonInputFormatter());
});

3. 控制器方法使用IAsyncEnumerable参数

[Route("Operation")]
[HttpPost]
public async Task Operation([FromBody] IAsyncEnumerable<Entity> commands)
{
    await foreach (Entity command in commands)
    {
        await ProcessSingleCommand(command);
    }
}

注意事项 #

• 请求的Content-Type必须设置为application/x-ndjson
• 请求体格式要求每行一个独立的JSON对象，示例：

  {"Id":1,"Name":"Entity1"}
  {"Id":2,"Name":"Entity2"}
  {"Id":3,"Name":"Entity3"}
  

• 该方案适合处理大数据量的流式输入，避免一次性加载所有数据到内存，提升系统性能和内存利用率

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