dotnet-pipeline-behaviors

MediatR 管道行为

概述

管道行为用于实现横切关注点，在每个命令或查询处理程序执行前后运行：

• 验证 - 在处理程序执行前验证请求
• 日志记录 - 记录请求/响应详情
• 异常处理 - 将异常转换为 Result
• 事务 - 使用数据库事务包装处理程序
• 缓存 - 缓存查询结果
• 性能 - 监控运行缓慢的操作

快速参考

行为	用途	顺序
LoggingBehavior	记录请求	第一（最外层）
ValidationBehavior	验证输入	第二
ExceptionHandlingBehavior	转换异常	第三
TransactionBehavior	数据库事务	第四
CachingBehavior	缓存响应	第五（最内层）

---

行为结构

/Application/Abstractions/Behaviors/
├── LoggingBehavior.cs
├── ValidationBehavior.cs
├── ExceptionHandlingBehavior.cs
├── TransactionBehavior.cs
├── QueryCachingBehavior.cs
└── PerformanceBehavior.cs

---

模板：日志记录行为

// src/{name}.application/Abstractions/Behaviors/LoggingBehavior.cs
using MediatR;
using Microsoft.Extensions.Logging;
using Serilog.Context;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Logs all requests and responses with timing information
/// </summary>
public sealed class LoggingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly ILogger<LoggingBehavior<TRequest, TResponse>> _logger;

    public LoggingBehavior(ILogger<LoggingBehavior<TRequest, TResponse>> logger)
    {
        _logger = logger;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        var requestName = typeof(TRequest).Name;
        var requestId = Guid.NewGuid();

        using (LogContext.PushProperty("RequestId", requestId))
        using (LogContext.PushProperty("RequestName", requestName))
        {
            _logger.LogInformation(
                "Handling {RequestName} ({RequestId})",
                requestName,
                requestId);

            var stopwatch = System.Diagnostics.Stopwatch.StartNew();

            try
            {
                var response = await next();

                stopwatch.Stop();

                _logger.LogInformation(
                    "Handled {RequestName} ({RequestId}) in {ElapsedMs}ms",
                    requestName,
                    requestId,
                    stopwatch.ElapsedMilliseconds);

                return response;
            }
            catch (Exception ex)
            {
                stopwatch.Stop();

                _logger.LogError(
                    ex,
                    "Error handling {RequestName} ({RequestId}) after {ElapsedMs}ms",
                    requestName,
                    requestId,
                    stopwatch.ElapsedMilliseconds);

                throw;
            }
        }
    }
}

---

模板：验证行为

// src/{name}.application/Abstractions/Behaviors/ValidationBehavior.cs
using FluentValidation;
using MediatR;
using {name}.domain.abstractions;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Validates requests using FluentValidation validators
/// Returns ValidationResult with errors instead of throwing
/// </summary>
public sealed class ValidationBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly IEnumerable<IValidator<TRequest>> _validators;

    public ValidationBehavior(IEnumerable<IValidator<TRequest>> validators)
    {
        _validators = validators;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        if (!_validators.Any())
        {
            return await next();
        }

        var context = new ValidationContext<TRequest>(request);

        var validationResults = await Task.WhenAll(
            _validators.Select(v => v.ValidateAsync(context, cancellationToken)));

        var errors = validationResults
            .SelectMany(result => result.Errors)
            .Where(failure => failure is not null)
            .Select(failure => new Error(
                failure.PropertyName,
                failure.ErrorMessage))
            .Distinct()
            .ToArray();

        if (errors.Length != 0)
        {
            return CreateValidationResult<TResponse>(errors);
        }

        return await next();
    }

    private static TResponse CreateValidationResult<TResponse>(Error[] errors)
    {
        // Handle Result type
        if (typeof(TResponse) == typeof(Result))
        {
            return (TResponse)(object)ValidationResult.WithErrors(errors);
        }

        // Handle Result<T> type
        var resultType = typeof(TResponse);

        if (resultType.IsGenericType &&
            resultType.GetGenericTypeDefinition() == typeof(Result<>))
        {
            var valueType = resultType.GetGenericArguments()[0];

            var validationResultType = typeof(ValidationResult<>).MakeGenericType(valueType);

            var validationResult = Activator.CreateInstance(
                validationResultType,
                BindingFlags.Instance | BindingFlags.NonPublic,
                null,
                new object[] { errors },
                null);

            return (TResponse)validationResult!;
        }

        throw new InvalidOperationException(
            $"Cannot create validation result for type {typeof(TResponse).Name}");
    }
}

---

模板：异常处理行为

// src/{name}.application/Abstractions/Behaviors/ExceptionHandlingBehavior.cs
using MediatR;
using Microsoft.Extensions.Logging;
using {name}.domain.abstractions;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Catches unhandled exceptions and converts them to Result.Failure
/// </summary>
public sealed class ExceptionHandlingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
    where TResponse : Result
{
    private readonly ILogger<ExceptionHandlingBehavior<TRequest, TResponse>> _logger;

    public ExceptionHandlingBehavior(
        ILogger<ExceptionHandlingBehavior<TRequest, TResponse>> logger)
    {
        _logger = logger;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        try
        {
            return await next();
        }
        catch (Exception ex)
        {
            var requestName = typeof(TRequest).Name;

            _logger.LogError(
                ex,
                "Unhandled exception for request {RequestName}",
                requestName);

            return CreateExceptionResult<TResponse>(ex);
        }
    }

    private static TResponse CreateExceptionResult<TResponse>(Exception exception)
    {
        var error = new Error(
            "Error.Unhandled",
            exception.Message);

        if (typeof(TResponse) == typeof(Result))
        {
            return (TResponse)(object)Result.Failure(error);
        }

        var resultType = typeof(TResponse);

        if (resultType.IsGenericType &&
            resultType.GetGenericTypeDefinition() == typeof(Result<>))
        {
            var valueType = resultType.GetGenericArguments()[0];

            var failureMethod = typeof(Result)
                .GetMethod(nameof(Result.Failure), new[] { typeof(Error) })!
                .MakeGenericMethod(valueType);

            return (TResponse)failureMethod.Invoke(null, new object[] { error })!;
        }

        throw new InvalidOperationException(
            $"Cannot create exception result for type {typeof(TResponse).Name}");
    }
}

---

模板：事务行为

// src/{name}.application/Abstractions/Behaviors/TransactionBehavior.cs
using MediatR;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using {name}.application.abstractions.messaging;
using {name}.infrastructure;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Wraps command handlers in database transactions
/// Only applies to commands (write operations)
/// </summary>
public sealed class TransactionBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : ICommand<TResponse>  // Only commands
{
    private readonly ApplicationDbContext _dbContext;
    private readonly ILogger<TransactionBehavior<TRequest, TResponse>> _logger;

    public TransactionBehavior(
        ApplicationDbContext dbContext,
        ILogger<TransactionBehavior<TRequest, TResponse>> logger)
    {
        _dbContext = dbContext;
        _logger = logger;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        var requestName = typeof(TRequest).Name;

        // Check if already in a transaction
        if (_dbContext.Database.CurrentTransaction is not null)
        {
            return await next();
        }

        await using var transaction = await _dbContext.Database
            .BeginTransactionAsync(cancellationToken);

        _logger.LogInformation(
            "Beginning transaction for {RequestName}",
            requestName);

        try
        {
            var response = await next();

            await transaction.CommitAsync(cancellationToken);

            _logger.LogInformation(
                "Committed transaction for {RequestName}",
                requestName);

            return response;
        }
        catch (Exception ex)
        {
            await transaction.RollbackAsync(cancellationToken);

            _logger.LogError(
                ex,
                "Rolled back transaction for {RequestName}",
                requestName);

            throw;
        }
    }
}

---

模板：查询缓存行为

// src/{name}.application/Abstractions/Caching/ICachedQuery.cs
namespace {name}.application.abstractions.caching;

/// <summary>
/// Marker interface for queries that should be cached
/// </summary>
public interface ICachedQuery
{
    string CacheKey { get; }
    TimeSpan? CacheDuration { get; }
}

/// <summary>
/// Strongly-typed cached query
/// </summary>
public interface ICachedQuery<TResponse> : IQuery<TResponse>, ICachedQuery
{
}

// src/{name}.application/Abstractions/Behaviors/QueryCachingBehavior.cs
using MediatR;
using Microsoft.Extensions.Caching.Distributed;
using Microsoft.Extensions.Logging;
using System.Text.Json;
using {name}.application.abstractions.caching;
using {name}.domain.abstractions;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Caches query results using distributed cache
/// Only applies to queries implementing ICachedQuery
/// </summary>
public sealed class QueryCachingBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : ICachedQuery<TResponse>
    where TResponse : class
{
    private readonly IDistributedCache _cache;
    private readonly ILogger<QueryCachingBehavior<TRequest, TResponse>> _logger;

    private static readonly JsonSerializerOptions JsonOptions = new()
    {
        PropertyNamingPolicy = JsonNamingPolicy.CamelCase
    };

    public QueryCachingBehavior(
        IDistributedCache cache,
        ILogger<QueryCachingBehavior<TRequest, TResponse>> logger)
    {
        _cache = cache;
        _logger = logger;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        var cacheKey = request.CacheKey;

        // Try to get from cache
        var cachedValue = await _cache.GetStringAsync(cacheKey, cancellationToken);

        if (!string.IsNullOrEmpty(cachedValue))
        {
            _logger.LogInformation(
                "Cache hit for {CacheKey}",
                cacheKey);

            return JsonSerializer.Deserialize<TResponse>(cachedValue, JsonOptions)!;
        }

        _logger.LogInformation(
            "Cache miss for {CacheKey}",
            cacheKey);

        // Execute query
        var response = await next();

        // Cache the result if successful
        if (response is Result { IsSuccess: true })
        {
            var cacheOptions = new DistributedCacheEntryOptions
            {
                AbsoluteExpirationRelativeToNow = request.CacheDuration ?? TimeSpan.FromMinutes(5)
            };

            var serialized = JsonSerializer.Serialize(response, JsonOptions);

            await _cache.SetStringAsync(
                cacheKey,
                serialized,
                cacheOptions,
                cancellationToken);

            _logger.LogInformation(
                "Cached response for {CacheKey}",
                cacheKey);
        }

        return response;
    }
}

使用缓存查询

// src/{name}.application/{Feature}/Get{Entity}ById/Get{Entity}ByIdQuery.cs
public sealed record Get{Entity}ByIdQuery(Guid Id) 
    : ICachedQuery<{Entity}Response>
{
    public string CacheKey => $"{Entity}:{Id}";
    public TimeSpan? CacheDuration => TimeSpan.FromMinutes(10);
}

---

模板：性能行为

// src/{name}.application/Abstractions/Behaviors/PerformanceBehavior.cs
using System.Diagnostics;
using MediatR;
using Microsoft.Extensions.Logging;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Logs a warning for slow requests (>500ms by default)
/// </summary>
public sealed class PerformanceBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IRequest<TResponse>
{
    private readonly ILogger<PerformanceBehavior<TRequest, TResponse>> _logger;
    private readonly Stopwatch _timer;
    private const int SlowRequestThresholdMs = 500;

    public PerformanceBehavior(
        ILogger<PerformanceBehavior<TRequest, TResponse>> logger)
    {
        _logger = logger;
        _timer = new Stopwatch();
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        _timer.Start();

        var response = await next();

        _timer.Stop();

        var elapsedMs = _timer.ElapsedMilliseconds;

        if (elapsedMs > SlowRequestThresholdMs)
        {
            var requestName = typeof(TRequest).Name;

            _logger.LogWarning(
                "Long running request: {RequestName} ({ElapsedMs}ms) - {@Request}",
                requestName,
                elapsedMs,
                request);
        }

        return response;
    }
}

---

模板：幂等性行为

// src/{name}.application/Abstractions/Idempotency/IIdempotentCommand.cs
namespace {name}.application.abstractions.idempotency;

/// <summary>
/// Marker interface for commands that support idempotency
/// </summary>
public interface IIdempotentCommand
{
    Guid IdempotencyKey { get; }
}

// src/{name}.application/Abstractions/Behaviors/IdempotencyBehavior.cs
using MediatR;
using Microsoft.Extensions.Logging;
using {name}.application.abstractions.idempotency;
using {name}.domain.abstractions;

namespace {name}.application.abstractions.behaviors;

/// <summary>
/// Prevents duplicate command execution using idempotency keys
/// </summary>
public sealed class IdempotencyBehavior<TRequest, TResponse>
    : IPipelineBehavior<TRequest, TResponse>
    where TRequest : IIdempotentCommand, IRequest<TResponse>
    where TResponse : Result
{
    private readonly IIdempotencyService _idempotencyService;
    private readonly ILogger<IdempotencyBehavior<TRequest, TResponse>> _logger;

    public IdempotencyBehavior(
        IIdempotencyService idempotencyService,
        ILogger<IdempotencyBehavior<TRequest, TResponse>> logger)
    {
        _idempotencyService = idempotencyService;
        _logger = logger;
    }

    public async Task<TResponse> Handle(
        TRequest request,
        RequestHandlerDelegate<TResponse> next,
        CancellationToken cancellationToken)
    {
        // Check if already processed
        if (await _idempotencyService.ExistsAsync(
            request.IdempotencyKey,
            cancellationToken))
        {
            _logger.LogInformation(
                "Duplicate request detected with key {IdempotencyKey}",
                request.IdempotencyKey);

            // Return cached response or success
            return await _idempotencyService
                .GetResponseAsync<TResponse>(request.IdempotencyKey, cancellationToken)
                ?? CreateSuccessResult<TResponse>();
        }

        var response = await next();

        // Store the response
        await _idempotencyService.SaveAsync(
            request.IdempotencyKey,
            response,
            cancellationToken);

        return response;
    }

    private static TResponse CreateSuccessResult<TResponse>()
    {
        if (typeof(TResponse) == typeof(Result))
        {
            return (TResponse)(object)Result.Success();
        }

        var resultType = typeof(TResponse);

        if (resultType.IsGenericType &&
            resultType.GetGenericTypeDefinition() == typeof(Result<>))
        {
            // Return default success - caller should use cached response instead
            throw new InvalidOperationException(
                "Cannot create default success for generic Result. " +
                "Cached response should be used.");
        }

        throw new InvalidOperationException(
            $"Cannot create success result for type {typeof(TResponse).Name}");
    }
}

---

注册行为

// src/{name}.application/DependencyInjection.cs
using FluentValidation;
using Microsoft.Extensions.DependencyInjection;
using {name}.application.abstractions.behaviors;

namespace {name}.application;

public static class DependencyInjection
{
    public static IServiceCollection AddApplication(this IServiceCollection services)
    {
        services.AddMediatR(configuration =>
        {
            configuration.RegisterServicesFromAssembly(typeof(DependencyInjection).Assembly);

            // Register behaviors in order (outer to inner)
            // Logging is outermost - sees everything
            configuration.AddOpenBehavior(typeof(LoggingBehavior<,>));

            // Performance monitoring
            configuration.AddOpenBehavior(typeof(PerformanceBehavior<,>));

            // Validation - reject invalid requests early
            configuration.AddOpenBehavior(typeof(ValidationBehavior<,>));

            // Exception handling - convert exceptions to Results
            configuration.AddOpenBehavior(typeof(ExceptionHandlingBehavior<,>));

            // Transaction - wrap commands in transactions
            // Note: Only add if using EF Core directly in Application layer
            // configuration.AddOpenBehavior(typeof(TransactionBehavior<,>));
        });

        services.AddValidatorsFromAssembly(typeof(DependencyInjection).Assembly);

        return services;
    }
}

---

行为执行顺序

Request
    │
    ▼
┌─────────────────────┐
│  LoggingBehavior    │  ← Outermost: logs request start
│  ┌─────────────────┐│
│  │ PerformanceBeh. ││  ← Starts timer
│  │ ┌─────────────┐ ││
│  │ │ Validation  │ ││  ← Validates request
│  │ │ ┌─────────┐ │ ││
│  │ │ │Exception│ │ ││  ← Catches exceptions
│  │ │ │ ┌─────┐ │ │ ││
│  │ │ │ │Trans.│ │ │ ││  ← Begins transaction
│  │ │ │ │ ┌─┐ │ │ │ ││
│  │ │ │ │ │H│ │ │ │ ││  ← Handler executes
│  │ │ │ │ └─┘ │ │ │ ││
│  │ │ │ └─────┘ │ │ ││  ← Commits/Rolls back
│  │ │ └─────────┘ │ ││  ← Catches, converts to Result
│  │ └─────────────┘ ││  ← Stops timer, logs slow
│  └─────────────────┘│
└─────────────────────┘  ← Logs request end
    │
    ▼
Response

---

关键规则

1. 注册顺序很重要 - 最先注册的行为位于最外层
2. 泛型约束 - 仅用于命令的行为使用 where TRequest : ICommand
3. 不要吞掉异常 - 记录日志并重新抛出，或转换为 Result
4. 保持行为专注 - 每个行为只负责一项职责
5. 使用开放泛型 - 使用 typeof(Behavior<,>)，而不是 typeof(Behavior<Cmd, Resp>)
6. 全程使用异步 - 绝不要使用 .Result 或 .Wait() 阻塞
7. 不要修改请求 - 行为是观察者，而非转换器
8. 事务行为放在最后 - 在处理程序执行前、验证后
9. 缓存读取，不缓存写入 - 仅缓存查询结果
10. 在适当的级别记录日志 - 正常情况使用 Info，运行缓慢时使用 Warning，失败时使用 Error

---

应避免的反模式

// ❌ WRONG: Behavior that modifies request
public async Task<TResponse> Handle(...)
{
    request.ModifiedAt = DateTime.UtcNow;  // Don't modify!
    return await next();
}

// ✅ CORRECT: Behaviors observe, don't modify
public async Task<TResponse> Handle(...)
{
    _logger.LogInformation("Processing at {Time}", DateTime.UtcNow);
    return await next();
}

// ❌ WRONG: Swallowing exceptions silently
try { return await next(); }
catch { return default!; }  // Silent failure!

// ✅ CORRECT: Log and convert or rethrow
try { return await next(); }
catch (Exception ex)
{
    _logger.LogError(ex, "Error in handler");
    return CreateFailureResult(ex);
}

// ❌ WRONG: Blocking async code
var result = next().Result;  // Deadlock risk!

// ✅ CORRECT: Await properly
var result = await next();

// ❌ WRONG: Caching commands
public sealed class CachingBehavior<TRequest, TResponse>
    where TRequest : ICommand<TResponse>  // Commands shouldn't be cached!

// ✅ CORRECT: Cache only queries
public sealed class CachingBehavior<TRequest, TResponse>
    where TRequest : ICachedQuery<TResponse>

---

相关技能

• dotnet-cqrs-command-generator - 通过行为流转的命令
• dotnet-cqrs-query-generator - 通过行为流转的查询
• dotnet-result-pattern - 行为使用的 Result 类型
• dotnet-clean-architecture - 应用层的放置位置