UnitOfWork和DbContext：用DI线程安全

Question

我正在研究承载IHostedService的.NET Core 2.2控制台应用程序：

public class MqttClientHostedService : IHostedService, IDisposable
{
    [...]
    public MqttClientHostedService(
        ILogger<MqttClientHostedService> logger, 
        IOptions<MqttClientConfiguration> mqttConfiguration, 
        IPositionService positionService)
    {
        this.logger = logger;
        this.config = mqttConfiguration;
        this.positionService = positionService;
    }

    public async Task StartAsync(CancellationToken cancellationToken)
    {
        mqttClient = new MqttFactory().CreateMqttClient();
        mqttClient.Connected += async (s, e) => await MqttClient_Connected(s, e);
        mqttClient.ApplicationMessageReceived +=
            async (s, e) => await MqttClient_ApplicationMessageReceived(s, e);

        await mqttClient.ConnectAsync(
            new MqttClientOptionsBuilder()
                .WithTcpServer(config.Value.Host, config.Value.Port).Build());
    }

    private async Task MqttClient_ApplicationMessageReceived(
        object sender, MqttApplicationMessageReceivedEventArgs e)
    {
        string message = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
        await positionService.HandleMessage(message);
    }

    [...]
}

这个IPositionService是一个管理器，检查消息并检查它是否可以保存在我们的数据库中：

public class PositionService : IPositionService
{
    [...]

    public PositionService(
        IUnitOfWork unitOfWork, ILogger<PositionService> logger)
    {
        this.unitOfWork = unitOfWork;
        this.logger = logger;
    }

    public async Task HandleMessage(string message)
    {
        Entity entity = await unitOfWork.EntityRepository.GetByMessage(message);

        [...]

        await unitOfWork.EntityRepository.UpdateAsync(entity);
        await unitOfWork.Save();
    }

    [...]
}

IUnitOfWork是实体框架核心DbContext的包装（请不要评判我，我有理由这样做）：

public class UnitOfWork : IUnitOfWork
{
    [...]

    public UnitOfWork(MyContext myContext)
    {
        this.myContext = myContext;

        EntityRepository = new EFRepository<Entity>(myContext);
    }

    public async Task Save()
    {
        await myContext.SaveChangesAsync();
    }
}

EFRepository<T>，实现IRepository<T>接口，是DbSet<T>的包装（再次，请不要评判我）。这里没有相关代码。

控制台应用程序Program.cs配置如下：

[...]
.ConfigureServices((hostContext, services) =>
{
    services.AddDbContext<MyContext>(
        c => c.UseSqlServer("[...]", options => options.UseNetTopologySuite()),
        ServiceLifetime.Transient);

    services.AddTransient<IPositionService, PositionService>();
    services.AddTransient(typeof(IRepository<>), typeof(EFRepository<>));
    services.AddTransient<IUnitOfWork, UnitOfWork>();

    services.AddHostedService<MqttClientHostedService>();

    [...]
});

问题是PositionService.HandleMessage每秒被调用多次，并且由于DbContext不是线程安全的，我收到此错误消息：

在上一个操作完成之前，在此上下文中开始第二个操作。

我通过从IUnitOfWork的依赖项中删除PositionService解决了这个问题，而是注入了IServiceScopeFactory，并执行：

using (IServiceScope serviceScope = serviceScopeFactory.CreateScope())
{
    IUnitOfWork unitOfWork = serviceScope.ServiceProvider.GetService<IUnitOfWork>();

    [...]
}

这种方式有效，但我不喜欢它。这似乎是一个技巧，我不喜欢我的PositionService知道Dependency Injection并且必须处理范围的事实。

我的问题是：有一种更好的方法来解决这个问题而不涉及我的课程？我应该使整个UnitOfWork线程安全吗？或者可以手动创建而不使用DI？

Answer 1

问题的根源是MyContext在以下对象图中被俘虏为俘虏依赖：

MqttClientHostedService
    -> PositionService
        -> UnitOfWork
            -> MyContext 

此图中的所有类型都注册为Transient，但是，作为托管服务的服务（例如，您的MqttClientHostedService）在应用程序的持续时间内仅解析一次并且无限期地缓存。这有效地使他们成为单身人士。

换句话说，MyContext被单个MqttClientHostedService意外地保持活着，并且因为多个消息可以并行进入，你有自己的竞争条件。

解决方案是让每个ApplicationMessageReceived事件在其自己独特的小气泡（范围）中运行，并从该气泡内解析新的IPositionService。例如：

public class MqttClientHostedService : IHostedService, IDisposable
{
    [...]
    public MqttClientHostedService(
        ILogger<MqttClientHostedService> logger, 
        IOptions<MqttClientConfiguration> mqttConfiguration, 
        IServiceProvider provider)
    {
        this.logger = logger;
        this.config = mqttConfiguration;
        this.provider = provider;
    }
    [...]
    private async Task MqttClient_ApplicationMessageReceived(
        object sender, MqttApplicationMessageReceivedEventArgs e)
    {
        using (var scope = provider.CreateScope())
        {
            positionService = scope.ServiceProvider
                .GetRequiredService<IPositionService>();
            string message = Encoding.UTF8.GetString(e.ApplicationMessage.Payload);
            await positionService.HandleMessage(message);
        }
    }

    [...]
}