插入 SQL 数据库时的 C# 嵌套 Parallel.ForEach答案

【问题标题】：C# Nested Parallel.ForEach while inserting into SQL database插入 SQL 数据库时的 C# 嵌套 Parallel.ForEach
【发布时间】：2018-06-27 07:51:23
【问题描述】：

我们有一个对象（XML 或 JSON），我们成功地将它映射到 DTO，插入数据库需要很长时间（5~7 分钟），所以我们通过Parallel.ForEach，但最终，我们注意到有一些数据输入不正确，比如Category 的所有项目都具有相同的名称，但其他不同的属性是 100% 正确的，在其他情况下，我们注意到一个类别中的所有数据都是相同的，尽管提供的 JSON 对象没有。

我承认它太快了，不到一分钟，但是插入错误，请看下面的代码：

JSON

[
  {
    "CategoryId": 1,
    "CategoryName": "Drinks",
    "SortOrder": 1,
    "Products": [
      {
        "ProductId": 100,
        "ProductName": "Black Tea",
        "SortOrder": 1,
        "Price": 5,
        "Choices": []
      },
      {
        "ProductId": 101,
        "ProductName": "Turkish Coffee",
        "SortOrder": 2,
        "Price": 7.5,
        "Choices": []
      },
      {
        "ProductId": 102,
        "ProductName": "Green Tea",
        "SortOrder": 3,
        "Price": 6,
        "Choices": []
      },
      {
        "ProductId": 103,
        "ProductName": "Café Latte Medium",
        "SortOrder": 4,
        "Price": 10,
        "Choices": []
      },
      {
        "ProductId": 104,
        "ProductName": "Orange Juice",
        "SortOrder": 5,
        "Price": 11,
        "Choices": []
      },
      {
        "ProductId": 105,
        "ProductName": "Mixed Berry Juice",
        "SortOrder": 6,
        "Price": 12.5,
        "Choices": []
      }
    ]
  },
  {
    "CategoryId": 1,
    "CategoryName": "Meals",
    "SortOrder": 1,
    "Products": [
      {
        "ProductId": 200,
        "ProductName": "Breakfast Meal",
        "SortOrder": 1,
        "Price": 16,
        "Choices": [
          {
            "ChoiceId": 3000,
            "ChoiceName": "Strawberry Jam",
            "SortOrder": 1,
            "Price": 0
          },
          {
            "ChoiceId": 3001,
            "ChoiceName": "Apricot Jam",
            "SortOrder": 2,
            "Price": 0
          },
          {
            "ChoiceId": 3002,
            "ChoiceName": "Orange Jam",
            "SortOrder": 3,
            "Price": 0
          },
          {
            "ChoiceId": 3003,
            "ChoiceName": "Café Latte",
            "SortOrder": 4,
            "Price": 2
          }
        ]
      },
      {
        "ProductId": 201,
        "ProductName": "Mixed Grill",
        "SortOrder": 1,
        "Price": 30,
        "Choices": [
          {
            "ChoiceId": 3004,
            "ChoiceName": "Moutabal",
            "SortOrder": 1,
            "Price": 0
          },
          {
            "ChoiceId": 3005,
            "ChoiceName": "Mineral Water",
            "SortOrder": 2,
            "Price": 0
          },
          {
            "ChoiceId": 3006,
            "ChoiceName": "French Fries",
            "SortOrder": 2,
            "Price": 0
          },
          {
            "ChoiceId": 3007,
            "ChoiceName": "Grilled Potatoes",
            "SortOrder": 2,
            "Price": 0
          }
        ]
      }
    ]
  }
]

C# 代码

Parallel.ForEach(categories, (category) =>
{
    var newCreatedCategoryId = 0;
    using (var connection = new SqlConnection("CONNECTION_STRING_HERE"))
    {
        connection.Open();
        using (var command = new SqlCommand("SP_INSERT_INTO_CATEGORIES", connection))
        {
            command.CommandType = CommandType.StoredProcedure;
            command.Parameters.AddWithValue("@P1", category.CategoryName);
            command.Parameters.AddWithValue("@P2", category.SortOrder);
            newCreatedCategoryId = int.Parse(command.ExecuteScalar().ToString());
            command.Dispose();
        }

        connection.Close();
    }

    if (newCreatedCategoryId > 0)
    {
        Parallel.ForEach(category.Products, (product) =>
        {
            using (var connection = new SqlConnection("CONNECTION_STRING_HERE"))
            {
                connection.Open();
                using (var command = new SqlCommand("SP_INSERT_INTO_PRODUCTS", connection))
                {
                    command.CommandType = CommandType.StoredProcedure;
                    command.Parameters.AddWithValue("@P1", product.ProductName);
                    command.Parameters.AddWithValue("@P2", product.Price);
                    command.Parameters.AddWithValue("@P3", product.SortOrder);
                    command.Parameters.AddWithValue("@P4", newCreatedCategoryId);
                    command.ExecuteNonQuery();
                    command.Dispose();
                }

                connection.Close();
            }
        });
    }
});

我查看了here，但这不是我们的问题，我们已经在使用SCOPE_IDENTITY() 来获取当前执行范围内最后生成的身份。

另一方面，即使没有TableLock，也不允许使用SqlBulkCopy插入这么多数据。

【问题讨论】：

你能发布实际代码吗？我问的原因是通常你必须在SqlCommand 上将CommandType 指定为StoredProcedure 才能传递参数，这让我认为这不是你的实际代码。
你的 json 有多大？你生成了多少行？
顺便说一句，您可能会发现使用“表值参数”会更快。
还有这个孩子，这就是为什么你不让人们玩 dbs 和并行 foreach :) 说真的，这不是线程安全的
是的，在这种情况下，数据库服务器成为瓶颈，因为它可能（请注意，这取决于服务器设置）只是阻止任何多余的调用。如果它只允许来自一个客户端的 16 个并行连接，而您用 32、64 甚至更多来轰炸它，那么超出限制的那些无论如何都会被阻止。无论如何，线程安全是这里的主要问题。先把它修好。

标签： c# sql-server parallel-processing ado.net parallel.foreach

【解决方案1】：

问题出在newCreatedCategoryId，令我困惑的是为什么你在内部循环中再次调用newCreatedCategoryId = int.Parse(command.ExecuteScalar().ToString());。我的意思是，如果它只是类别的 id，则不需要再次递增。

看看下面的编辑。将第二个Parallel.ForEach 放入标准foreach 可能会更好，我的意思是这一切都是并行工作的。最后 Parallel.ForEach 并不真正适合 IO 任务，正确的模式是异步和等待。说您可能可以使用 TPL 数据流中的 ActionBlock 来利用两全其美的优势。看看我回答的这个问题中的数据流示例Downloading 1,000+ files fast?

Parallel.ForEach(categories, (category) =>
{
    var newCreatedCategoryId = 0;
    using (var connection = new SqlConnection("CONNECTION_STRING_HERE"))
    {
        connection.Open();
        using (var command = new SqlCommand("SP_INSERT_INTO_CATEGORIES", connection))
        {
            command.CommandType = CommandType.StoredProcedure;
            command.Parameters.AddWithValue("@P1", category.CategoryName);
            command.Parameters.AddWithValue("@P2", category.SortOrder);
            newCreatedCategoryId = int.Parse(command.ExecuteScalar().ToString());
            command.Dispose();
        }

        connection.Close();
    }

    if (newCreatedCategoryId > 0)
    {
        foreach(product in category.Products)
        {
            using (var connection = new SqlConnection("CONNECTION_STRING_HERE"))
            {
                connection.Open();
                using (var command = new SqlCommand("SP_INSERT_INTO_PRODUCTS", connection))
                {
                    command.CommandType = CommandType.StoredProcedure;
                    command.Parameters.AddWithValue("@P1", product.ProductName);
                    command.Parameters.AddWithValue("@P2", product.Price);
                    command.Parameters.AddWithValue("@P3", product.SortOrder);
                    command.Parameters.AddWithValue("@P4", newCreatedCategoryId);
                    command.Dispose();
                }

                connection.Close();
            }
        }//);
    }
});

【讨论】：

对于代码，这是我这边的复制/过去错误，我只是在那里调用command.ExecuteNonQuery()，但是从嵌套的newCreatedCategoryId 访问newCreatedCategoryId 是否有问题@ 987654329@ ??
@RahmaAbdulhameed 如果这只是一个复制和粘贴错误，则没有真正的线程问题，即它完全独立并且应该可以工作
非常感谢您的帮助，但从您的角度来看，问题是什么？或者我可以从我身边尝试什么来再次测试它
@RahmaAbdulhameed 尝试我所做的更改，它仍然会发生吗？
我可以看到您已经将嵌套的Parallel.ForEach 转换为连续的foreach，而我本来可以嵌套Parallel.ForEach，但我会尝试多次，看看我是否是否遇到相同的问题...但是，您能否澄清一下转换的目的是什么？它有助于解决这个问题吗？在其他作品中，这是原因吗？

【解决方案2】：

您循环的对象不是线程安全的。您可以添加一个锁定对象，但是这会使操作序列化，并使Parallel.Foreach. You need to change theParallel.ForEach 的目的失效为标准的 ForEach 循环。

Potential Pitfalls in Data and Task Parallelism

【讨论】：

【解决方案3】：

您更改了 Parallel.ForEach 中的 newCreatedCategoryId，这可能会导致数据不正确，因为查询不会按顺序运行。

【讨论】：

这是一个复制/过去错误，我在这里编辑它就像在我的应用程序中一样，我不这样做