C#多线程编程系列(三)- 线程同步
目录
- 1.1 简介
- 1.2 执行基本原子操作
- 1.3 使用Mutex类
- 1.4 使用SemaphoreSlim类
- 1.5 使用AutoResetEvent类
- 1.6 使用ManualResetEventSlim类
- 1.7 使用CountDownEvent类
- 1.8 使用Barrier类
- 1.9 使用ReaderWriterLockSlim类
- 1.10 使用SpinWait类
- 参考书籍
- 笔者水平有限,如果错误欢迎各位批评指正!
1.1 简介
本章介绍在C#中实现线程同步的几种方法。因为多个线程同时访问共享数据时,可能会造成共享数据的损坏,从而导致与预期的结果不相符。为了解决这个问题,所以需要用到线程同步,也被俗称为“加锁”。但是加锁绝对不对提高性能,最多也就是不增不减,要实现性能不增不减还得靠高质量的同步源语(Synchronization Primitive)。但是因为正确永远比速度更重要,所以线程同步在某些场景下是必须的。
线程同步有两种源语(Primitive)构造:用户模式(user - mode)和内核模式(kernel - mode),当资源可用时间短的情况下,用户模式要优于内核模式,但是如果长时间不能获得资源,或者说长时间处于“自旋”,那么内核模式是相对来说好的选择。
但是我们希望兼具用户模式和内核模式的优点,我们把它称为混合构造(hybrid construct),它兼具了两种模式的优点。
在C#中有多种线程同步的机制,通常可以按照以下顺序进行选择。
- 如果代码能通过优化可以不进行同步,那么就不要做同步。
- 使用原子性的
Interlocked
方法。- 使用
lock/Monitor
类。- 使用异步锁,如
SemaphoreSlim.WaitAsync()
。- 使用其它加锁机制,如
ReaderWriterLockSlim、Mutex、Semaphore
等。- 如果系统提供了
*Slim
版本的异步对象,那么请选用它,因为*Slim
版本全部都是混合锁,在进入内核模式前实现了某种形式的自旋。
在同步中,一定要注意避免死锁的发生,死锁的发生必须满足以下4个基本条件,所以只需要破坏任意一个条件,就可避免发生死锁。
- 排他或互斥(Mutual exclusion):一个线程(ThreadA)独占一个资源,没有其它线程(ThreadB)能获取相同的资源。
- 占有并等待(Hold and wait):互斥的一个线程(ThreadA)请求获取另一个线程(ThreadB)占有的资源.
- 不可抢先(No preemption):一个线程(ThreadA)占有资源不能被强制拿走(只能等待ThreadA主动释放它的资源)。
- 循环等待条件(Circular wait condition):两个或多个线程构成一个循环等待链,它们锁定两个或多个相同的资源,每个线程都在等待链中的下一个线程占有的资源。
1.2 执行基本原子操作
CLR保证了对这些数据类型的读写是原子性的:Boolean、Char、(S)Byte、(U)Int16、(U)Int32、(U)IntPtr和Single
。但是如果读写Int64
可能会发生读取撕裂(torn read)的问题,因为在32位操作系统中,它需要执行两次Mov
操作,无法在一个时间内执行完成。
那么在本节中,就会着重的介绍System.Threading.Interlocked
类提供的方法,Interlocked
类中的每个方法都是执行一次的读取以及写入操作。更多与Interlocked
类相关的资料请参考链接,戳一戳本文不在赘述。
演示代码如下所示,分别使用了三种方式进行计数:错误计数方式、lock
锁方式和Interlocked
原子方式。
private static void Main(string[] args)
{
Console.WriteLine("错误的计数");
var c = new Counter();
Execute(c);
Console.WriteLine("--------------------------");
Console.WriteLine("正确的计数 - 有锁");
var c2 = new CounterWithLock();
Execute(c2);
Console.WriteLine("--------------------------");
Console.WriteLine("正确的计数 - 无锁");
var c3 = new CounterNoLock();
Execute(c3);
Console.ReadLine();
}
static void Execute(CounterBase c)
{
// 统计耗时
var sw = new Stopwatch();
sw.Start();
var t1 = new Thread(() => TestCounter(c));
var t2 = new Thread(() => TestCounter(c));
var t3 = new Thread(() => TestCounter(c));
t1.Start();
t2.Start();
t3.Start();
t1.Join();
t2.Join();
t3.Join();
sw.Stop();
Console.WriteLine($"Total count: {c.Count} Time:{sw.ElapsedMilliseconds} ms");
}
static void TestCounter(CounterBase c)
{
for (int i = 0; i < 100000; i++)
{
c.Increment();
c.Decrement();
}
}
class Counter : CounterBase
{
public override void Increment()
{
_count++;
}
public override void Decrement()
{
_count--;
}
}
class CounterNoLock : CounterBase
{
public override void Increment()
{
// 使用Interlocked执行原子操作
Interlocked.Increment(ref _count);
}
public override void Decrement()
{
Interlocked.Decrement(ref _count);
}
}
class CounterWithLock : CounterBase
{
private readonly object _syncRoot = new Object();
public override void Increment()
{
// 使用Lock关键字 锁定私有变量
lock (_syncRoot)
{
// 同步块
Count++;
}
}
public override void Decrement()
{
lock (_syncRoot)
{
Count--;
}
}
}
abstract class CounterBase
{
protected int _count;
public int Count
{
get
{
return _count;
}
set
{
_count = value;
}
}
public abstract void Increment();
public abstract void Decrement();
}
运行结果如下所示,与预期结果基本相符。
1.3 使用Mutex类
System.Threading.Mutex
在概念上和System.Threading.Monitor
几乎一样,但是Mutex
同步对文件或者其他跨进程的资源进行访问,也就是说Mutex
是可跨进程的。因为其特性,它的一个用途是限制应用程序不能同时运行多个实例。
Mutex
对象支持递归,也就是说同一个线程可多次获取同一个锁,这在后面演示代码中可观察到。由于Mutex
的基类System.Theading.WaitHandle
实现了IDisposable
接口,所以当不需要在使用它时要注意进行资源的释放。更多资料:戳一戳
演示代码如下所示,简单的演示了如何创建单实例的应用程序和Mutex
递归获取锁的实现。
const string MutexName = "CSharpThreadingCookbook";
static void Main(string[] args)
{
// 使用using 及时释放资源
using (var m = new Mutex(false, MutexName))
{
if (!m.WaitOne(TimeSpan.FromSeconds(5), false))
{
Console.WriteLine("已经有实例正在运行!");
}
else
{
Console.WriteLine("运行中...");
// 演示递归获取锁
Recursion();
Console.ReadLine();
m.ReleaseMutex();
}
}
Console.ReadLine();
}
static void Recursion()
{
using (var m = new Mutex(false, MutexName))
{
if (!m.WaitOne(TimeSpan.FromSeconds(2), false))
{
// 因为Mutex支持递归获取锁 所以永远不会执行到这里
Console.WriteLine("递归获取锁失败!");
}
else
{
Console.WriteLine("递归获取锁成功!");
}
}
}
运行结果如下图所示,打开了两个应用程序,因为使用Mutex
实现了单实例,所以第二个应用程序无法获取锁,就会显示已有实例正在运行。
1.4 使用SemaphoreSlim类
SemaphoreSlim
类与之前提到的同步类有锁不同,之前提到的同步类都是互斥的,也就是说只允许一个线程进行访问资源,而SemaphoreSlim
是可以允许多个访问。
在之前的部分有提到,以*Slim
结尾的线程同步类,都是工作在混合模式下的,也就是说开始它们都是在用户模式下"自旋",等发生第一次竞争时,才切换到内核模式。但是SemaphoreSlim
不同于Semaphore
类,它不支持系统信号量,所以它不能用于进程之间的同步。
该类使用比较简单,演示代码演示了6个线程竞争访问只允许4个线程同时访问的数据库,如下所示。
static void Main(string[] args)
{
// 创建6个线程 竞争访问AccessDatabase
for (int i = 1; i <= 6; i++)
{
string threadName = "线程 " + i;
// 越后面的线程,访问时间越久 方便查看效果
int secondsToWait = 2 + 2 * i;
var t = new Thread(() => AccessDatabase(threadName, secondsToWait));
t.Start();
}
Console.ReadLine();
}
// 同时允许4个线程访问
static SemaphoreSlim _semaphore = new SemaphoreSlim(4);
static void AccessDatabase(string name, int seconds)
{
Console.WriteLine($"{name} 等待访问数据库.... {DateTime.Now.ToString("HH:mm:ss.ffff")}");
// 等待获取锁 进入临界区
_semaphore.Wait();
Console.WriteLine($"{name} 已获取对数据库的访问权限 {DateTime.Now.ToString("HH:mm:ss.ffff")}");
// Do something
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine($"{name} 访问完成... {DateTime.Now.ToString("HH:mm:ss.ffff")}");
// 释放锁
_semaphore.Release();
}
运行结果如下所示,可见前4个线程马上就获取到了锁,进入了临界区,而另外两个线程在等待;等有锁被释放时,才能进入临界区。
1.5 使用AutoResetEvent类
AutoResetEvent
叫自动重置事件,虽然名称中有事件一词,但是重置事件和C#中的委托没有任何关系,这里的事件只是由内核维护的Boolean
变量,当事件为false
,那么在事件上等待的线程就阻塞;事件变为true
,那么阻塞解除。
在.Net中有两种此类事件,即AutoResetEvent(自动重置事件)
和ManualResetEvent(手动重置事件)
。这两者均是采用内核模式,它的区别在于当重置事件为true
时,自动重置事件它只唤醒一个阻塞的线程,会自动将事件重置回false,造成其它线程继续阻塞。而手动重置事件不会自动重置,必须通过代码手动重置回false。
因为以上的原因,所以在很多文章和书籍中不推荐使用AutoResetEvent(自动重置事件)
,因为它很容易在编写生产者线程时发生失误,造成它的迭代次数多余消费者线程。
演示代码如下所示,该代码演示了通过AutoResetEvent
实现两个线程的互相同步。
static void Main(string[] args)
{
var t = new Thread(() => Process(10));
t.Start();
Console.WriteLine("等待另一个线程完成工作!");
// 等待工作线程通知 主线程阻塞
_workerEvent.WaitOne();
Console.WriteLine("第一个操作已经完成!");
Console.WriteLine("在主线程上执行操作");
Thread.Sleep(TimeSpan.FromSeconds(5));
// 发送通知 工作线程继续运行
_mainEvent.Set();
Console.WriteLine("现在在第二个线程上运行第二个操作");
// 等待工作线程通知 主线程阻塞
_workerEvent.WaitOne();
Console.WriteLine("第二次操作完成!");
Console.ReadLine();
}
// 工作线程Event
private static AutoResetEvent _workerEvent = new AutoResetEvent(false);
// 主线程Event
private static AutoResetEvent _mainEvent = new AutoResetEvent(false);
static void Process(int seconds)
{
Console.WriteLine("开始长时间的工作...");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("工作完成!");
// 发送通知 主线程继续运行
_workerEvent.Set();
Console.WriteLine("等待主线程完成其它工作");
// 等待主线程通知 工作线程阻塞
_mainEvent.WaitOne();
Console.WriteLine("启动第二次操作...");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("工作完成!");
// 发送通知 主线程继续运行
_workerEvent.Set();
}
运行结果如下图所示,与预期结果符合。
1.6 使用ManualResetEventSlim类
ManualResetEventSlim
使用和ManualResetEvent
类基本一致,只是ManualResetEventSlim
工作在混合模式下,而它与AutoResetEventSlim
不同的地方就是需要手动重置事件,也就是调用Reset()
才能将事件重置为false
。
演示代码如下,形象的将ManualResetEventSlim
比喻成大门,当事件为true
时大门打开,线程解除阻塞;而事件为false
时大门关闭,线程阻塞。
static void Main(string[] args)
{
var t1 = new Thread(() => TravelThroughGates("Thread 1", 5));
var t2 = new Thread(() => TravelThroughGates("Thread 2", 6));
var t3 = new Thread(() => TravelThroughGates("Thread 3", 12));
t1.Start();
t2.Start();
t3.Start();
// 休眠6秒钟 只有Thread 1小于 6秒钟,所以事件重置时 Thread 1 肯定能进入大门 而 Thread 2 可能可以进入大门
Thread.Sleep(TimeSpan.FromSeconds(6));
Console.WriteLine($"大门现在打开了! 时间:{DateTime.Now.ToString("mm:ss.ffff")}");
_mainEvent.Set();
// 休眠2秒钟 此时 Thread 2 肯定可以进入大门
Thread.Sleep(TimeSpan.FromSeconds(2));
_mainEvent.Reset();
Console.WriteLine($"大门现在关闭了! 时间:{DateTime.Now.ToString("mm: ss.ffff")}");
// 休眠10秒钟 Thread 3 可以进入大门
Thread.Sleep(TimeSpan.FromSeconds(10));
Console.WriteLine($"大门现在第二次打开! 时间:{DateTime.Now.ToString("mm: ss.ffff")}");
_mainEvent.Set();
Thread.Sleep(TimeSpan.FromSeconds(2));
Console.WriteLine($"大门现在关闭了! 时间:{DateTime.Now.ToString("mm: ss.ffff")}");
_mainEvent.Reset();
Console.ReadLine();
}
static void TravelThroughGates(string threadName, int seconds)
{
Console.WriteLine($"{threadName} 进入睡眠 时间:{DateTime.Now.ToString("mm:ss.ffff")}");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine($"{threadName} 等待大门打开! 时间:{DateTime.Now.ToString("mm:ss.ffff")}");
_mainEvent.Wait();
Console.WriteLine($"{threadName} 进入大门! 时间:{DateTime.Now.ToString("mm:ss.ffff")}");
}
static ManualResetEventSlim _mainEvent = new ManualResetEventSlim(false);
运行结果如下,与预期结果相符。
1.7 使用CountDownEvent类
CountDownEvent
类内部构造使用了一个ManualResetEventSlim
对象。这个构造阻塞一个线程,直到它内部计数器(CurrentCount)
变为0
时,才解除阻塞。也就是说它并不是阻止对已经枯竭的资源池的访问,而是只有当计数为0
时才允许访问。
这里需要注意的是,当CurrentCount
变为0
时,那么它就不能被更改了。为0
以后,Wait()
方法的阻塞被解除。
演示代码如下所示,只有当Signal()
方法被调用2次以后,Wait()
方法的阻塞才被解除。
static void Main(string[] args)
{
Console.WriteLine($"开始两个操作 {DateTime.Now.ToString("mm:ss.ffff")}");
var t1 = new Thread(() => PerformOperation("操作 1 完成!", 4));
var t2 = new Thread(() => PerformOperation("操作 2 完成!", 8));
t1.Start();
t2.Start();
// 等待操作完成
_countdown.Wait();
Console.WriteLine($"所有操作都完成 {DateTime.Now.ToString("mm: ss.ffff")}");
_countdown.Dispose();
Console.ReadLine();
}
// 构造函数的参数为2 表示只有调用了两次 Signal方法 CurrentCount 为 0时 Wait的阻塞才解除
static CountdownEvent _countdown = new CountdownEvent(2);
static void PerformOperation(string message, int seconds)
{
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine($"{message} {DateTime.Now.ToString("mm:ss.ffff")}");
// CurrentCount 递减 1
_countdown.Signal();
}
运行结果如下图所示,可见只有当操作1和操作2都完成以后,才执行输出所有操作都完成。
1.8 使用Barrier类
Barrier
类用于解决一个非常稀有的问题,平时一般用不上。Barrier
类控制一系列线程进行阶段性的并行工作。
假设现在并行工作分为2个阶段,每个线程在完成它自己那部分阶段1的工作后,必须停下来等待其它线程完成阶段1的工作;等所有线程均完成阶段1工作后,每个线程又开始运行,完成阶段2工作,等待其它线程全部完成阶段2工作后,整个流程才结束。
演示代码如下所示,该代码演示了两个线程分阶段的完成工作。
static void Main(string[] args)
{
var t1 = new Thread(() => PlayMusic("钢琴家", "演奏一首令人惊叹的独奏曲", 5));
var t2 = new Thread(() => PlayMusic("歌手", "唱着他的歌", 2));
t1.Start();
t2.Start();
Console.ReadLine();
}
static Barrier _barrier = new Barrier(2,
Console.WriteLine($"第 {b.CurrentPhaseNumber + 1} 阶段结束"));
static void PlayMusic(string name, string message, int seconds)
{
for (int i = 1; i < 3; i++)
{
Console.WriteLine("----------------------------------------------");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine($"{name} 开始 {message}");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine($"{name} 结束 {message}");
_barrier.SignalAndWait();
}
}
运行结果如下所示,当“歌手”线程完成后,并没有马上结束,而是等待“钢琴家”线程结束,当"钢琴家"线程结束后,才开始第2阶段的工作。
1.9 使用ReaderWriterLockSlim类
ReaderWriterLockSlim
类主要是解决在某些场景下,读操作多于写操作而使用某些互斥锁当多个线程同时访问资源时,只有一个线程能访问,导致性能急剧下降。
如果所有线程都希望以只读的方式访问数据,就根本没有必要阻塞它们;如果一个线程希望修改数据,那么这个线程才需要独占访问,这就是ReaderWriterLockSlim
的典型应用场景。这个类就像下面这样来控制线程。
- 一个线程向数据写入是,请求访问的其他所有线程都被阻塞。
- 一个线程读取数据时,请求读取的线程允许读取,而请求写入的线程被阻塞。
- 写入线程结束后,要么解除一个写入线程的阻塞,使写入线程能向数据接入,要么解除所有读取线程的阻塞,使它们能并发读取数据。如果线程没有被阻塞,锁就可以进入*使用的状态,可供下一个读线程或写线程获取。
- 从数据读取的所有线程结束后,一个写线程被解除阻塞,使它能向数据写入。如果线程没有被阻塞,锁就可以进入*使用的状态,可供下一个读线程或写线程获取。
ReaderWriterLockSlim
还支持从读线程升级为写线程的操作,详情请戳一戳。文本不作介绍。ReaderWriterLock
类已经过时,而且存在许多问题,没有必要去使用。
示例代码如下所示,创建了3个读线程,2个写线程,读线程和写线程竞争获取锁。
static void Main(string[] args)
{
// 创建3个 读线程
new Thread(() => Read("Reader 1")) { IsBackground = true }.Start();
new Thread(() => Read("Reader 2")) { IsBackground = true }.Start();
new Thread(() => Read("Reader 3")) { IsBackground = true }.Start();
// 创建两个写线程
new Thread(() => Write("Writer 1")) { IsBackground = true }.Start();
new Thread(() => Write("Writer 2")) { IsBackground = true }.Start();
// 使程序运行30S
Thread.Sleep(TimeSpan.FromSeconds(30));
Console.ReadLine();
}
static ReaderWriterLockSlim _rw = new ReaderWriterLockSlim();
static Dictionary<int, int> _items = new Dictionary<int, int>();
static void Read(string threadName)
{
while (true)
{
try
{
// 获取读锁定
_rw.EnterReadLock();
Console.WriteLine($"{threadName} 从字典中读取内容 {DateTime.Now.ToString("mm:ss.ffff")}");
foreach (var key in _items.Keys)
{
Thread.Sleep(TimeSpan.FromSeconds(0.1));
}
}
finally
{
// 释放读锁定
_rw.ExitReadLock();
}
}
}
static void Write(string threadName)
{
while (true)
{
try
{
int newKey = new Random().Next(250);
// 尝试进入可升级锁模式状态
_rw.EnterUpgradeableReadLock();
if (!_items.ContainsKey(newKey))
{
try
{
// 获取写锁定
_rw.EnterWriteLock();
_items[newKey] = 1;
Console.WriteLine($"{threadName} 将新的键 {newKey} 添加进入字典中 {DateTime.Now.ToString("mm:ss.ffff")}");
}
finally
{
// 释放写锁定
_rw.ExitWriteLock();
}
}
Thread.Sleep(TimeSpan.FromSeconds(0.1));
}
finally
{
// 减少可升级模式递归计数,并在计数为0时 推出可升级模式
_rw.ExitUpgradeableReadLock();
}
}
}
运行结果如下所示,与预期结果相符。
1.10 使用SpinWait类
SpinWait
是一个常用的混合模式的类,它被设计成使用用户模式等待一段时间,人后切换至内核模式以节省CPU时间。
它的使用非常简单,演示代码如下所示。
static void Main(string[] args)
{
var t1 = new Thread(UserModeWait);
var t2 = new Thread(HybridSpinWait);
Console.WriteLine("运行在用户模式下");
t1.Start();
Thread.Sleep(20);
_isCompleted = true;
Thread.Sleep(TimeSpan.FromSeconds(1));
_isCompleted = false;
Console.WriteLine("运行在混合模式下");
t2.Start();
Thread.Sleep(5);
_isCompleted = true;
Console.ReadLine();
}
static volatile bool _isCompleted = false;
static void UserModeWait()
{
while (!_isCompleted)
{
Console.Write(".");
}
Console.WriteLine();
Console.WriteLine("等待结束");
}
static void HybridSpinWait()
{
var w = new SpinWait();
while (!_isCompleted)
{
w.SpinOnce();
Console.WriteLine(w.NextSpinWillYield);
}
Console.WriteLine("等待结束");
}
运行结果如下两图所示,首先程序运行在模拟的用户模式下,使CPU有一个短暂的峰值。然后使用SpinWait
工作在混合模式下,首先标志变量为False
处于用户模式自旋中,等待以后进入内核模式。
参考书籍
本文主要参考了以下几本书,在此对这些作者表示由衷的感谢你们提供了这么好的资料。
- 《CLR via C#》
- 《C# in Depth Third Edition》
- 《Essential C# 6.0》
- 《Multithreading with C# Cookbook Second Edition》
源码下载点击链接 示例源码下载