Ứng dụng C # đa luồng với các lệnh gọi cơ sở dữ liệu SQL Server

Đây là vấn đề của tôi:

• Khi sử dụng nhiều luồng để chèn / cập nhật / truy vấn dữ liệu trong SQL Server hoặc bất kỳ cơ sở dữ liệu nào, thì deadlock là một thực tế của cuộc sống. Bạn phải cho rằng chúng sẽ xảy ra và xử lý chúng một cách thích hợp.

• Điều đó không có nghĩa là chúng ta không nên cố gắng hạn chế sự xuất hiện của các bế tắc. Tuy nhiên, thật dễ dàng để đọc các nguyên nhân cơ bản gây ra bế tắc và thực hiện các bước để ngăn chặn chúng, nhưng SQL Server sẽ luôn làm bạn ngạc nhiên :-)

Một số lý do dẫn đến bế tắc:

• Quá nhiều luồng - hãy cố gắng giới hạn số luồng ở mức tối thiểu, nhưng tất nhiên chúng tôi muốn có nhiều luồng hơn để có hiệu suất tối đa.

• Không đủ chỉ mục. Nếu các lựa chọn và cập nhật không đủ chọn lọc, SQL sẽ loại bỏ các khóa phạm vi lớn hơn là tốt. Cố gắng chỉ định các chỉ mục thích hợp.

• Quá nhiều chỉ mục. Cập nhật chỉ mục gây ra bế tắc, vì vậy hãy cố gắng giảm chỉ mục xuống mức tối thiểu cần thiết.

• Mức cô lập giao dịch quá cao. Mức cô lập mặc định khi sử dụng .NET là 'Có thể nối tiếp', trong khi mặc định khi sử dụng SQL Server là 'Đã đọc cam kết'. Giảm mức độ cô lập có thể giúp ích rất nhiều (tất nhiên là nếu thích hợp).

Đây là cách tôi có thể giải quyết vấn đề của bạn:

• Tôi sẽ không sử dụng giải pháp phân luồng của riêng mình, tôi sẽ sử dụng thư viện TaskParallel. Phương thức chính của tôi sẽ giống như sau:

  using (var dc = new TestDataContext())
  {
      // Get all the ids of interest.
      // I assume you mark successfully updated rows in some way
      // in the update transaction.
      List<int> ids = dc.TestItems.Where(...).Select(item => item.Id).ToList();
  
      var problematicIds = new List<ErrorType>();
  
      // Either allow the TaskParallel library to select what it considers
      // as the optimum degree of parallelism by omitting the 
      // ParallelOptions parameter, or specify what you want.
      Parallel.ForEach(ids, new ParallelOptions {MaxDegreeOfParallelism = 8},
                          id => CalculateDetails(id, problematicIds));
  }
  

• Thực thi phương thức CalculatedDetails với các lần thử lại đối với các lỗi deadlock

  private static void CalculateDetails(int id, List<ErrorType> problematicIds)
  {
      try
      {
          // Handle deadlocks
          DeadlockRetryHelper.Execute(() => CalculateDetails(id));
      }
      catch (Exception e)
      {
          // Too many deadlock retries (or other exception). 
          // Record so we can diagnose problem or retry later
          problematicIds.Add(new ErrorType(id, e));
      }
  }
  

• Phương pháp CalculDetails cốt lõi

  private static void CalculateDetails(int id)
  {
      // Creating a new DeviceContext is not expensive.
      // No need to create outside of this method.
      using (var dc = new TestDataContext())
      {
          // TODO: adjust IsolationLevel to minimize deadlocks
          // If you don't need to change the isolation level 
          // then you can remove the TransactionScope altogether
          using (var scope = new TransactionScope(
              TransactionScopeOption.Required,
              new TransactionOptions {IsolationLevel = IsolationLevel.Serializable}))
          {
              TestItem item = dc.TestItems.Single(i => i.Id == id);
  
              // work done here
  
              dc.SubmitChanges();
              scope.Complete();
          }
      }
  }
  

• Và tất nhiên, việc triển khai trình trợ giúp thử lại bế tắc của tôi

  public static class DeadlockRetryHelper
  {
      private const int MaxRetries = 4;
      private const int SqlDeadlock = 1205;
  
      public static void Execute(Action action, int maxRetries = MaxRetries)
      {
          if (HasAmbientTransaction())
          {
              // Deadlock blows out containing transaction
              // so no point retrying if already in tx.
              action();
          }
  
          int retries = 0;
  
          while (retries < maxRetries)
          {
              try
              {
                  action();
                  return;
              }
              catch (Exception e)
              {
                  if (IsSqlDeadlock(e))
                  {
                      retries++;
                      // Delay subsequent retries - not sure if this helps or not
                      Thread.Sleep(100 * retries);
                  }
                  else
                  {
                      throw;
                  }
              }
          }
  
          action();
      }
  
      private static bool HasAmbientTransaction()
      {
          return Transaction.Current != null;
      }
  
      private static bool IsSqlDeadlock(Exception exception)
      {
          if (exception == null)
          {
              return false;
          }
  
          var sqlException = exception as SqlException;
  
          if (sqlException != null && sqlException.Number == SqlDeadlock)
          {
              return true;
          }
  
          if (exception.InnerException != null)
          {
              return IsSqlDeadlock(exception.InnerException);
          }
  
          return false;
      }
  }
  

• Một khả năng khác là sử dụng chiến lược phân vùng

Nếu bảng của bạn có thể được phân vùng một cách tự nhiên thành nhiều bộ dữ liệu riêng biệt, thì bạn có thể sử dụng các bảng và chỉ mục được phân vùng của SQL Server hoặc bạn có thể chia các bảng hiện có của mình thành nhiều bộ bảng theo cách thủ công. Tôi khuyên bạn nên sử dụng phân vùng của SQL Server, vì tùy chọn thứ hai sẽ lộn xộn. Tính năng phân vùng tích hợp cũng chỉ khả dụng trên SQL Enterprise Edition.

Nếu có thể phân vùng cho bạn, bạn có thể chọn một lược đồ phân vùng đã phá vỡ dữ liệu của bạn, giả sử có 8 tập hợp riêng biệt. Bây giờ bạn có thể sử dụng mã luồng đơn ban đầu của mình, nhưng có 8 luồng, mỗi luồng nhắm mục tiêu một phân vùng riêng biệt. Bây giờ sẽ không có bất kỳ (hoặc ít nhất một số lượng tối thiểu) bế tắc.

Tôi hy vọng điều đó đúng.