C#多线程之高级篇（上）

前言

抛开死锁不谈，只聊性能问题，尽管锁总能粗暴的满足同步需求，但一旦存在竞争关系，意味着一定会有线程被阻塞，竞争越激烈，被阻塞的线程越多，上下文切换次数越多，调度成本越大，显然在高并发的场景下会损害性能。在高并发高性能且要求线程安全的述求下，无锁构造（非阻塞构造）闪亮登场。
参考文档：
C# - 理论与实践中的 C# 内存模型，第 2 部分 | Microsoft Docs
volatile 关键字 (C# 参考)
一、非阻塞同步

重排序与缓存

我们观察下面这个例子：

1. public class Foo
2. {
3.     private int _answer;
4.     private bool _complete;
6.     void A() //A 1
7.     {
8.         _answer = 10;
9.         _complete = true;
10.     }
12.     void B() //B 2
13.     {
14.         if (_complete) Console.WriteLine(_answer);
15.     }
16. }

复制代码

如果方法A和B在不同的线程上并发运行，B可能会打印 “ 0 “ 吗？答案是会的，原因如下：

• 编译器、CLR 或 CPU 可能会对代码/指令进行重排序（reorder）以提高效率。
  
• 编译器、CLR 或 CPU 可能会进行缓存优化，导致其它线程不能马上看到变量的新值。
  

请务必重视它们，它们将是幽灵般的存在

1. int x = 0, y = 0, a = 0, b = 0;
3. var task1 = Task.Run(() => // A 1
4. {
5.     a = 1; // 1
6.     x = b; // 2
7. });
8. var task2 = Task.Run(() => // B 2
9. {
10.     b = 2; // 3
11.     y = a; // 4
12. });
13. Task.WaitAll(task1, task2);
14. Console.WriteLine("x:" + x + " y:" + y);

复制代码

直觉和经验告诉我们，程序至顶向下执行：代码1一定发生在代码2之前，代码3一定发生在代码4之前，然鹅
在一个独立的线程中，每一个语句的执行顺序是可以被保证的，但在不使用lock，waithandle这样的显式同步操作时，我们就没法保证事件在不同的线程中看到的执行顺序是一致的了。尽管线程A中一定需要观察到a=1执行成功之后才会去执行x=b，但它没法确保自己观察得到线程B中对b的写入，所以A还可能会打印出y的一个旧版的值。这就叫指令重排序。

1. x:0 y:1 #1-2-3-4
2. x:2 y:0 #3-4-1-2
3. x:2 y:1 #1-3-2-4

复制代码

可实际运行时还是有些让我们惊讶的情况：

1. x:0 y:0 #??

复制代码

这就是缓存问题，如果两个线程在不同的CPU上执行，每一个核心有自己的缓存，这样一个线程的写入对于其它线程，在主存同步之前就是不可见的了。

C#编译器和CLR运行时会非常小心的保证上述优化不会破坏普通的单线程代码，和正确使用锁的多线程代码。但有时，你仍然需要通过显示的创建内存屏障（memory barrier，也称作内存栅栏 （memory fence））来对抗这些优化，限制指令重排序和读写缓存产生的影响。

内存屏障

参考博客小林野夫
处理器支持哪种内存重排序（LoadLoad重排序、LoadStore重排序、StoreStore重排序、StoreLoad重排序），就会提供相对应能够禁止重排序的指令，而这些指令就被称之为内存屏障（LoadLoad屏障、LoadStore屏障、StoreStore屏障、StoreLoad屏障）
屏障名称示例具体作用StoreLoadStore1;Store2;Store3;StoreLoad;Load1;Load2;Load3禁止StoreLoad重排序，确保屏障之前任何一个写（如Store2）的结果都会在屏障后任意一个读操作（如Load1）加载之前被写入StoreStoreStore1;Store2;Store3;StoreStore;Store4;Store5;Store6禁止StoreStore重排序，确保屏障之前任何一个写（如Store1）的结果都会在屏障后任意一个写操作（如Store4）之前被写入LoadLoadLoad1;Load2;Load3;LoadLoad;Load4;Load5;Load6禁止LoadLoad重排序，确保屏障之前任何一个读（如Load1）的数据都会在屏障后任意一个读操作（如Load4）之前被加载LoadStoreLoad1;Load2;Load3;LoadStore;Store1;Store2;Store3禁止LoadStore重排序，确保屏障之前任何一个读（如Load1）的数据都会在屏障后任意一个写操作（如Store1）的结果被写入高速缓存（或主内存）前被加载读屏障告诉处理器在执行任何的加载前，执行所有已经在失效队列（Invalidte Queues）中的失效（I）指令。即：所有load barrier之前的store指令对之后（本核心和其他核心）的指令都是可见的。
Store Memory Barrier：写屏障，等同于前文的StoreStore Barriers 将store buffer都写入缓存。
写屏障告诉处理器在执行这之后的指令之前，执行所有已经在存储缓存（store buffer）中的修改（M）指令。即：所有store barrier之前的修改（M）指令都是对之后的指令可见。
最简单的内存屏障是完全内存屏障（full memory barrier，或全栅栏（full fence）），它可以阻止所有跨越栅栏的指令进行重排并提交修改和刷新缓存。内存屏障之前的所有写操作都要写入内存，并将内存中的新值刷到缓存，使得其它CPU核心能够读取到最新值，完全保证了数据的强一致性，进而解决CPU缓存带来的可见性问题。
我们简单修改一下前面的案例

1. void A()
2. {
3.     _answer = 10;
4.     Thread.MemoryBarrier(); // 1
5.     _complete = true;
6.     Thread.MemoryBarrier(); // 3
7. }
8. void B()
9. {
10.     Thread.MemoryBarrier(); // 2
11.     if (_complete)
12.     {
13.         _testOutputHelper.WriteLine(_answer.ToString());
14.     }
15. }

复制代码

屏障1,3使得这个例子不可能打印出0，屏障2保证如果B在A之后执行，_complete一定读到的是true
内存屏障离我们并不遥远，以下方式都会隐式的使用全栅栏：

• lock语法糖或Monitor.Enter / Monitor.Exit
  
• Interlocked类中的所有方法
  
• 使用线程池的异步回调，包括异步委托，APM回调，以及任务延续（task continuations）
  
• 信号构造的等待/复位
  
• 任何依赖信号同步的情况，比如启动或等待Task，因此下面的代码也是线程安全的
  1. int x = 0;
  2. Task t = Task.Factory.StartNew (() => x++);
  3. t.Wait();
  4. Console.WriteLine (x);    // 1

  复制代码

volatile

另一个（更高级的）解决这个问题的方法是对_complete字段使用volatile关键字。

1. volatile bool _complete;

复制代码

volatile关键字通知编译器在每个读这个字段的地方使用一个读栅栏（acquire-fence），并且在每个写这个字段的地方使用一个写栅栏（release-fence）。
这种“半栅栏（half-fences）”比全栅栏更快，因为它给了运行时和硬件更大的优化空间。
读栅栏：也就是读屏障（Store Memory Barrier），等同于前文的LoadLoad Barriers 将Invalidate的 都执行完成。告诉处理器在执行任何的加载前，执行所有已经在失效队列（Invalidte Queues）中的失效（I）指令。即：所有load barrier之前的store指令对之后（本核心和其他核心）的指令都是可见的。
写栅栏：也就是写屏障（Store Memory Barrier），等同于前文的StoreStore Barriers 将store buffer都写入主存。
告诉处理器在执行这之后的指令之前，执行所有已经在存储缓存（store buffer）中的修改（M）指令。即：所有store barrier之前的修改（M）指令都是对之后的指令可见。

巧的是，Intel 的 X86 和 X64 处理器总是在读时使用读栅栏，写时使用写栅栏，无论是否使用volatile关键字。所以在使用这些处理器的情况下，这个关键字对硬件来说是无效的。然而，volatile关键字对编译器和 CLR 进行的优化是有作用的，以及在 64 位 AMD 和 Itanium 处理器上也是有作用的。这意味着不能因为你的客户端运行在特定类型的 CPU 上而放松警惕。

注意：使用volatile不能阻止写-读被交换
第一条指令第二条指令是否会被交换读读不会读写不会写写不会（CLR 确保写-写操作永远不会被交换，就算是没有volatile关键字）写读会！在下面案例中仍然有可能会打印00的情况（对a的读取可能发生在写入前--重排序）

1. int a = 0, b = 0;
2. int x = 0, y = 0;
3. var task1 = Task.Run(() =>
4. {
5.     Thread.VolatileWrite(ref a, 1);
6.     x = Thread.VolatileRead(ref b);
7. });
8. var task2 = Task.Run(() =>
9. {
10.     Thread.VolatileWrite(ref b, 2);
11.     y = Thread.VolatileRead(ref a);
12. });
13. Task.WaitAll(task1, task2);
15. Console.WriteLine("x:" + x + " y:" + y);

复制代码

volatile关键字不能应用于数组元素，不能用在捕获的局部变量：这些情况下你必须使用VolatileRead和VolatileWrite方法
从上面的例子我们可以看出，写-读操作可能被重新排序，官方的解释是：

在多处理器系统上，易失性读取操作不保证获取由任何处理器写入该内存位置的最新值。 同样，易失性写入操作不保证写入的值会立即对其他处理器可见。
（我的理解是：volatile关键字只能解决重排序问题，解决不了多处理器的缓存一致性问题）

注意double 和 long无法标记为 volatile，因为对这些类型的字段的读取和写入不能保证是原子的。 若要保护对这些类型字段的多线程访问，请使用 Interlocked 类成员或使用 lock 语句保护访问权限。
Interlocked

位于System.Threading，为多个线程共享的变量提供原子操作，这也是DOTNET为数不多的线程安全类型之一。
Interlocked通过将原子性的需求传达给操作系统和CLR来进行实现其功能，此类的成员不会引发异常。
可以防止 1.线程上下文切换，2.线程更新可由其他线程访问的变量时，或者当两个线程同时在不同的处理器上执行时  可能会出现的错误。
场景：

1. int i = 0;
2. i ++;

复制代码

在大多数计算机上，自增并不是原子操作，需要以下步骤：

• 将变量i的值加载到寄存器中。
  
• 计算i + 1。
  
• 将上面的计算结果存储在变量i中。
  

假设A线程执行完1-2时被抢占，B线程执行1-2-3，当A线程恢复时继续执行3，此时B线程的值就被覆盖掉了。
使用Increment即可解决，123会被打包成一个操作，以原子的方式实现自增
CAS

定义(摘自百度百科)：
CAS 操作包含三个操作数 —— 内存位置（V）、预期原值（A）和新值(B)。如果内存位置的值与预期原值相匹配，那么处理器会自动将该位置值更新为新值。否则，处理器不做任何操作。无论哪种情况，它都会在 CAS 指令之前返回该位置的值。（在 CAS 的一些特殊情况下将仅返回 CAS 是否成功，而不提取当前值。）CAS 有效地说明了“我认为位置 V 应该包含值 A；如果包含该值，则将 B 放到这个位置；否则，不要更改该位置，只告诉我这个位置的值即可。”
Interlocked.CompareExchange，实现了CAS：比较两个值是否相等，如果相等，则替换第一个值，否则什么都不做，最终返回这个位置的原始值。

1. Interlocked.CompareExchange(ref _num, 1000, 500);

复制代码

CAS在保证原子性读写的同时，没有加锁，保障了程序并发度，但也存在缺陷：

• ABA问题
  
• 只能保证一个地址的读写原子性
  
• 自旋CAS时间过长，容易给CPU带来大开销
  

二、延迟初始化

面试时候经常问：单例模式中的懒汉模式线程安全问题
场景：某个字段构造开销非常大，使得在初始化A时需要承担初始化Expensive的开销，即使Expensive字段不会被用到。

1. public class A
2. {
3.     public readonly Expensive Expensive = new Expensive();
4.     // ..
5. }
7. public class Expensive
8. {
9.     // 构造开销非常昂贵
10. }

复制代码

自然会想到懒汉模式：按需加载

1. public class B
2. {
3.     private Expensive _expensive;
5.     public Expensive GetExpensiveInstance()
6.     {
7.         if (_expensive == null) _expensive = new Expensive();
9.         return _expensive;
10.     }
11. }

复制代码

新的问题产生：GetExpensiveInstance是线程安全的吗？我们可以通过加锁解决

1. public class C
2. {
3.     private readonly object _locker = new object();
4.     private Expensive _expensive;
6.     public Expensive GetExpensiveInstance()
7.     {
8.         lock (_locker)
9.         {
10.             if (_expensive == null) _expensive = new Expensive();
11.             return _expensive;
12.         }
13.     }
14. }

复制代码

现在面试官继续问：还有性能更好的版本吗？..
Lazy

net standard1.0 提供System.Lazy来帮助你以线程安全且高效的方式（DCL）解决延迟初始化问题，只需

1. public class D
2. {
3.     private Lazy<Expensive> _expensive = new Lazy<Expensive>(() => new Expensive(), true);
5.     public Expensive GetExpensiveInstance() => _expensive.Value;
6. }

复制代码

第一个参数是一个委托，告知如何构建，第二个参数是boolean类型，传false实现的就是上面提到的plain B非线程安全迟初始化
双检锁 double checked locking会进行一次额外的易失读（volatile read），在对象已经完成初始化时，能够避免获取锁产生的开销。

1. public class E
2. {
3.     private readonly object _locker = new object();
4.     private volatile Expensive _expensive;
6.     public Expensive GetExpensiveInstance()
7.     {
8.         // 额外的易失读（volatile read）
9.         if (_expensive == null)
10.         {
11.             lock (_locker)
12.             {
13.                 if (_expensive == null) _expensive = new Expensive();
14.             }
15.         }
16.         
17.         return _expensive;
18.     }
19. }

复制代码

LazyInitializer

LazyInitializer是一个静态类，提供EnsureInitialized方法，第一个参数是需要构造的变量地址，第二个参数是一个委托，告知如何构造

1. public class F
2. {
3.     private Expensive _expensive;
5.     public Expensive GetExpensiveInstance()
6.     {
7.         LazyInitializer.EnsureInitialized(ref _expensive,
8.             () => new Expensive());
9.         return _expensive;
10.     }
11. }

复制代码

它使用竞争初始化模式的实现，比双检锁更快（在多核心情况下），因为它的实现完全不使用锁。这是一个很少需要用到的极端优化，并且会带来以下代价：

• 当参与初始化的线程数大于核心数时，它会更慢。
  
• 可能会因为进行了多余的初始化而浪费 CPU 资源。
  
• 初始化逻辑必须是线程安全的（例如，Expensive的构造器对静态字段进行写，就不是线程安全的）。
  
• 如果初始化的对象是需要进行销毁的，多余的对象需要额外的逻辑才能被销毁。
  

竞争初始化（race-to-initialize）模式，通过易失性和CAS，实现无锁构造

1. public class G
2. {
3.     private volatile Expensive _expensive;
4.     public Expensive Expensive
5.     {
6.         get
7.         {
8.             if (_expensive == null)
9.             {
10.                 var instance = new Expensive();
11.                 Interlocked.CompareExchange (ref _expensive, instance, null);
12.             }
13.             return _expensive;
14.         }
15.     }
16. }

复制代码

三、线程局部存储

我们花费了大量篇幅来讲并发访问公共数据问题，前文提到的锁构造，信号构造，无锁构造本质上都是使用同步构造，使得多线程在访问公共数据时能安全的进行，然而有时我们会希望数据在线程间是隔离的，局部变量就能实现这个目的，但他们的生命周期总是那么短暂（随代码块而释放），我们期待更大作用域的隔离数据，线程局部变量（thread-local storage，TLS）就可以实现这个目的。
ThreadStatic

被ThreadStatic标记的static字段不会在线程间共享，每个执行线程都有一个单独的字段实例
Note：

• 被标记的必须是static字段，不能在实例字段上使用（添加了也无效）
  
• 请不要给被标记的字段指定初始值，因为这种初始化只会在类被构造时执行一次，影响一个线程，因此他依赖零值
  

如果你需要使用实例字段，或者非零值，请使用ThreadLocal

1. public class ThreadStatic测试
2. {
3.     private readonly ITestOutputHelper _testOutputHelper;
4.     [ThreadStatic] private static int _num;
6.     public ThreadStatic测试(ITestOutputHelper testOutputHelper)
7.     {
8.         _testOutputHelper = testOutputHelper;
9.     }
11.     [Fact]
12.     void Show()
13.     {
14.         void Work()
15.         {
16.             for (int i = 0; i < 100000; i++)
17.             {
18.                 _num++;
19.                 _testOutputHelper.WriteLine(_num.ToString());
20.             }
21.         }
23.         var t1 = new Thread(Work);
24.         var t2 = new Thread(Work);
26.         t1.Start();
27.         t2.Start();
28.         t1.Join();
29.         t2.Join();
31.         _testOutputHelper.WriteLine(_num.ToString());
32.     }
33. }

复制代码

输出：

1. 100000
2. 100000
3. 0

复制代码

LocalDataStoreSlot

封装内存槽以存储本地数据。 此类不能被继承。.NET Framework 1.1加入，但在standard2.0+才有。

1. public sealed class LocalDataStoreSlot

复制代码

.NET Framework 提供了两种机制，用于使用线程本地存储 (TLS) ：LocalDataStoreSlot和ThreadStaticAttribute
LocalDataStoreSlot比ThreadStaticAttribute更慢，更尴尬。此外，数据存储为类型 Object，因此必须先将其强制转换为正确的类型，然后再使用它。
有关使用 TLS 的详细信息，请参阅 线程本地存储。
同样，.NET Framework 提供了两种使用上下文本地存储的机制：LocalDataStoreSlot和ContextStaticAttribute。 上下文相对静态字段是用属性标记的 ContextStaticAttribute 静态字段。 请参考注解

1. // 同一个 LocalDataStoreSlot 对象可以跨线程使用。
2. LocalDataStoreSlot _slot = Thread.AllocateNamedDataSlot("mySlot");
3. void Work()
4. {
5.     for (int i = 0; i < 100000; i++)
6.     {
7.         int num = (int)(Thread.GetData(_slot)??0);
8.         Thread.SetData(_slot, num + 1);
9.     }
10.     _testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());
11. }
12. var t1 = new Thread(Work);
13. var t2 = new Thread(Work);
14. t1.Start();
15. t2.Start();
16. t1.Join();
17. t2.Join();
18. _testOutputHelper.WriteLine(((int)(Thread.GetData(_slot)??0)).ToString());

复制代码

输出效果和ThreadStaticAttribute一样：

1. 100000
2. 100000
3. 0

复制代码

使用Thread.FreeNamedDataSlot("mySlot");可以释放所有线程上的指定槽，但是只有在所有对该槽的引用都出了其作用域，并且被垃圾回收后才会真正释放。这确保了只要保持对槽的引用，就能继续使用槽。
你也可以通过Thread.AllocateDataSlot()来创建一个无名槽位，与命名槽的区别是无名槽需要自行控制作用域
当然我们也可以对上面复杂的᠍᠍᠍᠍᠍Thread.GetData,Thread.SetData进行封装

1. LocalDataStoreSlot _secSlot = Thread.GetNamedDataSlot ("securityLevel");
2. int Num
3. {
4.     get
5.     {
6.         object data = Thread.GetData(_secSlot);
7.         return data == null ? 0 : (int) data;    // null 相当于未初始化。
8.     }
9.     set { Thread.SetData (_secSlot, value); }
10. }

复制代码

ThreadLocal

ThreadLocal是 Framework 4.0 加入的，涵盖在netstandard1.0。它提供了可用于静态字段和实例字段的线程局部存储，并且允许设置默认值。

1. public class ThreadLocal测试
2. {
3.     ThreadLocal<int> _num = new ThreadLocal<int> (() => 3);
4.     private readonly ITestOutputHelper _testOutputHelper;
7.     public ThreadLocal测试(ITestOutputHelper testOutputHelper)
8.     {
9.         _testOutputHelper = testOutputHelper;
10.     }
12.     [Fact]
13.     void Show()
14.     {
15.         void Work()
16.         {
17.             for (int i = 0; i < 100000; i++)
18.             {
19.                 _num.Value++;
20.             }
21.             _testOutputHelper.WriteLine(_num.ToString());
22.         }
24.         var t1 = new Thread(Work);
25.         var t2 = new Thread(Work);
27.         t1.Start();
28.         t2.Start();
29.         t1.Join();
30.         t2.Join();
32.         _testOutputHelper.WriteLine(_num.ToString());
33.     }
34. }

复制代码

输出

1. 100003
2. 100003
3. 3

复制代码

下面这个测试非常有意思

1. [Fact]
2. void Show()
3. {
4.     var threadName = new ThreadLocal<string>(() => "Thread" + Thread.CurrentThread.ManagedThreadId);
5.     Parallel.For(0, 13, x =>
6.     {
7.         bool repeat = threadName.IsValueCreated;
8.         _testOutputHelper.WriteLine($"ThreadName = {threadName.Value} {(repeat ? "(repeat)" : "")}");
9.     });
10.    
11.     threadName.Dispose();  // 释放资源
12. }

复制代码

你会发现当Parallel.For第二个参数超过你的逻辑内核后，repeat出现了！

1. ThreadName = Thread5
2. ThreadName = Thread8
3. ThreadName = Thread31
4. ThreadName = Thread29
5. ThreadName = Thread31 (repeat)
6. ThreadName = Thread30
7. ThreadName = Thread18
8. ThreadName = Thread12
9. ThreadName = Thread32
10. ThreadName = Thread28
11. ThreadName = Thread33
12. ThreadName = Thread35
13. ThreadName = Thread34

复制代码

Random类不是线程安全的，所以我们要不然在使用Random时加锁（这样限制了并发），如今我们有了ThreadLocal：

1. var localRandom = new ThreadLocal<Random>(() => new Random());

复制代码

很轻易的就解决了线程安全问题，但是上面的版本使用的Random的无参构造方法，会依赖系统时间作为生成随机数的种子，在大概 10ms 时间内创建的两个Random对象可能会使用相同的种子，下边是解决这个问题的一个办法：

1. var localRandom = new ThreadLocal<Random>(() => new Random (Guid.NewGuid().GetHashCode()) );

复制代码

特别注意，不要以为GUID全局唯一，GUID的HashCode也全局唯一，上面的随机数仍然不是真随机

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！