Hangfire, .Net Core и Entity Framework: исключение параллелизма

Question

Я занимаюсь разработкой базового приложения.Net с Hangfire и столкнулся со следующим исключением

Вторая операция началась в этом контексте до завершения предыдущей операции. Любые члены экземпляра не гарантированно являются потокобезопасными.

Я использовал 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) была заблокирована для ресурсов блокировки другого процесса и была выбрана в качестве жертвы тупика. Перезапустите транзакцию. */

Я планирую работу cron от Hangfire, как показано ниже:

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. */ 
  }
}

InsertOrder.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);
}

Insert.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. SqlAzureExecutionStrategy Исключение в CreateOrder.cs (транзакция заблокирована)

Ответ. Решил эту проблему, повторив запрос автоматически при возникновении взаимоблокировки.

Спасибо odinserj за предложения.