Hangfire、.Net Core 和 Entity Framework：并发异常

Question

我正在使用 Hangfire 开发一个 .Net 核心应用程序并面临以下异常

在前一个操作完成之前，在此上下文中启动了第二个操作。不保证任何实例成员都是线程安全的。

我使用 Hangfire 以 1 小时的间隔安排作业。当新进程/作业在早期作业完成其进程之前启动时，我面临上述问题。

我们如何实现多个 Hangfire 进程/作业（多个工作人员）来工作（并行）以完成任务。 （现已解决，使用默认的 AspNetCoreJobActivator）

var scopeFactory = serviceProvider.GetService<IServiceScopeFactory>();
            if (scopeFactory != null)
                GlobalConfiguration.Configuration.UseActivator(new AspNetCoreJobActivator(scopeFactory));

现在，我在 CreateOrderData.cs 中收到以下异常：-

/*System.InvalidOperationException: 引发了一个异常 很可能是由于暂时性故障。如果您要连接到 SQL Azure 数据库考虑使用 SqlAzureExecutionStrategy。 ---> Microsoft.EntityFrameworkCore.DbUpdateException：发生错误 在更新条目时。有关详细信息，请参阅内部异常。 ---> System.Data.SqlClient.SqlException：事务（进程 ID 103）是 与另一个进程在锁定资源上死锁并已被选中 作为僵局的受害者。重新运行事务。 */

我正在安排 hangfire cron 作业，如下所示：-

RecurringJob.AddOrUpdate<IS2SScheduledJobs>(x => x.ProcessInputXML(), Cron.MinuteInterval(1));

Startup.cs

public void ConfigureServices(IServiceCollection services)
{
    string hangFireConnection = Configuration["ConnectionStrings:HangFire"];
    GlobalConfiguration.Configuration.UseSqlServerStorage(hangFireConnection);

    var config = new AutoMapper.MapperConfiguration(cfg =>
    {
       cfg.AddProfile(new AutoMapperProfileConfiguration());
    );

    var mapper = config.CreateMapper();
    services.AddSingleton(mapper);

    services.AddScoped<IHangFireJob, HangFireJob>();
    services.AddScoped<IScheduledJobs, ScheduledJobs>();
    services.AddScoped<BusinessLogic>();
    services.AddHangfire(opt => 
         opt.UseSqlServerStorage(Configuration["ConnectionStrings:HangFire"]));

    services.AddEntityFrameworkSqlServer().AddDbContext<ABCContext>(options => 
         options.UseSqlServer(Configuration["ConnectionStrings:ABC"]));
}

public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory loggerFactory, IServiceProvider serviceProvider)
{
    GlobalConfiguration.Configuration.UseActivator(new HangFireActivator(serviceProvider));

    //hangFireJob.Jobs();

    // add NLog to ASP.NET Core
    loggerFactory.AddConsole(Configuration.GetSection("Logging"));
    loggerFactory.AddDebug();
    loggerFactory.AddNLog();
    // app.UseCors("AllowSpecificOrigin");

    foreach (DatabaseTarget target in LogManager.Configuration.AllTargets.Where(t => t is DatabaseTarget))
    {
        target.ConnectionString = Configuration.GetConnectionString("Logging");
    }

    LogManager.ReconfigExistingLoggers();
}

Hangfire.cs

public class HangFireJob : IHangFireJob
{
        private ABCContext _abcContext;
        private IScheduledJobs scheduledJobs;

        public HangFireJob(ABCContext abcContext, IScheduledJobs scheduledJobs)
        {
            _abcContext = abcContext;
            this.scheduledJobs = scheduledJobs;           
        }

        public void Jobs()
        {
             RecurringJob.AddOrUpdate<IScheduledJobs>(x => x.ProcessInputXML(), Cron.HourInterval(1));
        }
}

ScheduledJobs.cs

public class S2SScheduledJobs : IS2SScheduledJobs
{
    private BusinessLogic _businessLogic;

    public ScheduledJobs(BusinessLogic businessLogic)
    {
        _businessLogic = businessLogic;
    }

    public async Task<string> ProcessInputXML()
    {
        await _businessLogic.ProcessXML();
    }
}

BusinessLogic.cs

public class BusinessLogic
{
    private ABCContext _abcContext;

    public BusinessLogic(ABCContext abcContext) : base(abcContext)
    {
            _abcContext = abcContext;
    }

    public async Task ProcessXML()
    {
       var batchRepository = new BatchRepository(_abcContext);
       var unprocessedBatchRecords = await BatchRepository.GetUnprocessedBatch();

       foreach (var batchRecord in unprocessedBatchRecords)
       {
         try
         {
           int orderId = await LoadDataToOrderTable(batchRecord.BatchId);  
           await UpdateBatchProcessedStatus(batchRecord.BatchId);

           if (orderId > 0)
           {
                await CreateOrderData(orderId);
           }
         }
         catch(Exception ex)
         {
         }
       }
    }

CreateOrderData.cs

public async Task<int> CreateOrderData(int orderId)
{
  try
  {
    await OrderRepo.InsertOrder(order);
    await _abcContext.SaveChangesAsync();   
  }
  catch(Exception ex)
  {
    /*System.InvalidOperationException: An exception has been raised that is likely due to a transient failure. If you are connecting to a SQL Azure database consider using SqlAzureExecutionStrategy. ---> Microsoft.EntityFrameworkCore.DbUpdateException: An error occurred while updating the entries. See the inner exception for details. ---> System.Data.SqlClient.SqlException: Transaction (Process ID 103) was deadlocked on lock resources with another process and has been chosen as the deadlock victim. Rerun the transaction. */ 
  }
}

插入订单.cs

public async Task InsertOrder(Order o)
{
   // creation of large number of entites(more than 50) to be inserted in the database
    woRepo.Insert(p);
    poRepo.Insert(y);
 //and many more like above

    Insert(order);
}

插入.cs

public virtual void Insert(TEntity entity)
    {
        entity.ObjectState = ObjectState.Added;
        if (entity is IXYZEntity xyzEntity)
        {
            xyzEntity.CreatedDate = DateTime.Now;
            xyzEntity.UpdatedDate = xyzEntity.CreatedDate;
            xyzEntity.CreatedBy = _context.UserName ?? string.Empty;
            xyzEntity.UpdatedBy = _context.UserName ?? string.Empty;
        }
        else if (entity is IxyzEntityNull xyzEntityNull)
        {
            xyzEntityNull.CreatedDate = DateTime.Now;
            xyzEntityNull.UpdatedDate = xyzEntityNull.CreatedDate;
            xyzEntityNull.CreatedBy = _context.UserName;
            xyzEntityNull.UpdatedBy = _context.UserName;
        }
        _dbSet.Add(entity);
        _context.SyncObjectState(entity);
    }

LoadDataToOrder.cs

public async Task<int> LoadDataToOrder(int batchId)
{
        //  using (var unitOfWork = new UnitOfWork(_abcContext))
        //  {
        var orderRepo = new OrderRepository(_abcContext);
        Entities.Order order = new Entities.Order();

        order.Guid = Guid.NewGuid();
        order.BatchId = batchId;
        order.VendorId = null;

        orderRepo.Insert(order);
        //unitOfWork.SaveChanges();
        await _abcContext.SaveChangesAsync();
        return order.OrderId;
        //  
}
}

HangfireActivator.cs

public class HangFireActivator : Hangfire.JobActivator
{
        private readonly IServiceProvider _serviceProvider;

        public HangFireActivator(IServiceProvider serviceProvider)
        {
            _serviceProvider = serviceProvider;
        }

        public override object ActivateJob(Type type)
        {
            return _serviceProvider.GetService(type);
        }
}

请指教。

谢谢。

Answer 1

以下解决方案适用于这 2 个问题：

1. 实现多个 Hangfire 进程/作业（多个工人）工作（并行）。 答：当我使用内置的AspNetCoreJobActivator 时，这个问题得到了解决，而不是开箱即用，即删除了 HangfireActivator 类并删除了对 UseActivator 方法的调用。

   var scopeFactory = serviceProvider.GetService<IServiceScopeFactory>();
   if (scopeFactory != null)
       GlobalConfiguration.Configuration.UseActivator(new AspNetCoreJobActivator(scopeFactory));
   

2. SqlAzureExecutionStrategyCreateOrder.cs 中的异常（事务已死锁）

回答：通过在发生死锁时自动重试查询解决了这个问题。

感谢 odinserj 的建议。