Skip to content

Feat/cqrs optimization #347

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
GeWuYou merged 5 commits into from
May 11, 2026
Merged

Feat/cqrs optimization #347

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
79ae5f0
test(cqrs-benchmarks): 拆分 streaming benchmark 观测口径
GeWuYou May 10, 2026
11a6b6a
test(cqrs-benchmarks): 扩展流式基准观测并补齐 scoped 生命周期矩阵
GeWuYou May 11, 2026
594798d
test(cqrs-benchmarks): 补齐 stream scoped 生命周期矩阵
GeWuYou May 11, 2026
0baa662
test(cqrs-benchmarks): 隔离并发 benchmark 运行工件
GeWuYou May 11, 2026
1420bd4
fix(cqrs): 修复 benchmark scoped 宿主与 PR 恢复入口
GeWuYou May 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 151 additions & 1 deletion GFramework.Cqrs.Benchmarks/Messaging/BenchmarkHostFactory.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,10 @@
// SPDX-License-Identifier: Apache-2.0

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using GFramework.Core.Abstractions.Logging;
using GFramework.Core.Ioc;
using GFramework.Cqrs.Abstractions.Cqrs;
Expand Down Expand Up @@ -158,14 +161,122 @@ internal static ServiceProvider CreateMediatRServiceProvider(
return services.BuildServiceProvider();
}

/// <summary>
/// 在真实的 request 级作用域内执行一次 GFramework.CQRS request 分发。
/// </summary>
/// <typeparam name="TResponse">请求响应类型。</typeparam>
/// <param name="runtime">复用的 scoped benchmark runtime。</param>
/// <param name="scopedContainer">负责为每次 request 激活独立作用域的只读容器适配层。</param>
/// <param name="context">当前 CQRS 分发上下文。</param>
/// <param name="request">要发送的 request。</param>
/// <param name="cancellationToken">取消令牌。</param>
/// <returns>当前 request 的响应结果。</returns>
/// <remarks>
/// 该入口只服务 request lifetime benchmark：它会复用同一个 dispatcher/runtime 实例，
/// 但在每次调用前后显式创建并释放新的 DI 作用域，
/// 让 `Scoped` handler 在真实 request 边界内解析，而不是退化为根容器解析或额外计入 runtime 构造成本。
/// </remarks>
internal static async ValueTask<TResponse> SendScopedGFrameworkRequestAsync<TResponse>(
ICqrsRuntime runtime,
ScopedBenchmarkContainer scopedContainer,
ICqrsContext context,
GFramework.Cqrs.Abstractions.Cqrs.IRequest<TResponse> request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(runtime);
ArgumentNullException.ThrowIfNull(scopedContainer);
ArgumentNullException.ThrowIfNull(context);
ArgumentNullException.ThrowIfNull(request);

using var scopeLease = scopedContainer.EnterScope();
return await runtime.SendAsync(context, request, cancellationToken).ConfigureAwait(false);
}

/// <summary>
/// 在真实的 request 级作用域内执行一次 MediatR request 分发。
/// </summary>
/// <typeparam name="TResponse">请求响应类型。</typeparam>
/// <param name="rootServiceProvider">当前 benchmark 的根 <see cref="ServiceProvider" />。</param>
/// <param name="request">要发送的 request。</param>
/// <param name="cancellationToken">取消令牌。</param>
/// <returns>当前 request 的响应结果。</returns>
/// <remarks>
/// 这里显式从新的 scope 解析 <see cref="IMediator" />，确保 `Scoped` handler 与其依赖绑定到 request 边界。
/// </remarks>
internal static async Task<TResponse> SendScopedMediatRRequestAsync<TResponse>(
ServiceProvider rootServiceProvider,
MediatR.IRequest<TResponse> request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(rootServiceProvider);
ArgumentNullException.ThrowIfNull(request);

using var scope = rootServiceProvider.CreateScope();
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
return await mediator.Send(request, cancellationToken).ConfigureAwait(false);
}

/// <summary>
/// 在真实的 request 级作用域内创建一次 GFramework.CQRS stream，并让该作用域覆盖整个异步枚举周期。
/// </summary>
/// <typeparam name="TResponse">stream 响应元素类型。</typeparam>
/// <param name="runtime">复用的 scoped benchmark runtime。</param>
/// <param name="scopedContainer">负责为每次 stream 激活独立作用域的只读容器适配层。</param>
/// <param name="context">当前 CQRS 分发上下文。</param>
/// <param name="request">要创建 stream 的 request。</param>
/// <param name="cancellationToken">取消令牌。</param>
/// <returns>绑定到单次显式作用域的异步响应序列。</returns>
/// <remarks>
/// stream 与 request 的区别在于：handler 解析发生在建流时，但 scoped 依赖必须一直存活到枚举完成。
/// 因此这里返回一个包装后的 async iterator，把 scope 的释放时机推迟到调用方结束枚举之后，
/// 避免 `Scoped` handler 退化成“建流后立刻释放 scope，再在根容器语义下继续枚举”的错误模型。
/// </remarks>
internal static IAsyncEnumerable<TResponse> CreateScopedGFrameworkStream<TResponse>(
ICqrsRuntime runtime,
ScopedBenchmarkContainer scopedContainer,
ICqrsContext context,
GFramework.Cqrs.Abstractions.Cqrs.IStreamRequest<TResponse> request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(runtime);
ArgumentNullException.ThrowIfNull(scopedContainer);
ArgumentNullException.ThrowIfNull(context);
ArgumentNullException.ThrowIfNull(request);

return EnumerateScopedGFrameworkStreamAsync(runtime, scopedContainer, context, request, cancellationToken);
}

/// <summary>
/// 在真实的 request 级作用域内创建一次 MediatR stream，并让该作用域覆盖整个异步枚举周期。
/// </summary>
/// <typeparam name="TResponse">stream 响应元素类型。</typeparam>
/// <param name="rootServiceProvider">当前 benchmark 的根 <see cref="ServiceProvider" />。</param>
/// <param name="request">要创建 stream 的 request。</param>
/// <param name="cancellationToken">取消令牌。</param>
/// <returns>绑定到单次显式作用域的异步响应序列。</returns>
/// <remarks>
/// 这里与 scoped request helper 保持同一组边界约束，但把 scope 生命周期延长到 stream 完整枚举结束，
/// 确保 `Scoped` handler 与依赖不会在首个元素产出前后被提前释放。
/// </remarks>
internal static IAsyncEnumerable<TResponse> CreateScopedMediatRStream<TResponse>(
ServiceProvider rootServiceProvider,
MediatR.IStreamRequest<TResponse> request,
CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(rootServiceProvider);
ArgumentNullException.ThrowIfNull(request);

return EnumerateScopedMediatRStreamAsync(rootServiceProvider, request, cancellationToken);
}

/// <summary>
/// 创建承载 NuGet `Mediator` source-generated concrete mediator 的最小对照宿主。
/// </summary>
/// <param name="configure">补充当前场景的显式服务注册。</param>
/// <returns>可直接解析 generated `Mediator.Mediator` 的 DI 宿主。</returns>
/// <remarks>
/// 当前 benchmark 只把 `Mediator` 作为单例 steady-state 对照组接入，
/// 因为它的 lifetime 由 source generator 在编译期塑形；若后续需要 `Transient` / `Scoped` 矩阵，
/// 因为它的 lifetime 由 source generator 在编译期塑形；若后续需要 `Transient` / `Scoped` 矩阵，
/// 应按 `Mediator` 官方 benchmark 的做法拆成独立 build config，而不是在同一编译产物里混用多个 lifetime。
/// </remarks>
internal static ServiceProvider CreateMediatorServiceProvider(Action<IServiceCollection>? configure)
Expand All @@ -191,4 +302,43 @@ internal static bool ImplementsOpenGenericContract(Type candidateType, Type open
interfaceType.IsGenericType &&
interfaceType.GetGenericTypeDefinition() == openGenericContract);
}

/// <summary>
/// 在单个显式作用域内创建并枚举 GFramework.CQRS stream。
/// </summary>
private static async IAsyncEnumerable<TResponse> EnumerateScopedGFrameworkStreamAsync<TResponse>(
ICqrsRuntime runtime,
ScopedBenchmarkContainer scopedContainer,
ICqrsContext context,
GFramework.Cqrs.Abstractions.Cqrs.IStreamRequest<TResponse> request,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
using var scopeLease = scopedContainer.EnterScope();
var stream = runtime.CreateStream(context, request, cancellationToken);

await foreach (var response in stream.ConfigureAwait(false))
{
cancellationToken.ThrowIfCancellationRequested();
yield return response;
}
}

/// <summary>
/// 在单个显式作用域内创建并枚举 MediatR stream。
/// </summary>
private static async IAsyncEnumerable<TResponse> EnumerateScopedMediatRStreamAsync<TResponse>(
ServiceProvider rootServiceProvider,
MediatR.IStreamRequest<TResponse> request,
[System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken)
{
using var scope = rootServiceProvider.CreateScope();
var mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
var stream = mediator.CreateStream(request, cancellationToken);

await foreach (var response in stream.ConfigureAwait(false))
{
cancellationToken.ThrowIfCancellationRequested();
yield return response;
}
}
}
70 changes: 58 additions & 12 deletions GFramework.Cqrs.Benchmarks/Messaging/RequestLifetimeBenchmarks.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,24 +23,27 @@ namespace GFramework.Cqrs.Benchmarks.Messaging;
/// 对比 request steady-state dispatch 在不同 handler 生命周期下的额外开销。
/// </summary>
/// <remarks>
/// 当前矩阵只覆盖 `Singleton` 与 `Transient`。
/// `Scoped` 在两个 runtime 中都依赖显式作用域边界，而当前 benchmark 宿主故意保持“单根容器最小宿主”模型，
/// 直接把 scoped 解析压到根作用域会让对照语义失真，因此留到未来有真实 scoped host 基线时再扩展。
/// 当前矩阵覆盖 `Singleton`、`Scoped` 与 `Transient`。
/// 其中 `Scoped` 会在每次 request 分发时显式创建并释放真实的 DI 作用域，
/// 避免把 scoped handler 错误地压到根容器解析而扭曲生命周期对照。
/// </remarks>
[Config(typeof(Config))]
public class RequestLifetimeBenchmarks
{
private MicrosoftDiContainer _container = null!;
private ICqrsRuntime _runtime = null!;
private ICqrsRuntime? _runtime;
private ScopedBenchmarkContainer? _scopedContainer;
private ICqrsRuntime? _scopedRuntime;
private ServiceProvider _serviceProvider = null!;
private IMediator _mediatr = null!;
private IMediator? _mediatr;
private BenchmarkRequestHandler _baselineHandler = null!;
private BenchmarkRequest _request = null!;
private ILogger _runtimeLogger = null!;

/// <summary>
/// 控制当前 benchmark 使用的 handler 生命周期。
/// </summary>
[Params(HandlerLifetime.Singleton, HandlerLifetime.Transient)]
[Params(HandlerLifetime.Singleton, HandlerLifetime.Scoped, HandlerLifetime.Transient)]
public HandlerLifetime Lifetime { get; set; }

/// <summary>
Expand All @@ -53,6 +56,11 @@ public enum HandlerLifetime
/// </summary>
Singleton,

/// <summary>
/// 每次 request 在显式作用域内解析并复用 handler 实例。
/// </summary>
Scoped,

/// <summary>
/// 每次分发都重新解析新的 handler 实例。
/// </summary>
Expand Down Expand Up @@ -90,23 +98,38 @@ public void Setup()
_baselineHandler = new BenchmarkRequestHandler();
_request = new BenchmarkRequest(Guid.NewGuid());

_runtimeLogger = LoggerFactoryResolver.Provider.CreateLogger(nameof(RequestLifetimeBenchmarks) + "." + Lifetime);

_container = BenchmarkHostFactory.CreateFrozenGFrameworkContainer(container =>
{
BenchmarkHostFactory.RegisterGeneratedBenchmarkRegistry<GeneratedRequestLifetimeBenchmarkRegistry>(container);
RegisterGFrameworkHandler(container, Lifetime);
});
// 容器内已提前保留默认 runtime 以支撑 generated registry 接线；
// 这里额外创建带生命周期后缀的 runtime，只是为了区分不同 benchmark 矩阵的 dispatcher 日志。
_runtime = GFramework.Cqrs.CqrsRuntimeFactory.CreateRuntime(
_container,
LoggerFactoryResolver.Provider.CreateLogger(nameof(RequestLifetimeBenchmarks) + "." + Lifetime));
if (Lifetime != HandlerLifetime.Scoped)
{
_runtime = GFramework.Cqrs.CqrsRuntimeFactory.CreateRuntime(
_container,
_runtimeLogger);
}
else
{
_scopedContainer = new ScopedBenchmarkContainer(_container);
_scopedRuntime = GFramework.Cqrs.CqrsRuntimeFactory.CreateRuntime(
_scopedContainer,
_runtimeLogger);
}

_serviceProvider = BenchmarkHostFactory.CreateMediatRServiceProvider(
configure: null,
typeof(RequestLifetimeBenchmarks),
static candidateType => candidateType == typeof(BenchmarkRequestHandler),
ResolveMediatRLifetime(Lifetime));
_mediatr = _serviceProvider.GetRequiredService<IMediator>();
if (Lifetime != HandlerLifetime.Scoped)
{
_mediatr = _serviceProvider.GetRequiredService<IMediator>();
}
}

/// <summary>
Expand Down Expand Up @@ -140,7 +163,17 @@ public ValueTask<BenchmarkResponse> SendRequest_Baseline()
[Benchmark]
public ValueTask<BenchmarkResponse> SendRequest_GFrameworkCqrs()
{
return _runtime.SendAsync(BenchmarkContext.Instance, _request, CancellationToken.None);
if (Lifetime == HandlerLifetime.Scoped)
{
return BenchmarkHostFactory.SendScopedGFrameworkRequestAsync(
_scopedRuntime!,
_scopedContainer!,
BenchmarkContext.Instance,
_request,
CancellationToken.None);
}

return _runtime!.SendAsync(BenchmarkContext.Instance, _request, CancellationToken.None);
}

/// <summary>
Expand All @@ -149,7 +182,15 @@ public ValueTask<BenchmarkResponse> SendRequest_GFrameworkCqrs()
[Benchmark]
public Task<BenchmarkResponse> SendRequest_MediatR()
{
return _mediatr.Send(_request, CancellationToken.None);
if (Lifetime == HandlerLifetime.Scoped)
{
return BenchmarkHostFactory.SendScopedMediatRRequestAsync(
_serviceProvider,
_request,
CancellationToken.None);
}

return _mediatr!.Send(_request, CancellationToken.None);
}

/// <summary>
Expand All @@ -171,6 +212,10 @@ private static void RegisterGFrameworkHandler(MicrosoftDiContainer container, Ha
container.RegisterSingleton<GFramework.Cqrs.Abstractions.Cqrs.IRequestHandler<BenchmarkRequest, BenchmarkResponse>, BenchmarkRequestHandler>();
return;

case HandlerLifetime.Scoped:
container.RegisterScoped<GFramework.Cqrs.Abstractions.Cqrs.IRequestHandler<BenchmarkRequest, BenchmarkResponse>, BenchmarkRequestHandler>();
return;

case HandlerLifetime.Transient:
container.RegisterTransient<GFramework.Cqrs.Abstractions.Cqrs.IRequestHandler<BenchmarkRequest, BenchmarkResponse>, BenchmarkRequestHandler>();
return;
Expand All @@ -189,6 +234,7 @@ private static ServiceLifetime ResolveMediatRLifetime(HandlerLifetime lifetime)
return lifetime switch
{
HandlerLifetime.Singleton => ServiceLifetime.Singleton,
HandlerLifetime.Scoped => ServiceLifetime.Scoped,
HandlerLifetime.Transient => ServiceLifetime.Transient,
_ => throw new ArgumentOutOfRangeException(nameof(lifetime), lifetime, "Unsupported benchmark handler lifetime.")
};
Expand Down
Loading
Loading