并行性对数据表写入事务的锁定有用吗答案

【问题标题】：Is parallelism usefull with a lock for datatable write transactions并行性对数据表写入事务的锁定有用吗
【发布时间】：2013-09-27 03:25:49
【问题描述】：

并行性是否有助于提高锁定对象的性能，应该单线程运行，还是有其他技术？

我注意到在访问数据集并从多个线程添加行时会引发异常。因此，我创建了一个“线程安全”版本，通过在更新行之前锁定表来添加行。此实现有效，但在处理许多事务时显得很慢。

public partial class HaMmeRffl
{
    public partial class PlayerStatsDataTable
    {
        public void AddPlayerStatsRow(int PlayerID, int Year, int StatEnum, int Value, DateTime Timestamp)
        {
            lock (TeamMemberData.Dataset.PlayerStats)
            {
                HaMmeRffl.PlayerStatsRow testrow = TeamMemberData.Dataset.PlayerStats.FindByPlayerIDYearStatEnum(PlayerID, Year, StatEnum);
                if (testrow == null)
                {
                    HaMmeRffl.PlayerStatsRow newRow = TeamMemberData.Dataset.PlayerStats.NewPlayerStatsRow();
                    newRow.PlayerID = PlayerID;
                    newRow.Year = Year;
                    newRow.StatEnum = StatEnum;
                    newRow.Value = Value;
                    newRow.Timestamp = Timestamp;
                    TeamMemberData.Dataset.PlayerStats.AddPlayerStatsRow(newRow);
                }
                else
                {
                    testrow.Value = Value;
                    testrow.Timestamp = Timestamp;
                }

            }
        }
    }
}

现在我可以从多个线程安全地调用它，但它真的给我带来了什么吗？我可以以不同的方式执行此操作以获得更好的性能。比如有没有什么方法可以使用System.Collections.Concurrent命名空间来优化性能或者其他方法？

另外，我在整个数据集更新后更新底层数据库，这需要很长时间。这是否会被视为 I/O 操作，是否值得通过在数据集中的每一行（或某些行数）更新后更新它来使用并行处理。

更新

我编写了一些代码来测试并发处理与顺序处理，这表明执行并发处理需要大约 30% 的时间，我应该在这里使用顺序处理。 我认为这是因为数据库上的锁定导致ConcurrentQueue 的开销比并行处理的收益更高。这个结论是正确的吗？我可以做些什么来加快处理速度，还是我坚持Datatable“你必须同步任何写操作”。

这是我的测试代码，可能不科学。这是定时器和它们之间的调用。

            dbTimer.Restart();
            Queue<HaMmeRffl.PlayersRow.PlayerValue> addPlayerRow = InsertToPlayerQ(addUpdatePlayers);
            Queue<HaMmeRffl.PlayerStatsRow.PlayerStatValue> addPlayerStatRow                     = InsertToPlayerStatQ(addUpdatePlayers);
            UpdatePlayerStatsInDB(addPlayerRow, addPlayerStatRow);
            dbTimer.Stop();
            System.Diagnostics.Debug.Print("Writing to the dataset took {0} seconds single threaded", dbTimer.Elapsed.TotalSeconds);

            dbTimer.Restart();
            ConcurrentQueue<HaMmeRffl.PlayersRow.PlayerValue> addPlayerRows                    = InsertToPlayerQueue(addUpdatePlayers);
            ConcurrentQueue<HaMmeRffl.PlayerStatsRow.PlayerStatValue> addPlayerStatRows                    = InsertToPlayerStatQueue(addUpdatePlayers);
            UpdatePlayerStatsInDB(addPlayerRows, addPlayerStatRows);
            dbTimer.Stop();
            System.Diagnostics.Debug.Print("Writing to the dataset took {0} seconds concurrently", dbTimer.Elapsed.TotalSeconds);

在这两个示例中，我以相同的单线程方式添加到 Queue 和 ConcurrentQueue。唯一的区别是插入到数据表中。单线程方法插入如下：

    private static void UpdatePlayerStatsInDB(Queue<HaMmeRffl.PlayersRow.PlayerValue> addPlayerRows, Queue<HaMmeRffl.PlayerStatsRow.PlayerStatValue> addPlayerStatRows)
    {
        try
        {
            HaMmeRffl.PlayersRow.PlayerValue row;
            while (addPlayerRows.Count > 0)
            {
                row = addPlayerRows.Dequeue();
                TeamMemberData.Dataset.Players.AddPlayersRow(
                    row.PlayerID, row.Name, row.PosEnum, row.DepthEnum,
                    row.TeamID, row.RosterTimestamp, row.DepthTimestamp,
                    row.Active, row.NewsUpdate);
            }
        }
        catch (Exception)
        {
            TeamMemberData.Dataset.Players.RejectChanges();
        }

        try
        {
            HaMmeRffl.PlayerStatsRow.PlayerStatValue row;
            while (addPlayerStatRows.Count > 0)
            {
                row = addPlayerStatRows.Dequeue();
                TeamMemberData.Dataset.PlayerStats.AddUpdatePlayerStatsRow(
                    row.PlayerID, row.Year, row.StatEnum, row.Value, row.Timestamp);
            }
        }
        catch (Exception)
        {
            TeamMemberData.Dataset.PlayerStats.RejectChanges();
        }

        TeamMemberData.Dataset.Players.AcceptChanges();
        TeamMemberData.Dataset.PlayerStats.AcceptChanges();

    }

并发添加如下

    private static void UpdatePlayerStatsInDB(ConcurrentQueue<HaMmeRffl.PlayersRow.PlayerValue> addPlayerRows, ConcurrentQueue<HaMmeRffl.PlayerStatsRow.PlayerStatValue> addPlayerStatRows)
    {
        Action actionPlayer = () =>
        {
            HaMmeRffl.PlayersRow.PlayerValue row;
            while (addPlayerRows.TryDequeue(out row))
            {
                TeamMemberData.Dataset.Players.AddPlayersRow(
                    row.PlayerID, row.Name, row.PosEnum, row.DepthEnum,
                    row.TeamID, row.RosterTimestamp, row.DepthTimestamp,
                    row.Active, row.NewsUpdate);
            }
        };

        Action actionPlayerStat = () =>
        {
            HaMmeRffl.PlayerStatsRow.PlayerStatValue row;
            while (addPlayerStatRows.TryDequeue(out row))
            {
                TeamMemberData.Dataset.PlayerStats.AddUpdatePlayerStatsRow(
                    row.PlayerID, row.Year, row.StatEnum, row.Value, row.Timestamp);
            }
        };

        Action[] actions = new Action[Environment.ProcessorCount * 2];
        for (int i = 0; i < Environment.ProcessorCount; i++)
        {
            actions[i * 2] = actionPlayer;
            actions[i * 2 + 1] = actionPlayerStat;
        }

        try
        {
            // Start ProcessorCount concurrent consuming actions.
            Parallel.Invoke(actions);
        }
        catch (Exception)
        {
            TeamMemberData.Dataset.Players.RejectChanges();
            TeamMemberData.Dataset.PlayerStats.RejectChanges();
        }

        TeamMemberData.Dataset.Players.AcceptChanges();
        TeamMemberData.Dataset.PlayerStats.AcceptChanges();

    }

单线程的时间差为 4.6 秒，parallel.Invoke 的时间差为 6.1。

【问题讨论】：

您可能想要查看某种消息队列系统，并让一个线程处理所有数据库/数据表操作，而程序的其余部分可以在其执行操作后继续执行队列。
@ScottChamberlain 这就是我现在看到的。我打算使用ConcurrentQueue 并尝试使用多个线程来尝试加速数据库事务，但后来想知道我是否应该只使用常规队列和单个线程
完整的实现取决于你如何做你的整个项目（例如除了AddPlayerStatsRow之外还有其他功能）。最简单的方法是BlockingCollection 由在它自己的线程上运行的ConcurrentQueue 支持，（您只需在foreach 循环中调用GetConsumingEnumerable()，然后让它永远运行）。更复杂的解决方案可能涉及编写您自己的扩展至 SynchronizationContext。

标签： c# multithreading concurrency parallel-processing strongly-typed-dataset

【解决方案1】：

锁定和事务不利于并行性和性能。

1)尽量避免锁：不同线程是否需要更新数据集中的同一行？

2) 最小化锁定时间。

对于 db 操作使用可以尝试 ADO.NET 的 Batch Update future：http://msdn.microsoft.com/en-us/library/ms810297.aspx

【讨论】：

不幸的是，当我在没有锁的情况下执行此操作时，我遇到了异常。我有时会用不同的线程更新数据集中的同一行。我不知道如何避免锁定关联的表。我确实相信我拥有最小的锁，因为它基本上是默认 Addrow 方法的包装，您认为我的代码有改进的空间吗？

【解决方案2】：

多线程可以在一定程度上提供帮助，因为一旦数据越过应用程序边界，您将开始等待 I/O，在这里您可以进行异步处理，因为您的应用程序无法控制各种参数（资源访问、网络速度等），这将提供更好的用户体验（如果是 UI 应用程序）。

现在对于您的场景，您可能希望使用某种生产者/消费者队列，只要队列中有一行可用，另一个线程就会开始处理它，但这在一定程度上会起作用。

【讨论】：