.NET中边写入Stream边向HTTP控制器发送数据的实现

客户端实现调整 #

核心是用System.IO.Pipelines双向管道实现「边写边发」：

• 创建Pipe实例，一边用Writer端分批写入JSON数据，一边用Reader端作为StreamContent的数据源发请求
• 启动后台任务负责写数据，同时发起HTTP请求，让写入和发送操作并行执行

示例代码：

public async Task<OutputModel> SendStreamAsync(CancellationToken cancellation)
{
    // 创建双向管道
    var pipe = new Pipe();
    
    // 后台任务：分批写入JSON数据到管道
    var writeTask = Task.Run(async () =>
    {
        await using var writer = new JsonTextWriter(new StreamWriter(pipe.Writer.AsStream()))
        {
            Formatting = Formatting.None // 禁用格式化减少数据体积
        };
        
        try
        {
            // 初始化JSON结构
            await writer.WriteStartObjectAsync(cancellation);
            await writer.WritePropertyNameAsync("Events", cancellation);
            await writer.WriteStartArrayAsync(cancellation);
            
            // 分批写入Event数据（替换成你的实际分批逻辑）
            foreach (var eventBatch in GetEventBatches())
            {
                foreach (var @event in eventBatch)
                {
                    await JsonSerializer.CreateDefault().SerializeAsync(writer, @event, cancellation);
                    await writer.FlushAsync(cancellation);
                    await pipe.Writer.FlushAsync(cancellation);
                }
            }
            
            // 结束JSON结构
            await writer.WriteEndArrayAsync(cancellation);
            await writer.WriteEndObjectAsync(cancellation);
            await writer.FlushAsync(cancellation);
        }
        finally
        {
            // 标记写入完成，让Reader知道没有更多数据
            pipe.Writer.Complete();
        }
    }, cancellation);
    
    // 构造请求内容：用PipeReader的Stream作为数据源
    var body = new StreamContent(pipe.Reader.AsStream());
    body.Headers.ContentType = new MediaTypeHeaderValue("application/json");
    
    // 发送请求（此时写入任务和发送操作并行）
    var response = await _httpClient.PostAsync(url, body, cancellation);
    await writeTask; // 等待写入任务完成，确保无写入错误
    
    var content = await response.Content.ReadAsStringAsync(cancellation);
    // 处理响应逻辑...
    
    return result;
}

// 示例：模拟分批获取Event数据
private IEnumerable<IEnumerable<Event>> GetEventBatches()
{
    // 替换成你的实际分批逻辑
    yield return new List<Event> { new Event { /* ... */ } };
    yield return new List<Event> { new Event { /* ... */ } };
}

服务端确认与优化 #

你的服务端代码已经支持「边接收边解析」，只需注意几点：

1. 不要用[FromBody]特性（它会把整个请求体读入内存），直接用HttpContext.Request.Body（你当前代码已经这么做了）
2. 确保ParseStreamAsync里的异步读取逻辑正确，用await ReadAsync避免阻塞
3. 保证AppendEventsAsync能正确响应取消令牌，请求取消后及时停止处理

关键原理 #

• 客户端的Pipe是生产者-消费者模式：写入任务（生产者）分批写数据到管道，HttpClient（消费者）从管道读数据发请求，无需等整个流生成完
• 服务端的JsonTextReader异步读取请求体流，每读到一段数据就解析出Event，通过IAsyncEnumerable实时传给业务逻辑，实现边接收边处理

内容的提问来源于stack exchange，提问作者Alexander I.