具有 SQL Server 数据库调用的多线程 C# 应用程序 [英] Multi threading C# application with SQL Server database calls

问题描述

我有一个 SQL Server 数据库，表 main 中有 500,000 条记录.还有另外三个表，称为child1、child2 和child3.child1、child2、child3和main之间的多对多关系是通过三个关系表实现的:main_child1_relationship、main_child2_relationship 和 main_child3_relationship.我需要读取main 中的记录，更新main，并在关系表中插入新行以及在子表中插入新记录.子表中的记录具有唯一性约束，因此实际计算的伪代码(CalculateDetails)将类似于:

I have a SQL Server database with 500,000 records in table main. There are also three other tables called child1, child2, and child3. The many to many relationships between child1, child2, child3, and main are implemented via the three relationship tables: main_child1_relationship, main_child2_relationship, and main_child3_relationship. I need to read the records in main, update main, and also insert into the relationship tables new rows as well as insert new records in the child tables. The records in the child tables have uniqueness constraints, so the pseudo-code for the actual calculation (CalculateDetails) would be something like:

for each record in main
{
   find its child1 like qualities
   for each one of its child1 qualities
   {
      find the record in child1 that matches that quality
      if found
      {
          add a record to main_child1_relationship to connect the two records
      }
      else
      {
          create a new record in child1 for the quality mentioned
          add a record to main_child1_relationship to connect the two records
      }
   }
   ...repeat the above for child2
   ...repeat the above for child3 
}

作为单线程应用程序运行良好.但它太慢了.C# 中的处理非常繁重并且需要很长时间.我想把它变成一个多线程应用.

This works fine as a single threaded app. But it is too slow. The processing in C# is pretty heavy duty and takes too long. I want to turn this into a multi-threaded app.

这样做的最佳方法是什么?我们正在使用 Linq to Sql.

What is the best way to do this? We are using Linq to Sql.

到目前为止，我的方法是为来自 main 的每批记录创建一个新的 DataContext 对象，并使用 ThreadPool.QueueUserWorkItem 来处理它.然而，这些批次正在相互影响，因为一个线程添加了一条记录，然后下一个线程尝试添加相同的记录……我遇到了各种有趣的 SQL Server 死锁.

So far my approach has been to create a new DataContext object for each batch of records from main and use ThreadPool.QueueUserWorkItem to process it. However these batches are stepping on each other's toes because one thread adds a record and then the next thread tries to add the same one and ... I am getting all kinds of interesting SQL Server dead locks.

代码如下:

    int skip = 0;
    List<int> thisBatch;
    Queue<List<int>> allBatches = new Queue<List<int>>();
    do
    {
        thisBatch = allIds
                .Skip(skip)
                .Take(numberOfRecordsToPullFromDBAtATime).ToList();
        allBatches.Enqueue(thisBatch);
        skip += numberOfRecordsToPullFromDBAtATime;

    } while (thisBatch.Count() > 0);

    while (allBatches.Count() > 0)
    {
        RRDataContext rrdc = new RRDataContext();

        var currentBatch = allBatches.Dequeue();
        lock (locker)  
        {
            runningTasks++;
        }
        System.Threading.ThreadPool.QueueUserWorkItem(x =>
                    ProcessBatch(currentBatch, rrdc));

        lock (locker) 
        {
            while (runningTasks > MAX_NUMBER_OF_THREADS)
            {
                 Monitor.Wait(locker);
                 UpdateGUI();
            }
        }
    }

这里是 ProcessBatch:

And here is ProcessBatch:

    private static void ProcessBatch( 
        List<int> currentBatch, RRDataContext rrdc)
    {
        var topRecords = GetTopRecords(rrdc, currentBatch);
        CalculateDetails(rrdc, topRecords);
        rrdc.Dispose();

        lock (locker)
        {
            runningTasks--;
            Monitor.Pulse(locker);
        };
    }

还有

    private static List<Record> GetTopRecords(RecipeRelationshipsDataContext rrdc, 
                                              List<int> thisBatch)
    {
        List<Record> topRecords;

        topRecords = rrdc.Records
                    .Where(x => thisBatch.Contains(x.Id))
                    .OrderBy(x => x.OrderByMe).ToList();
        return topRecords;
    }

CalculateDetails 最好由顶部的伪代码解释.

CalculateDetails is best explained by the pseudo-code at the top.

我认为必须有更好的方法来做到这一点.请帮忙.非常感谢！

I think there must be a better way to do this. Please help. Many thanks!

推荐答案

以下是我对这个问题的看法:

Here's my take on the problem:

• 当使用多线程在 SQL Server 或任何数据库中插入/更新/查询数据时，死锁是一个现实.你必须假设它们会发生并适当地处理它们.

• When using multiple threads to insert/update/query data in SQL Server, or any database, then deadlocks are a fact of life. You have to assume they will occur and handle them appropriately.

这并不是说我们不应该试图限制死锁的发生.但是，很容易了解死锁的基本原因并采取措施防止它们，但是SQL Server 总是会给你惊喜:-)

That's not so say we shouldn't attempt to limit the occurence of deadlocks. However, it's easy to read up on the basic causes of deadlocks and take steps to prevent them, but SQL Server will always surprise you :-)

导致死锁的一些原因:

• 线程过多 - 尝试将线程数限制在最低限度，但当然我们需要更多线程以获得最大性能.

• Too many threads - try to limit the number of threads to a minimum, but of course we want more threads for maximum performance.

索引不足.如果选择和更新没有足够的选择性，SQL 将取出比正常范围更大的范围锁.尝试指定适当的索引.

Not enough indexes. If selects and updates aren't selective enough SQL will take out larger range locks than is healthy. Try to specify appropriate indexes.

索引过多.更新索引会导致死锁，因此尽量将索引减少到所需的最小值.

Too many indexes. Updating indexes causes deadlocks, so try to reduce indexes to the minimum required.

事务隔离级别太高.使用 .NET 时的默认隔离级别是Serializable"，而使用 SQL Server 的默认设置是已提交读".降低隔离级别会有很大帮助(当然如果合适的话).

Transaction isolational level too high. The default isolation level when using .NET is 'Serializable', whereas the default using SQL Server is 'Read Committed'. Reducing the isolation level can help a lot (if appropriate of course).

这就是我可以解决您的问题的方式:

This is how I might tackle your problem:

• 我不会推出自己的线程解决方案，我会使用 TaskParallel 库.我的主要方法看起来像这样:

• I wouldn't roll my own threading solution, I would use the TaskParallel library. My main method would look something like this:

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

执行CalculateDetails 方法并重试死锁失败

Execute the CalculateDetails method with retries for deadlock failures

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

CalculateDetails 核心方法

The core CalculateDetails method

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();
        }
    }
}

当然还有我对死锁重试助手的实现

And of course my implementation of a deadlock retry helper

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;
    }
}

另一种可能是使用分区策略

One further possibility is to use a partitioning strategy

如果您的表可以自然地划分为几个不同的数据集，那么您可以使用 SQL Server 分区表和索引，或者您可以手动将您现有的表格分割 成几组表格.我建议使用 SQL Server 的分区，因为第二个选项会很麻烦.此外，内置分区仅适用于 SQL 企业版.

If your tables can naturally be partitioned into several distinct sets of data, then you can either use SQL Server partitioned tables and indexes, or you could manually split your existing tables into several sets of tables. I would recommend using SQL Server's partitioning, since the second option would be messy. Also built-in partitioning is only available on SQL Enterprise Edition.

如果您可以进行分区，您可以选择一种将您的数据分成 8 个不同集合的分区方案.现在您可以使用原始的单线程代码，但有 8 个线程，每个线程针对一个单独的分区.现在不会有任何(或至少是最少数量的)死锁.

If partitioning is possible for you, you could choose a partion scheme that broke you data in lets say 8 distinct sets. Now you could use your original single threaded code, but have 8 threads each targetting a separate partition. Now there won't be any (or at least a minimum number of) deadlocks.

我希望这是有道理的.

这篇关于具有 SQL Server 数据库调用的多线程 C# 应用程序的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！