在更新句子中 - 事务（进程 ID 64）被死锁

Question

我收到下一个随机错误“事务（进程 ID XX）在锁定资源上与\另一个进程发生死锁，并已被选为死锁牺牲品。重新运行事务”

此代码正在发送大量电子邮件，并更新 2 个表（一个将通知标志设置为 1，另一个存储发送的电子邮件 - 为每个员工）。

在代码中看起来电子邮件发送正确，问题出在更新上面表格的两个查询中。

如何保持并行化功能（出于性能原因需要）而不丢失一些更新的记录？如果需要，可以允许损失一些性能。

下一个示例代码是：

var options = new ParallelOptions();
options.MaxDegreeOfParallelism = 3;

Parallel.ForEach(listEmployees, options, item =>
{
       MailMessage mail = new MailMessage()
       {
            ExecutionTaskId = 2,
            From = example@example.com,
            To = item.email,
            BodyDesc = template.Body,
            SubjectDesc = template.Subject,
            Status = 0,
            CreatedBy = item.persNbr,
            CreatedDate = DateTime.UtcNow,
        };

        SendMail(mail);
});

private static void SendMail(MailMessage item)
{
      System.Net.Mail.MailMessage msg = new System.Net.Mail.MailMessage();

      ........

      msg.To.Add("to@example.com");
      msg.Body = "body";
      msg.BodyEncoding = System.Text.Encoding.UTF8;
      msg.From = new System.Net.Mail.MailAddress(example@example.com);

      item.BodyDesc = "body";
      item.SubjectDesc = "subject";
      
      using (var smtpClient = new System.Net.Mail.SmtpClient(SettingsRepository.GetSetting("WEB_SMTP")))
      {
            smtpClient.Send(msg);
            item.Status = 1;
            item.SentDate = DateTime.Now;

            if (item.ObjectTable.Contains("var_control"))
            {
                 psn.NotificationSent = 1;
                 MailRepository.UpdatePayslipNotification(psn);
                 MailRepository.Update(item);
            }
            else
            {
                  p6n.NotificationSent = 1;
                  MailRepository.UpdateP60Notification(p6n);
                  MailRepository.Update(item);
            }
      }
}

public static void UpdatePayslipNotification(var var1)
{
    .........
    builder.Clear();
    builder.AppendLine("Update [example].[table_example]");
    builder.AppendLine("SET [example].[table_example].[NotificationSent] = " + 1);
    builder.AppendLine("WHERE [example].[table_example].[Id] = " + var1.Id);
    
    using (var sqlCommand = new SqlCommand())
    {
          sqlCommand.Connection = sqlConnection;
          sqlCommand.Transaction = sqlTransaction;
          sqlCommand.CommandTimeout = 0;
          sqlCommand.CommandType = CommandType.Text;
          sqlCommand.CommandText = builder.ToString();

          sqlCommand.ExecuteNonQuery();
     }
     ........
}

public static void Update(MailMessage mail)
{
    .........
    builder.Clear();
    builder.AppendLine("delete from [example].[MailTemp]");

    using (var sqlCommand = new SqlCommand())
    {
          sqlCommand.Connection = sqlConnection;
          sqlCommand.Transaction = sqlTransaction;
          sqlCommand.CommandTimeout = 0;
          sqlCommand.CommandType = CommandType.Text;
          sqlCommand.CommandText = builder.ToString();

          sqlCommand.ExecuteNonQuery();
    }

    using (var sqlBulkCopy = new SqlBulkCopy(sqlConnection, 
    SqlBulkCopyOptions.Default, sqlTransaction))
    {
         sqlBulkCopy.DestinationTableName = "[example].[MailTemp]";
         sqlBulkCopy.BulkCopyTimeout = 0;

         sqlBulkCopy.WriteToServer(dataTable);
    }

    builder.Clear();
    builder.AppendLine("Update [MailMessage]");
    builder.AppendLine("SET [MailMessage].[To]=[example].[MailTemp]. 
    [To],[MailMessage].[Status]=[example].[MailTemp].[Status], 
    [MailMessage].[SentDate]=[example].[MailTemp].[SentDate], 
    [MailMessage].[ErrorMessage]=[example].[MailTemp].[ErrorMessage], 
    [MailMessage].[SubjectDesc]=[example].[MailTemp].[SubjectDesc], 
    [MailMessage].[BodyDesc]=[example].[MailTemp].[BodyDesc], 
    [MailMessage].[From]=[example].[MailTemp].[From]");
    
    builder.AppendLine("FROM [example].[MailMessage] INNER JOIN 
    [example].[MailTemp] ON [example].[MailMessage].[Id]= 
    [example].[MailTemp].[Id]");

    using (var sqlCommand = new SqlCommand())
    {
          sqlCommand.Connection = sqlConnection;
          sqlCommand.Transaction = sqlTransaction;
          sqlCommand.CommandTimeout = 0;
          sqlCommand.CommandType = CommandType.Text;
          sqlCommand.CommandText = builder.ToString();

          sqlCommand.ExecuteNonQuery();
    }

    sqlTransaction.Commit();
}

---部分解决方案---

（最初的问题是使用 MaxDegreeOfParallelism = 2 解决的，但是由于并行 DB 读取返回假值，因此存在一些附加问题。尝试锁定并行读取程序会返回有关并行情况的一般错误：

"发生了一个或多个错误。在 System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions) 在 System.Threading.Tasks.Task.Wait（Int32 毫秒超时，CancellationToken 取消令牌） 在 System.Threading.Tasks.Task.Wait() 在 System.Threading.Tasks.Parallel.ForWorker[TLocal](Int32 fromInclusive, Int32 toExclusive, ParallelOptions parallelOptions, Ac
tion1 body, Action2 bodyWithState, Func4 bodyWithLocal, Func1 localInit, Action1 localFinally) at System.Threading.Tasks.Parallel.ForEachWorker[TSource,TLocal](IList1 list, ParallelOptions parallelOptions, Action1 body, A\ ction2 bodyWithState, Action3 bodyWithStateAndIndex, Func4 bodyWithStateAndLocal, Func5 bodyWithEverything, Func1 localInit, A
ction`1 localFinally)")

object locker = new object();

var options = new ParallelOptions();
options.MaxDegreeOfParallelism = 2;

Parallel.ForEach(listEmployees, options, item => 
{
      MailMessage mail = new MailMessage()
      {
          ExecutionTaskId = 2,
          From = example@example.com,
          To = item.email,
          BodyDesc = template.Body,
          SubjectDesc = template.Subject,
          Status = 0,
          CreatedBy = item.persNbr,
          CreatedDate = DateTime.UtcNow,
      };

      SendMail(mail, locker);
});

private static void SendMail(MailMessage item, object locker)
{
      System.Net.Mail.MailMessage msg = new System.Net.Mail.MailMessage();

      ........

      msg.To.Add("to@example.com");
      msg.Body = "body";
      msg.BodyEncoding = System.Text.Encoding.UTF8;
      msg.From = new System.Net.Mail.MailAddress(example@example.com);

      item.BodyDesc = "body";
      item.SubjectDesc = "subject";
      
      using (var smtpClient = new System.Net.Mail.SmtpClient(SettingsRepository.GetSetting("WEB_SMTP")))
      {
            smtpClient.Send(msg);
            item.Status = 1;
            item.SentDate = DateTime.Now;

            lock (locker)
            {

                if (item.ObjectTable.Contains("var_control"))
                {
                     psn.NotificationSent = 1;
                     MailRepository.UpdatePayslipNotification(psn);
                     MailRepository.Update(item);
                }
                else
                {
                     p6n.NotificationSent = 1;
                     MailRepository.UpdateP60Notification(p6n);
                     MailRepository.Update(item);
                }
            }
      }
}

Answer 1

由于比较耗时的操作是发送邮件，而且更新数据库比较快，可以使用lock序列化更新数据库操作：

object locker = new();
Parallel.ForEach(listEmployees, options, item =>
{
    MailMessage mail = new() { /*...*/ };
    SendMail(mail);
    lock (locker)
    {
        UpdateDataBase(mail);
    }
});

在上面的示例中，使用了专用的object 储物柜。您也可以将listEmployees 或options 用作储物柜，前提是您不要在其他任何地方将相同的实例用于不相关的同步目的。