第2章 线程同步
原来以为线程同步就是lock，monitor等呢，看了第二章真是大开眼界啊！
第一章中我们遇到了一个叫做竞争条件的问题。引起的原因是没有进行正确的线程同步。当一个线程在执行操作时候，其他的线程需要依次等待。这样的问题通常被称为线程同步。
有多种方式来进行线程的同步。
第一：首先线程同步的原因是，多线程访问共享对象，如果可以通过重新设计程序来移除共享状态，从而去掉复杂的同步构造。
第二：使用原子操作，所谓原子操作就是一个操作只占用一个量子时间，一次就可以完成。所以只有当前操作结束之后，其他线程才能执行其他操作。这时无需实现其他线程等待当前操作完成，这就避免了使用锁，也就排除了死锁的可能。
第三：上面两者不可行，我们就要采用一些方式来协调线程。
方法就是通过将线程置为阻塞状态，当线程置为阻塞状态的时候，此时线程只会占用很少的cpu时间，但是会引入至少一次的上下文切换。上下文切换是指操作系统的线程调度器。该调度器会保存等待线程的状态，并切换到另一个线程，依次恢复等待的线程的状态。每一次线程间的切换是非常消耗资源的。
但是如果线程会挂起很长时间，这么做是值得的。这种方式叫做内核模式，因为只有内核才能阻止线程占用cpu时间。
如果线程只需要等待一小段时间，最好只是简单的等待，而不用将线程切换到阻塞状态。虽然线程等待时候回浪费CPU时间，但是这样却节省了上下文切换耗费的CPU时间。该方式被称为用户模式。这样的方式很轻量，速度很快，如果线程需要等待很长时间，则会浪费大量CPU时间。
为了缓解两种方式的问题，可以采用混合模式。混合模式先尝试用用户模式等待，超过一定时间限制，转为内核模式进入阻塞状态。
①原子操作
使用Interlocked类，这个类中提供一些Increment，Decrement和Add等基本的数学操作的原子方法。从而帮助我们在编写一些代码时候，无需使用锁。
例如：下面的例子，我们定义了两个计数的方法，分别用于自增和自减，区别在于第一次没有使用原子操作，第二次使用了原子操作。结果可以看出来，使用原子操作的结果是0，实现了线程间的同步。
using System; using System.Threading; namespace Chapter2.Recipe1
{
internal class Program
{
private static void Main(string[] args)
{
Console.WriteLine("Incorrect counter"); var c = new Counter(); 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(); Console.WriteLine("Total count: {0}", c.Count);
Console.WriteLine("--------------------------"); Console.WriteLine("Correct counter"); var c1 = new CounterNoLock(); t1 = new Thread(() => TestCounter(c1));
t2 = new Thread(() => TestCounter(c1));
t3 = new Thread(() => TestCounter(c1));
t1.Start();
t2.Start();
t3.Start();
t1.Join();
t2.Join();
t3.Join(); Console.WriteLine("Total count: {0}", c1.Count);
} static void TestCounter(CounterBase c)
{
for (int i = ; i < ; i++)
{
c.Increment();
c.Decrement();
}
} class Counter : CounterBase
{
private int _count; public int Count { get { return _count; } } public override void Increment()
{
_count++;
} public override void Decrement()
{
_count--;
}
} class CounterNoLock : CounterBase
{
private int _count; public int Count { get { return _count; } } public override void Increment()
{
Interlocked.Increment(ref _count);
} public override void Decrement()
{
Interlocked.Decrement(ref _count);
}
} abstract class CounterBase
{
public abstract void Increment(); public abstract void Decrement();
}
}
}
②使用Mutex类
互斥锁（Mutex）
互斥锁是一个互斥的同步对象，意味着同一时间有且仅有一个线程可以获取它。
互斥锁可适用于一个共享资源每次只能被一个线程访问的情况。
using System;
using System.Threading; namespace Chapter2.Recipe2
{
class Program
{
static void Main(string[] args)
{
const string MutexName = "CSharpThreadingCookbook"; using (var m = new Mutex(false, MutexName))
{
if (!m.WaitOne(TimeSpan.FromSeconds(), false))
{
Console.WriteLine("Second instance is running!");
}
else
{
Console.WriteLine("Running!");
Console.ReadLine();
m.ReleaseMutex();
}
}
}
}
}
Mutex在什么地方获取，在什么地方释放，这个要记住。
一般是在委托的方法中使用，先获得，在执行操作代码，然后释放掉mutex量
class Program
{
static Mutex mu = new Mutex();
static void Main(string[] args)
{
Thread1();
Thread2();
Console.ReadKey();
} static void Count()
{
mu.WaitOne();
for (int i = ; i < ; i++)
{
Console.WriteLine("{0} is writing {1}",
Thread.CurrentThread.Name, i.ToString());
}
mu.ReleaseMutex();
} static void Thread1()
{
Thread thread1 = new Thread(Count);
thread1.Name = "Thread1";
thread1.Start();
} static void Thread2()
{
Thread thread2 = new Thread(Count);
thread2.Name = "Thread2";
thread2.Start();
}
}
上面的例子，是比较简单的例子。
如果不使用Mutex的话，把上面Mutex注释掉
③使用SemaphoreSlim类
它是Semaphore的轻量级版本。
static void Main(string[] args)
{
for (int i = ; i <= ; i++)
{
string threadName = "Thread " + i;
int secondsToWait = + * i;
var t = new Thread(() => AccessDatabase(threadName, secondsToWait));
t.Start();
}
} static SemaphoreSlim _semaphore = new SemaphoreSlim(); static void AccessDatabase(string name, int seconds)
{
Console.WriteLine("{0} waits to access a database", name);
_semaphore.Wait();
Console.WriteLine("{0} was granted an access to a database", name);
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("{0} is completed", name);
_semaphore.Release(); }
}
1、程序一启动就生成一个限制了线程并发数的SemaphoreSlim实例，限制其并发数目为4个
2、启动了六个线程，每个线程拥有不同的线程执行时间。
3、执行程序过程中，我们发现，最大并发数真的就是4个，如下图所示
首先2、4、1、3都获得了权限，5、6就必须等待，当1执行完成释放信号量，6才获得权限。
线程间的通信！
在讨论这个问题之前，我们先了解这样一种观点，线程之间的通信是通过发信号来进行沟通的。
③使用AutoResetEvent
AutoResetEvent类来从一个线程向另外一个线程发送通知。AutoResetEvent类可以通知等待的线程有某事件发生。
namespace Recipe4_2
{
class Program
{
static AutoResetEvent myAutoReset = new AutoResetEvent(false);
static void Main(string[] args)
{
Thread threadA = new Thread(FunctionA);
threadA.Name = "ThreadA";
Thread threadB = new Thread(FunctionB);
threadB.Name = "ThreadB";
threadA.Start();
threadB.Start();
} static void FunctionA()
{
for (int i = ; i <= ; i++)
{
Console.WriteLine("This is ThreadA {0}",i);
Thread.Sleep();
}
myAutoReset.Set();
} static void FunctionB()
{
myAutoReset.WaitOne();
for (int i = ; i <= ; i++)
{
Console.WriteLine("This is ThreadB {0}", i);
Thread.Sleep();
}
myAutoReset.Set();
}
}
}
上例中，程序启动创建了一个AutoResetEvent实例，并赋初始值为false，这个false定义了这个AutoResetEvent的实例的初始状态是unsignaled状态。这意味着我们调用这个实例的WaitOne方法将会被阻塞，直到我们调用set方法。如果初始值为true，那么AutoResetEvent实例的状态为signaled，如果调用WaitOne则立即被处理，然后事件状态立即转为unsignaled。AutoResetEvent采用的是内核时间模式，所以时间不能太长。如果需要处理长时间的操作，那么使用ManualResetEventSlim类更好。
④使用ManualResetEventSlim类
class Program
{
static void Main(string[] args)
{
var t1 = new Thread(() => TravelThroughGates("Thread 1", ));
var t2 = new Thread(() => TravelThroughGates("Thread 2", ));
var t3 = new Thread(() => TravelThroughGates("Thread 3", ));
t1.Start();
t2.Start();
t3.Start();
Thread.Sleep(TimeSpan.FromSeconds());
Console.WriteLine("The gates are now open!");
_mainEvent.Set();
Thread.Sleep(TimeSpan.FromSeconds()); //大门打开两秒钟关闭
_mainEvent.Reset();
Console.WriteLine("The gates have been closed!");
Thread.Sleep(TimeSpan.FromSeconds());
Console.WriteLine("The gates are now open for the second time!");
_mainEvent.Set();
Thread.Sleep(TimeSpan.FromSeconds());
Console.WriteLine("The gates have been closed!");
_mainEvent.Reset();
} static void TravelThroughGates(string threadName, int seconds)
{
Console.WriteLine("{0} falls to sleep", threadName);
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("{0} waits for the gates to open!", threadName);
_mainEvent.Wait();
Console.WriteLine("{0} enters the gates!", threadName);
} static ManualResetEventSlim _mainEvent = new ManualResetEventSlim(false);
}
AutoResetEvent这个类有点像旋转门，一次只允许一个人通过。ManualResetEventSlim是ManualResetEvent的混合版本，像一个手动开关的大门，一直保持大门的敞开直到调用Reset方法。当调用set方法的时候，相当于打开了大门从而允许准备好的线程接收信号并继续工作。没有准备好的线程，没有赶上大门打开的时间。调用Reset方法相当于关闭了大门。
⑤使用CountDownEvent类
使用CountDownEvent信号类来等待一定数量的操作完成。
例子1：
class Program
{
static void Main(string[] args)
{
Console.WriteLine("Starting two operations");
var t1 = new Thread(() => PerformOperation("Operation 1 is completed", ));
var t2 = new Thread(() => PerformOperation("Operation 2 is completed", ));
t1.Start();
t2.Start();
_countdown.Wait();
Console.WriteLine("Both operations have been completed.");
_countdown.Dispose();
} static CountdownEvent _countdown = new CountdownEvent();//初始就定义为2 static void PerformOperation(string message, int seconds)
{
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine(message);
_countdown.Signal();
}
}
CountdownEvent.Signal 方法
|
名称 |
说明 |
|
向 CountdownEvent 注册信号，同时减小 CurrentCount 的值。 |
|
|
向 CountdownEvent 注册多个信号，同时将 CurrentCount 的值减少指定数量。 |
|
|
阻止当前线程，直到设置了 CountdownEvent 为止。 |
|
|
释放 CountdownEvent 类的当前实例所使用的所有资源。 |
缺点:如果调用Signal()没有达到指定的次数，那么Wait就会一直等待下去。所以每次线程使用结束之后都要调用一次Signal()方法。
例子2：
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading; namespace CountdownEventDemo
{
class Program
{
static void Main(string[] args)
{
var customers = Enumerable.Range(, ); using (var countdown = new CountdownEvent())
{
foreach (var customer in customers)
{
int currentCustomer = customer;
ThreadPool.QueueUserWorkItem(delegate
{
BuySomeStuff(currentCustomer);
countdown.Signal();
});
countdown.AddCount();
} countdown.Signal();
countdown.Wait();
} Console.WriteLine("All Customers finished shopping...");
Console.ReadKey();
} static void BuySomeStuff(int customer)
{
// Fake work
Thread.SpinWait(); Console.WriteLine("Customer {0} finished", customer);
}
}
}
http://www.cnblogs.com/shanyou/archive/2009/10/27/1590890.html
|
将 CountdownEvent 的当前计数加 1。 |
|
|
将 CountdownEvent 的当前计数增加指定值。 |
⑥使用Barrier类
Barrier类用于组织多个线程及时在某个时刻碰面。其提供了一个回调函数，每次线程调用了SignalAndWait方法后该回调函数会被执行。
class Program
{
static void Main(string[] args)
{
var t1 = new Thread(() => PlayMusic("the guitarist", "play an amazing solo", ));
var t2 = new Thread(() => PlayMusic("the singer", "sing his song", )); t1.Start();
t2.Start();
} static Barrier _barrier = new Barrier(,
b => Console.WriteLine("End of phase {0}", b.CurrentPhaseNumber + )); static void PlayMusic(string name, string message, int seconds)
{
for (int i = ; i < ; i++)
{
Console.WriteLine("----------------------------------------------");
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("{0} starts to {1}", name, message);
Thread.Sleep(TimeSpan.FromSeconds(seconds));
Console.WriteLine("{0} finishes to {1}", name, message);
_barrier.SignalAndWait();
}
}
}
⑦使用ReaderWriterLockSilm类
这个类用来创建一个线程安全机制，在多线程中对一个集合进行读写操作。ReaderWriterLockSilm类代表了一个管理资源访问的锁，允许多个线程同时进行读取以及独占写。
下例中定义了三个读线程和两个写线程，读没什么好讲的，就是判断数据在不在集合里，在的话就读出来，不在就释放读锁。写操作是先创建一个新key，获取一下更新锁，判断新key在不在集合当中，不在的话，升级更新锁变成写锁，写入数据，然后释放写锁，延时，释放更新锁。
class Program
{
static void Main(string[] args)
{
new Thread(Read){ IsBackground = true }.Start();
new Thread(Read){ IsBackground = true }.Start();
new Thread(Read){ IsBackground = true }.Start(); new Thread(() => Write("Thread 1")){ IsBackground = true }.Start();
new Thread(() => Write("Thread 2")){ IsBackground = true }.Start(); Thread.Sleep(TimeSpan.FromSeconds());
} static ReaderWriterLockSlim _rw = new ReaderWriterLockSlim();
static Dictionary<int, int> _items = new Dictionary<int, int>(); static void Read()
{
Console.WriteLine("Reading contents of a dictionary");
while (true)
{
try
{
_rw.EnterReadLock();
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();
_rw.EnterUpgradeableReadLock();
if (!_items.ContainsKey(newKey))
{
try
{
_rw.EnterWriteLock();
_items[newKey] = ;
Console.WriteLine("New key {0} is added to a dictionary by a {1}", newKey, threadName);
}
finally
{
_rw.ExitWriteLock();
}
}
Thread.Sleep(TimeSpan.FromSeconds(0.1));
}
finally
{
_rw.ExitUpgradeableReadLock();
}
}
}
}
启动了三个读线程和两个写线程，ReaderWriterLockSlim类有两种锁，读锁和写锁。读锁允许多线程进行数据读取，写锁再被释放之前会阻塞其他线程的所有操作。
有一种场景是这样的，当我们需要根据当前读取的数据进行判读是否需要进行修改的时候，如果这时获取写锁，就会阻塞所有阅读者的权限，从而浪费大量的时间。为了减少阻塞的时间，可以使用EnterUpgradeableReadLock和ExitUpgradeaReadLock方法。这时当我们获得读锁进行数据读取的时候，如果发现必须要修改，只需要使用EnterWriteLock方法进行升级锁，然后快速一次写操作，最后用ExitWriteLock释放写锁。
看了一下有一个地方没明白，就是EnterUpgradeableReadLock和EnterReadLock他俩的区别，感觉很像！上网查了一下，大概明白了。
可更新锁：
再一个原子操作里将读锁升级为写锁是很有用的，例如，假设你想要再一个list 里面写一些不存在的项的时候，你可能会执行下面的一些步骤：
1获取一个读锁。
2测试，如果要写的东西在列表中，那么释放锁，然后返回。
3释放读锁。
4获取一个写锁
5添加项，写东西，
6释放写锁。
问题是：在第三步和第四步之间，可能有另一个线程修改了列表。
ReaderWriterLockSlim 通过一个叫做可更新锁( upgradeable lock),来解决这个问题。
一个可更新锁除了它可以在一个原子操作中变成写锁外很像一个读锁，你可以这样使用它：
调用EnterUpgradeableReadLock 获取可更新锁。执行一些读操作，例如判断要写的东西在不在List中。调用EnterWriteLock , 这个方法会将可更新锁 升级为 写锁。执行写操作，调用ExitWriteLock 方法，这个方法将写锁转换回可更新锁。继续执行一些读操作，或什么都不做。
调用ExitUpgradeableReadLock 释放可更新锁。
从调用者的角度来看，它很像一个嵌套/递归锁，从功能上讲，在第三步，
ReaderWriterLockSlim 在一个原子操作里面释放读锁，然后获取写锁。
可更新锁和读锁的重要区别是:尽管可更新锁可以和读锁共存，但是一次只能有一个可更新锁被获取。这样的主要目的是防止死锁。
貼り付け元 <http://www.jb51.net/article/36798.htm>
⑧使用SpinWait类
这个东西是利用混合模式来让线程进行等待，可以类比于Thread.Sleep()，感觉书上的Demo虽然解释起来还算方便，但是效果真的不好观察，于是我就在网上查一个文章，感觉总结的还挺不错。