test(cqrs): 补充流式请求基准场景

- 新增 StreamingBenchmarks 并对齐 baseline、GFramework.Cqrs 与 MediatR 的完整枚举对照

- 更新 benchmark README 与 CQRS ai-plan 恢复点,记录 stream 场景落地
This commit is contained in:
gewuyou 2026-05-06 09:14:33 +08:00
parent 96729ddcf1
commit e6f98cb4af
4 changed files with 230 additions and 4 deletions

View File

@ -0,0 +1,183 @@
// Copyright (c) 2025-2026 GeWuYou
// SPDX-License-Identifier: Apache-2.0
using BenchmarkDotNet.Attributes;
using BenchmarkDotNet.Columns;
using BenchmarkDotNet.Configs;
using BenchmarkDotNet.Diagnosers;
using BenchmarkDotNet.Jobs;
using BenchmarkDotNet.Order;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using GFramework.Core.Abstractions.Logging;
using GFramework.Core.Ioc;
using GFramework.Core.Logging;
using GFramework.Cqrs.Abstractions.Cqrs;
using MediatR;
using Microsoft.Extensions.DependencyInjection;
namespace GFramework.Cqrs.Benchmarks.Messaging;
/// <summary>
/// 对比单个 stream request 在直接调用、GFramework.CQRS runtime 与 MediatR 之间的完整枚举开销。
/// </summary>
[Config(typeof(Config))]
public class StreamingBenchmarks
{
private MicrosoftDiContainer _container = null!;
private ICqrsRuntime _runtime = null!;
private ServiceProvider _serviceProvider = null!;
private IMediator _mediatr = null!;
private BenchmarkStreamHandler _baselineHandler = null!;
private BenchmarkStreamRequest _request = null!;
/// <summary>
/// 配置 stream benchmark 的公共输出格式。
/// </summary>
private sealed class Config : ManualConfig
{
public Config()
{
AddJob(Job.Default);
AddColumnProvider(DefaultColumnProviders.Instance);
AddColumn(new CustomColumn("Scenario", static (_, _) => "StreamRequest"));
AddDiagnoser(MemoryDiagnoser.Default);
WithOrderer(new DefaultOrderer(SummaryOrderPolicy.FastestToSlowest, MethodOrderPolicy.Declared));
}
}
/// <summary>
/// 构建 stream dispatch 所需的最小 runtime 宿主和对照对象。
/// </summary>
[GlobalSetup]
public void Setup()
{
LoggerFactoryResolver.Provider = new ConsoleLoggerFactoryProvider
{
MinLevel = LogLevel.Fatal
};
Fixture.Setup("StreamRequest", handlerCount: 1, pipelineCount: 0);
_container = new MicrosoftDiContainer();
_baselineHandler = new BenchmarkStreamHandler();
_container.RegisterTransient<GFramework.Cqrs.Abstractions.Cqrs.IStreamRequestHandler<BenchmarkStreamRequest, BenchmarkResponse>, BenchmarkStreamHandler>();
_runtime = GFramework.Cqrs.CqrsRuntimeFactory.CreateRuntime(
_container,
LoggerFactoryResolver.Provider.CreateLogger(nameof(StreamingBenchmarks)));
var services = new ServiceCollection();
services.AddSingleton<MediatR.IStreamRequestHandler<BenchmarkStreamRequest, BenchmarkResponse>, BenchmarkStreamHandler>();
services.AddMediatR(static options => options.RegisterServicesFromAssembly(typeof(StreamingBenchmarks).Assembly));
_serviceProvider = services.BuildServiceProvider();
_mediatr = _serviceProvider.GetRequiredService<IMediator>();
_request = new BenchmarkStreamRequest(Guid.NewGuid(), 3);
}
/// <summary>
/// 释放 MediatR 对照组使用的 DI 宿主。
/// </summary>
[GlobalCleanup]
public void Cleanup()
{
_serviceProvider.Dispose();
}
/// <summary>
/// 直接调用 handler 并完整枚举响应序列,作为 stream dispatch 额外开销的 baseline。
/// </summary>
[Benchmark(Baseline = true)]
public async ValueTask Stream_Baseline()
{
await foreach (var response in _baselineHandler.Handle(_request, CancellationToken.None).ConfigureAwait(false))
{
_ = response;
}
}
/// <summary>
/// 通过 GFramework.CQRS runtime 创建并完整枚举 stream。
/// </summary>
[Benchmark]
public async ValueTask Stream_GFrameworkCqrs()
{
await foreach (var response in _runtime.CreateStream(BenchmarkContext.Instance, _request, CancellationToken.None)
.ConfigureAwait(false))
{
_ = response;
}
}
/// <summary>
/// 通过 MediatR 创建并完整枚举 stream作为外部设计对照。
/// </summary>
[Benchmark]
public async ValueTask Stream_MediatR()
{
await foreach (var response in _mediatr.CreateStream(_request, CancellationToken.None).ConfigureAwait(false))
{
_ = response;
}
}
/// <summary>
/// Benchmark stream request。
/// </summary>
/// <param name="Id">请求标识。</param>
/// <param name="ItemCount">返回元素数量。</param>
public sealed record BenchmarkStreamRequest(Guid Id, int ItemCount) :
GFramework.Cqrs.Abstractions.Cqrs.IStreamRequest<BenchmarkResponse>,
MediatR.IStreamRequest<BenchmarkResponse>;
/// <summary>
/// 复用 request benchmark 的响应结构,保持跨场景可比性。
/// </summary>
/// <param name="Id">响应标识。</param>
public sealed record BenchmarkResponse(Guid Id);
/// <summary>
/// 同时实现 GFramework.CQRS 与 MediatR 契约的最小 stream handler。
/// </summary>
public sealed class BenchmarkStreamHandler :
GFramework.Cqrs.Abstractions.Cqrs.IStreamRequestHandler<BenchmarkStreamRequest, BenchmarkResponse>,
MediatR.IStreamRequestHandler<BenchmarkStreamRequest, BenchmarkResponse>
{
/// <summary>
/// 处理 GFramework.CQRS stream request。
/// </summary>
public IAsyncEnumerable<BenchmarkResponse> Handle(
BenchmarkStreamRequest request,
CancellationToken cancellationToken)
{
return EnumerateAsync(request, cancellationToken);
}
/// <summary>
/// 处理 MediatR stream request。
/// </summary>
IAsyncEnumerable<BenchmarkResponse> MediatR.IStreamRequestHandler<BenchmarkStreamRequest, BenchmarkResponse>.Handle(
BenchmarkStreamRequest request,
CancellationToken cancellationToken)
{
return EnumerateAsync(request, cancellationToken);
}
/// <summary>
/// 为 benchmark 构造稳定、低噪声的异步响应序列。
/// </summary>
private static async IAsyncEnumerable<BenchmarkResponse> EnumerateAsync(
BenchmarkStreamRequest request,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
for (int index = 0; index < request.ItemCount; index++)
{
cancellationToken.ThrowIfCancellationRequested();
yield return new BenchmarkResponse(request.Id);
await Task.CompletedTask.ConfigureAwait(false);
}
}
}
}

View File

@ -18,6 +18,8 @@
- direct handler、`GFramework.Cqrs` runtime 与 `MediatR` 的 request steady-state dispatch 对比
- `Messaging/NotificationBenchmarks.cs`
- `GFramework.Cqrs` runtime 与 `MediatR` 的单处理器 notification publish 对比
- `Messaging/StreamingBenchmarks.cs`
- direct handler、`GFramework.Cqrs` runtime 与 `MediatR` 的 stream request 完整枚举对比
## 最小使用方式
@ -31,5 +33,4 @@ dotnet run --project GFramework.Cqrs.Benchmarks/GFramework.Cqrs.Benchmarks.cspro
- pipeline behavior 数量矩阵
- generated invoker provider 与纯反射 dispatch 对比
- stream request benchmark
- cold-start 与 registration 成本对比

View File

@ -7,7 +7,7 @@ CQRS 迁移与收敛。
## 当前恢复点
- 恢复点编号:`CQRS-REWRITE-RP-084`
- 恢复点编号:`CQRS-REWRITE-RP-085`
- 当前阶段:`Phase 8`
- 当前 PR 锚点:`PR #323`
- 当前结论:
@ -19,7 +19,8 @@ CQRS 迁移与收敛。
- `RP-081` 已继续补齐基础 generation gate 的 logging 与 DI runtime contract 缺失分支
- 当前 `RP-082` 已补齐基础 generation gate 的 request handler runtime contract 缺失分支
- `RP-083` 已补齐 mixed direct / reflected-implementation request 与 stream invoker provider 发射顺序回归
- 当前 `RP-084` 已引入独立 `GFramework.Cqrs.Benchmarks` 项目,作为持续吸收 `Mediator` benchmark 组织方式的第一落点
- `RP-084` 已引入独立 `GFramework.Cqrs.Benchmarks` 项目,作为持续吸收 `Mediator` benchmark 组织方式的第一落点
- 当前 `RP-085` 已补齐 stream request benchmark对齐 `Mediator` messaging benchmark 的第二个核心场景
- `ai-plan` active 入口现以 `PR #323``RP-082` 为唯一权威恢复锚点;`PR #307`、其他更早 PR 与阶段细节均以下方归档或说明为准
## 当前活跃事实
@ -51,6 +52,9 @@ CQRS 迁移与收敛。
- 结果:通过,`2/2` passed
- `dotnet build GFramework.Cqrs.Benchmarks/GFramework.Cqrs.Benchmarks.csproj -c Release`
- 结果:通过,`0 warning / 0 error`
- `dotnet build GFramework.Cqrs.Benchmarks/GFramework.Cqrs.Benchmarks.csproj -c Release`
- 结果:通过,`0 warning / 0 error`
- 备注:包含新增 `StreamingBenchmarks` 后再次复核通过
- `GIT_DIR=<worktree-git-dir> GIT_WORK_TREE=<worktree-root> python3 scripts/license-header.py --check`
- 结果:通过
- `git diff --check`
@ -103,7 +107,7 @@ CQRS 迁移与收敛。
## 下一推荐步骤
1. 继续处理 `PR #323` 的剩余 review 收尾,优先保持 `ai-plan` active 入口与 trace 的单一锚点一致
2. 若继续推进“吸收 Mediator 设计哲学”的切片,优先扩展 benchmark 场景矩阵到 pipeline、stream、cold-start 与 generated invoker provider 对照
2. 若继续推进“吸收 Mediator 设计哲学”的切片,优先扩展 benchmark 场景矩阵到 request pipeline 数量矩阵、cold-start / initialization 与 generated invoker provider 对照
3. 在进入下一批 runtime / generator 收敛前,保持最小 Release build、targeted test 或 benchmark project build 作为权威验证
## 活跃文档

View File

@ -296,3 +296,41 @@
1. 继续扩展 `GFramework.Cqrs.Benchmarks`,优先补齐 pipeline、stream、cold-start 与 generated invoker provider 对照场景
2. 当后续有具体 runtime 优化切片时,用该 benchmark 项目验证是否真正吸收到了 `Mediator` 的低开销 dispatch 设计收益
### 阶段stream request benchmark 对照CQRS-REWRITE-RP-085
- 继续沿用 `$gframework-batch-boot 50`,当前 branch diff 相对 `origin/main` 仍明显低于阈值
- 在 `RP-084` 已建立独立 benchmark 项目后,本轮优先补齐 `ai-libs/Mediator/benchmarks/Mediator.Benchmarks/Messaging/StreamingBenchmarks.cs` 对应的最小 stream 场景
- 选择 stream 作为第二批 benchmark 的原因:
- 已有独立的 `CreateStream` runtime 路径和单独的 stream invoker provider 元数据契约
- 与 `Mediator` 的 messaging benchmark 分层直接对应
- 不需要像 pipeline / cold-start 那样先进一步澄清运行时或宿主边界
- 本轮新增:
- `GFramework.Cqrs.Benchmarks/Messaging/StreamingBenchmarks.cs`
- `GFramework.Cqrs.Benchmarks/README.md` 中的 stream 场景说明
- 设计约束:
- 保持与前一批一致的三路对照:`Baseline``GFramework.Cqrs``MediatR`
- 基准测量“完整枚举 3 个元素”的全量消费成本,而不是只测创建异步枚举器
- 使用最小 `ICqrsContext` marker继续避免把完整 `ArchitectureContext` 初始化成本混入 steady-state stream dispatch
- 结论:
- 当前 benchmark 项目已经覆盖 `Request``Notification``StreamRequest` 三个核心 messaging steady-state 场景
- 下一批更适合转向 request pipeline 数量矩阵或 cold-start / initialization而不是继续扩同层次的 messaging 基线
### 验证RP-085
- `dotnet build GFramework.Cqrs.Benchmarks/GFramework.Cqrs.Benchmarks.csproj -c Release`
- 结果:通过,`0 warning / 0 error`
- `GIT_DIR=<worktree-git-dir> GIT_WORK_TREE=<worktree-root> python3 scripts/license-header.py --check`
- 结果:通过
- `git diff --check`
- 结果:通过
### 当前 stop-condition 度量RP-085
- primary metricbranch diff files vs `origin/main`
- 当前说明:新增 stream benchmark 后仍处于 `50` 文件阈值以内,适合继续下一批 request pipeline 或 cold-start 场景
### 当前下一步RP-085
1. 继续扩展 `GFramework.Cqrs.Benchmarks`,优先补齐 request pipeline 数量矩阵,随后再评估 cold-start / initialization
2. 当需要验证 generated invoker provider 的实际收益时,把 request benchmark 扩展为 reflection / generated provider 对照,而不是只停留在框架间对比