.NET Core：无控制器场景下的自定义Scoped依赖注入作用域

针对你的场景——每个AMQP消息需要独立的Scoped级数据实例，且能在多个服务间共享——核心思路是为每一条消息的处理生命周期手动创建独立的DI作用域，完全复刻HTTP请求中Scoped服务的隔离效果。下面是具体的实现步骤和代码示例：

1. 定义Scoped级的上下文对象 #

首先创建你需要全程收集数据的类，将其注册为Scoped服务，确保每个作用域内只有一个实例：

public class MessageProcessingContext
{
    // 标记当前消息的唯一ID，用于追踪
    public Guid MessageId { get; set; }
    // 存储收集到的业务数据
    public Dictionary<string, object> CollectedData { get; } = new();
}

在Program.cs中注册这个Scoped服务：

builder.Services.AddScoped<MessageProcessingContext>();
// 同时注册你的业务服务（比如消息处理器、数据收集器等）
builder.Services.AddScoped<IMessageProcessor, OrderMessageProcessor>();
builder.Services.AddScoped<IBusinessDataCollector, OrderDataCollector>();

2. 在消息消费逻辑中创建作用域 #

因为默认的Scoped绑定HTTP请求生命周期，而AMQP消息没有这个天然的边界，所以我们需要借助IServiceScopeFactory手动为每条消息创建独立作用域。

以RabbitMQ原生消费者为例，实现代码如下：

public class AmqpMessageConsumer
{
    private readonly IServiceScopeFactory _scopeFactory;
    private readonly IModel _rabbitMqChannel;

    // 注入Singleton级的IServiceScopeFactory（安全且高效）
    public AmqpMessageConsumer(IServiceScopeFactory scopeFactory, IModel rabbitMqChannel)
    {
        _scopeFactory = scopeFactory;
        _rabbitMqChannel = rabbitMqChannel;
    }

    public void StartListening()
    {
        var consumer = new EventingBasicConsumer(_rabbitMqChannel);
        consumer.Received += async (sender, args) =>
        {
            // 为当前消息创建独立作用域，using语句确保作用域自动释放
            using var scope = _scopeFactory.CreateScope();
            var serviceProvider = scope.ServiceProvider;

            try
            {
                // 从作用域中获取Scoped上下文，每个消息对应唯一实例
                var messageContext = serviceProvider.GetRequiredService<MessageProcessingContext>();
                messageContext.MessageId = Guid.Parse(args.BasicProperties.MessageId);

                // 获取消息处理器（同样是当前作用域内的实例）
                var processor = serviceProvider.GetRequiredService<IMessageProcessor>();
                await processor.ProcessMessageAsync(args.Body.ToArray(), messageContext);

                // 处理成功后确认消息
                _rabbitMqChannel.BasicAck(args.DeliveryTag, multiple: false);
            }
            catch (Exception ex)
            {
                // 处理失败逻辑：拒绝消息并重新入队（根据业务调整）
                _rabbitMqChannel.BasicNack(args.DeliveryTag, multiple: false, requeue: true);
                // 这里添加日志记录
            }
        };

        _rabbitMqChannel.BasicConsume(queue: "order_queue", autoAck: false, consumer: consumer);
    }
}

3. 在业务服务中共享上下文 #

在你的业务服务（比如消息处理器、数据收集器）中，直接注入MessageProcessingContext即可，DI容器会自动提供当前消息作用域内的实例：

public interface IMessageProcessor
{
    Task ProcessMessageAsync(byte[] messageBody, MessageProcessingContext context);
}

public class OrderMessageProcessor : IMessageProcessor
{
    private readonly IBusinessDataCollector _dataCollector;
    private readonly MessageProcessingContext _context;

    // 注入的上下文就是当前消息对应的实例
    public OrderMessageProcessor(IBusinessDataCollector dataCollector, MessageProcessingContext context)
    {
        _dataCollector = dataCollector;
        _context = context;
    }

    public async Task ProcessMessageAsync(byte[] messageBody, MessageProcessingContext context)
    {
        // 解析消息内容
        var orderMessage = JsonSerializer.Deserialize<OrderMessage>(messageBody);
        
        // 向上下文收集数据
        _context.CollectedData.Add("OrderId", orderMessage.OrderId);
        
        // 调用其他服务，共享同一个上下文
        await _dataCollector.CollectOrderMetadataAsync(_context);
        
        // 后续业务逻辑：比如保存数据、触发其他流程等
    }
}

4. 第三方AMQP框架的简化方案 #

如果你使用MassTransit、RabbitMQ.Client.Extensions等成熟框架，它们已经内置了消息级的作用域支持，无需手动创建作用域：

以MassTransit为例，只需注册Scoped服务，消费者中直接注入即可：

// Program.cs中注册MassTransit
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<OrderMessageConsumer>();
    x.UsingRabbitMq((ctx, cfg) =>
    {
        cfg.Host("localhost");
        cfg.ReceiveEndpoint("order_queue", e => e.ConfigureConsumer<OrderMessageConsumer>(ctx));
    });
});

// 消费者类
public class OrderMessageConsumer : IConsumer<OrderMessage>
{
    private readonly MessageProcessingContext _context;
    private readonly IBusinessDataCollector _dataCollector;

    public OrderMessageConsumer(MessageProcessingContext context, IBusinessDataCollector dataCollector)
    {
        _context = context;
        _dataCollector = dataCollector;
    }

    public async Task Consume(ConsumeContext<OrderMessage> context)
    {
        _context.MessageId = context.MessageId.Value;
        _context.CollectedData.Add("Order", context.Message);
        await _dataCollector.ProcessAsync(_context);
    }
}

核心原理说明 #

• 每个消息处理时创建的IServiceScope是完全隔离的，Scoped服务的实例仅在当前作用域内有效；
• using语句确保作用域处理完消息后自动释放，避免内存泄漏；
• 并发处理多条消息时，每个消息的作用域相互独立，不会出现数据串扰的问题（只要上下文对象本身是线程安全的，或者单消息处理是单线程操作）。

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