主页 / 问题 / 1600766
Accepted
air
Asked:2024-11-25 04:49:54 +0000 UTC2024-11-25 04:49:54 +0000 UTC

使用线程安全字典时的数据竞争

  • 772

问题很简单：为什么线程安全字典允许数据竞争，因为该字典中的所有方法都是原子的，并且两个线程不应该接收相同的值？

在下面的示例中，可以显示至少10个、至少12个、至少14个。

public class Program
{
    private static ConcurrentDictionary<string, int> _map = [];

    static async Task Main(string[] args)
    {
        _map["air"] = 0;
        var tasks = new List<Task>();

        for (int i = 0; i < 10000; i++)
        {
            tasks.Add(Task.Run(() =>
            {
                for (int i = 0; i < 10; i++)
                {
                    _map.TryGetValue("air", out int value);
                    _map.AddOrUpdate("air", value, (key, oldValue) => oldValue + 1);

                    _map.TryGetValue("air", out int newValue);
                    if (newValue >= 10)
                    {
                        _map.TryRemove("air", out _);
                        _map.TryAdd("air1", newValue);
                    }
                }
            }));
        }

        await Task.WhenAll(tasks);
        Console.WriteLine(_map["air1"]);
    }
}
c#

1 个回答

  1. Best Answer

    字典负责在每个单独的操作中访问它时的线程安全，并且您有多个操作。操作之间任何事情都可能发生。此外，lambdaAddOrUpdate不是在锁定下执行的，也不是原子操作的一部分。

    顺便说一句，您可能会感到困惑，但所有这些循环都没有意义，因为_map.TryAdd("air1", newValue);它们在程序的整个执行过程中只会工作一次，即当键不在字典中时。我假设您想要对这个键中的值求和，因此下面的代码考虑了此错误的修复。

    您可以解决字典之外的计算，即从那里获取一个值，用它做任何您需要做的事情，然后返回它。

    private static ConcurrentDictionary<string, int> _map = [];

static async Task Main(string[] args)
{
    var tasks = new List<Task>();
    for (int i = 0; i < 10000; i++)
    {
        tasks.Add(Task.Run(() =>
        {
            while (!_map.TryAdd("air", 0)) { }
            for (int i = 0; i < 10; i++)
            {
                int value;
                while (!_map.TryRemove("air", out value)) { }
                value++;
                if (value >= 10)
                {
                    _map.AddOrUpdate("air1", 10, (k, v) => v += 10);
                }
                else
                    while (!_map.TryAdd("air", value)) { }
            }
        }));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine(_map["air1"]);
}

    大概如此

    static async Task Main(string[] args)
{
    var tasks = new List<Task>();
    _map.TryAdd("air1", 0);
    for (int i = 0; i < 10000; i++)
    {
        tasks.Add(Task.Run(() =>
        {
            while (!_map.TryAdd("air", 0)) { }
            for (int i = 0; i < 10; i++)
            {
                int value;
                while (!_map.TryRemove("air", out value)) { }
                value++;
                if (value >= 10)
                {
                    do
                    {
                        value = _map["air1"];
                    }
                    while (!_map.TryUpdate("air1", value + 10, value));
                }
                else
                    while (!_map.TryAdd("air", value)) { }
            }
        }));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine(_map["air1"]);
}

    这种迷你循环称为自旋锁。自旋锁的目的是在估计其持续时间很短并且迭代次数很少时确保线程安全。理想情况下，自旋锁的持续时间应使处理器上的负载在其余应用程序代码的总资源消耗的测量误差范围内，即不可见。

    这段代码运行大约一秒钟。也就是说，比从 0 到 100000 的循环要慢得多。但是 99% 的时间不是花在计算上，甚至不是花在自旋锁上，而是花在 10,000 个线程的上下文启动和切换上。

    这段代码很酷的一点是，线程可以处理彼此的值，即在调用字典之间，不同的线程可以执行计算，即相互交换数据，并且结果在 100% 的情况下都是正确的。

    事实上，你需要更加小心自旋锁，让我们来计算一下点击次数。

    static async Task Main(string[] args)
{
    var tasks = new List<Task>();
    int spinCount = 0;
    for (int i = 0; i < 10000; i++)
    {
        tasks.Add(Task.Run(() =>
        {
            while (!_map.TryAdd("air", 0)) { Interlocked.Increment(ref spinCount); }
            for (int i = 0; i < 10; i++)
            {
                int value;
                while (!_map.TryRemove("air", out value)) { Interlocked.Increment(ref spinCount); }
                value++;
                if (value >= 10)
                {
                    _map.AddOrUpdate("air1", 10, (k, v) => v += 10);
                }
                else
                    while (!_map.TryAdd("air", value)) { Interlocked.Increment(ref spinCount); }
            }
        }));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine(spinCount);
    Console.WriteLine(_map["air1"]);
}

    太多了

    4942023
100000

    我通过实验发现，如果在最后一个周期添加最小等待时间，情况就会改变。

    else
    while (!_map.TryAdd("air", value)) { Thread.Sleep(1); Interlocked.Increment(ref spinCount); }

    事实证明

    6141
100000

    这是完全不同的事情。这可以认为是一个稳定的解决方案。

    解决方案是通过 .NET 内置的框架SpinWait，它比循环更聪明一些

    private static ConcurrentDictionary<string, int> _map = [];

static async Task Main(string[] args)
{
    var tasks = new List<Task>();
    for (int i = 0; i < 10000; i++)
    {
        tasks.Add(Task.Run(() =>
        {
            SpinWait.SpinUntil(() => _map.TryAdd("air", 0));
            for (int i = 0; i < 10; i++)
            {
                int value = 0;
                SpinWait.SpinUntil(() => _map.TryRemove("air", out value));
                value++;
                if (value >= 10)
                {
                    _map.AddOrUpdate("air1", 10, (k, v) => v += 10);
                }
                else
                {
                    SpinWait.SpinUntil(() => _map.TryAdd("air", value));
                }
            }
        }));
    }

    await Task.WhenAll(tasks);
    Console.WriteLine(_map["air1"]);
}

    它的工作速度也非常快，有效地减少了碰撞的可能性。

    • 3

相关问题