C# 线程安全详解

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

介绍

在 .NET4.0 之前，如果我们需要在多线程环境下使用 Dictionary 类，除了自己实现线程同步来保证线程安全外，我们没有其他选择。很多开发人员肯定都实现过类似的线程安全方案，可能是通过创建全新的线程安全字典，或者仅是简单的用一个类封装一个 Dictionary 对象，并在所有方法中加上锁机制，我们称这种方案叫 “Dictionary+Locks” 。

System.Collections.Concurrent 命名空间下提供多个线程安全集合类，只要多个线程同时访问集合，就应使用这些类来代替 System.Collections 和 System.Collections.Generic 命名空间中的相应类型。但是，不保证通过扩展方法或通过显式接口实现访问集合对象是线程安全的，可能需要由调用方进行同步。

经典生产消费问题

介绍

这个问题是最为经典的多线程应用问题问题就是：有一个或多个线程（生产者线程）产生一些数据，还有一个或者多个线程（消费者线程）要取出这些数据并执行一些相应的工作。

在这里插入图片描述

Queue

接下来，我们是使用程序去描述这个问题，看下面代码


static void Main(string[] args)
{
    int count = 0;
    // 临界资源区
    var queue = new Queue<string>();
    // 生产者线程
    Task.Factory.StartNew(() =>
    {
        while (true)
        {
            queue.Enqueue("mesg" + count);
            count++;
        }
    });
    // 消费者线程1
    Task.Factory.StartNew(() =>
    {
        while (true)
        {
            if (queue.Count > 0)
            {
                string value = queue.Dequeue();
                Console.WriteLine("Worker A: " + value);
            }
        }
    });
    // 消费者线程2
    Task.Factory.StartNew(() =>
    {
        while (true)
        {
            if (queue.Count > 0)
            {
                string value = queue.Dequeue();
                Console.WriteLine("Worker B: " + value);
            }
        }
    });
    Thread.Sleep(50000);
}

我们使用 Queue 模拟了一个简单的资源池，一个生产者放数据，两个消费者消费数据。

这个程序运行以后会产生异常，异常的原因很简单。当某时刻，第一个消费者判断 queue.Count > 0 为true 时，就会到 Queue 中取数据。但是，此时这个数据可能会被第二个消费者拿走了，因为第二个消费者也判断出此时有数据可取。第一个消费者取取数据时就会发生异常，这就是一个简单的临界资源线程安全问题。

知道问题了，那么如何解决呢？有两种方案，接下来进行讲解

ConcurrentQueue

1 . 加锁

这个方案是可行的，很多时候我们也是这么做的，包括微软早期实现线程安全的 ArrayList 和 Hashtable 内部 (Synchronized方法) 也是这么实现的。这个方案适用于只有少量的消费者，并且每个消费者都会执行大量操作的时候，这时 lock 并没什么太大问题，但是，如果是大批量短小精悍的消费者存在的话，lock 会严重影响代码的执行效率。

2 . 线程安全的集合区

这个就是 .NET4.0 后 System.Collections.Concurrent 命名空间下提供多个线程安全集合类方案。

新的线程安全的这些集合内部不再使用lock机制这种比较低效的方式去实现线程安全，而是转而使用SpinWait 和 Interlocked 等机制，间接实现了线程安全，这种方式的效率要高于使用lock的方式。


var queue = new ConcurrentQueue<string>();
Task.Factory.StartNew(() =>
{
    while (true)
    {
        queue.Enqueue("msg" + count);
        count++;
    }
});
Task.Factory.StartNew(() =>
{
    while (true)
    {
        string value;
        if (queue.TryDequeue(out value))
        {
            Console.WriteLine("Worker A: " + value);
        }
    }
});
Task.Factory.StartNew(() =>
{
    while (true)
    {
        string value;
        if (queue.TryDequeue(out value))
        {
            Console.WriteLine("Worker B: " + value);
        }
    }
});

ConcurrentQueue.TryDequeue(T) 方法会尝试获取消费，那能不能不要去判断集合是否为空，集合当自己没有元素的时候自己 Block 一下可以吗？答案是，可以的

BlockingCollection

针对上面的问题，我们可以使用 BlockingCollection 即可。接下来我来看


var blockingCollection = new BlockingCollection<string>();
Task.Factory.StartNew(() =>
{
    while (true)
    {
        blockingCollection.Add("msg" + count);
        count++;
    }
});
Task.Factory.StartNew(() =>
{
    while (true)
    {
        Console.WriteLine("Worker A: " + blockingCollection.Take());
    }
});
Task.Factory.StartNew(() =>
{
    while (true)
    {
        Console.WriteLine("Worker B: " + blockingCollection.Take());
    }
});

BlockingCollection 集合是一个拥有阻塞功能的集合，它就是完成了经典生产者消费者的算法功能。它没有实现底层的存储结构，而是使用了实现 IProducerConsumerCollection 接口的几个集合作为底层的数据结构，例如 ConcurrentBag， ConcurrentStack 或者是 ConcurrentQueue。你可以在构造BlockingCollection 实例的时候传入这个参数，如果不指定的话，则默认使用 ConcurrentQueue 作为存储结构。

而对于生产者来说，只需要通过调用其Add方法放数据，消费者只需要调用Take方法来取数据就可以了。

当然了上面的消费者代码中还有一点是让人不爽的，那就是 while 语句，可以更优雅一点吗？答案是，可以的。


Task.Factory.StartNew(() =>
{
        foreach (string value in blockingCollection.GetConsumingEnumerable())
        {
            Console.WriteLine("Worker A: " + value);
        }
});

BlockingCollection.GetConsumingEnumerable 方法是关键，这个方法会遍历集合取出数据，一旦发现集合空了，则阻塞自己，直到集合中又有元素了再开始遍历。

此时，完美了解决了生产者消费者问题。然而通常来说，还有下面两个问题我们有时需要去控制

1 . 控制集合中数据的最大数量

这个问题由 BlockingCollection 构造函数解决，构造该对象实例的时候，构造函数中的 BoundedCapacity 决定了集合最大的可容纳数据数量，这个比较简单。

2 . 何时停止的问题

这个问题由 CompleteAdding 和 IsCompleted 两个配合解决。CompleteAdding 方法是直接不允许任何元素被加入集合；当使用了 CompleteAdding 方法后且集合内没有元素的时候，另一个属性 IsCompleted 此时会为 True，这个属性可以用来判断是否当前集合内的所有元素都被处理完。生产者修改后的代码：


Task.Factory.StartNew(() =>
{
    for (int count = 0; count < 10; count++)
    {
        blockingCollection.Add("msg" + count);
    }
    blockingCollection.CompleteAdding();
});

当使用了 CompleteAdding 方法后，对象停止往集合中添加数据，这时如果是使用 GetConsumingEnumerable 枚举的，那么这种枚举会自然结束，不会再 Block 住集合，这种方式最优雅，也是推荐的写法。

但是如果是使用 TryTake 访问元素的，则需要使用 IsCompleted 判断一下，因为这个时候使用 TryTake 会抛InvalidOperationException 异常。接着我们看下最后的完整代码：


static void Main(string[] args)
{
    var blockingCollection = new BlockingCollection<string>();
    var producer = Task.Factory.StartNew(() =>
    {
        for (int count = 0; count < 10; count++)
        {
            blockingCollection.Add("msg" + count);
            Thread.Sleep(300);
        }
        blockingCollection.CompleteAdding();
    });
    var consumer1 = Task.Factory.StartNew(() =>
    {
        foreach (string value in blockingCollection.GetConsumingEnumerable())
        {
            Console.WriteLine("Worker A: " + value);
        }
    });
    var consumer2 = Task.Factory.StartNew(() =>
    {
        foreach (string value in blockingCollection.GetConsumingEnumerable())
        {
            Console.WriteLine("Worker B: " + value);
        }
    });
    Task.WaitAll(producer, consumer1, consumer2);
}

BlockingCollection 枚举

此外，需要注意 BlockingCollection 有两种枚举方法，

1 . foreach

首先 BlockingCollection 本身继承自IEnumerable，所以它自己就可以被 foreach 枚举，首先 BlockingCollection 包装了一个线程安全集合，那么它自己也是线程安全的，而当多个线程在同时修改或访问线程安全容器时，BlockingCollection 自己作为 IEnumerable 会返回一个一定时间内的集合片段，也就是只会枚举在那个时间点上内部集合的元素。使用这种方式枚举的时候，不会有 Block 效果。

2 . GetConsumingEnumerable

另外一种方式就是我们上面使用的 GetConsumingEnumerable 方式的枚举，这种方式会有 Block 效果，直到 CompleteAdding 被调用为止。

BlockingCollection 扩展

实现 IProducerConsumerCollection 接口的几个集合：ConcurrentBag (线程安全的无序的元素集合)， ConcurrentStack (线程安全的堆栈) 和 ConcurrentQueue (线程安全的队列)。这些都很简单，功能与非线程安全的那些集合都一样，只不过是多了 TryXXX 方法，多线程环境下使用这些方法就好了。