Mutex类、Event类、SemaphoreSlim类和ReaderWriterLockSlim类等提供了多个进程之间的线程同步。
1、WaitHandle 基类
WaitHandle抽象类,用于等待一个信号的设置。可以根据其派生类的不同,等待不同的信号。异步委托的BeginInvoke()方法返回一个实现了IAsycResult接口的对象。使用IAsycResult接口可以用AsycWaitHandle属性访问WaitHandle基类。在调用WaitOne()方法时,线程会等待接收一个和等待句柄相关的信号:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
|
static void Main( string [] args) { Func< int > func = new Func< int >( () => { Thread.Sleep(1500); return 1; }); IAsyncResult ar = func.BeginInvoke( null , null ); int count = 0; while ( true ) { Interlocked.Increment( ref count); Console.WriteLine( "第{0}周期循环等待结果。" , count); if (ar.AsyncWaitHandle.WaitOne(100, false )) { Console.WriteLine( "获得返回结果。" ); break ; } } int result = func.EndInvoke(ar); Console.WriteLine( "结果为:{0}" , result); } |
使用WaitHandle基类可以等待一个信号的出现(WaitHandle()方法)、等待多个对象都必须发出信号(WaitAll()方法)、等待多个对象中任一一个发出信号(WaitAny()方法)。其中WaitAll()方法和WaitAny()方法时WaitHandle类的静态方法,接收一个WaitHandle参数数组。
WaitHandle基类的SafeWaitHandle属性,其中可以将一个本机句柄赋予一个系统资源,等待该句柄,如I/O操作,或者自定义的句柄。
2、Mutex 类
Mutex类继承自WaitHandle类,提供跨多个进程同步访问的一个类。类似于Monitor类,只能有一个线程拥有锁定。在Mutex类的构造函数各参数含义:
- initiallyOwned: 如果为 true,则给予调用线程已命名的系统互斥体的初始所属权(如果已命名的系统互斥体是通过此调用创建的);否则为 false。
- name:系统互斥体的名称。 如果值为 null,则 System.Threading.Mutex 是未命名的。
- createdNew: 在此方法返回时,如果创建了局部互斥体(即,如果 name 为 null 或空字符串)或指定的命名系统互斥体,则包含布尔值 true;如果指定的命名系统互斥体已存在,则为false。 该参数未经初始化即被传递。
- mutexSecurity: 一个 System.Security.AccessControl.MutexSecurity 对象,表示应用于已命名的系统互斥体的访问控制安全性。
互斥也可以在另一个进程中定义,操作系统能够识别有名称的互斥,它由进程之间共享。如果没有指定互斥的名称,则不在不同的进程之间共享。该方法可以检测程序是否已运行,可以禁止程序启动两次。
1
2
3
4
5
6
7
8
9
10
11
|
static void Main( string [] args) { // ThreadingTimer(); // TimersTimer(); bool isCreateNew = false ; Mutex mutex = new Mutex( false , "MyApp" , out isCreateNew); //查询是否已有互斥“MyApp”存在 if (isCreateNew== false ) { //已存在互斥 } } |
要打开已有互斥,可以使用Mutex.OpenExisting()方法,不需要构造函数创建互斥时需要的相同.Net权限。可以使用WaitOne()方法获得互斥的锁定,成为该互斥的拥有着。调用ReleaseMutex()方法释放互斥:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
if (mutex.WaitOne()) //设置互斥锁定 { try { //执行代码 } finally { mutex.ReleaseMutex(); //释放互斥 } } else { //出现问题 } |
3、Semaphore 类
信号量是一种计数的互斥锁定,可以同时由多个线程使用。信号量可定义允许同时访问受旗语锁定保护的资源的线程个数。Semaphore和SemaphoreSlim两个类具有信号量功能。Semaphore类可以指定名称,让其在系统资源范围内查找到,允许在不同的进程之间同步。Semaphore类是对较短等待时间进行优化了的轻型版本。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
|
static void Main() { int taskCount = 6; int semaphoreCount = 3; Semaphore semaphore = new Semaphore(semaphoreCount, semaphoreCount, "Test" ); //创建计数为3的信号量 /* 第一个参数为初始释放的锁定数,第二个参数为可锁定的总数。如果第一个参数小于第二个参数,其差值就是已分配线程的计量数。 * 第三个参数为信号指定的名称,能让它在不同的进程之间共享。 */ var tasks = new Task[taskCount]; for ( int i = 0; i < taskCount; i++) { tasks[i] = Task.Run(() => TaskMain(semaphore)); //创建6个任务 } Task.WaitAll(tasks); Console.WriteLine( "All tasks finished" ); } //锁定信号的任务 static void TaskMain(Semaphore semaphore) { bool isCompleted = false ; while (!isCompleted) //循环等待被释放的信号量 { if (semaphore.WaitOne(600)) //最长等待600ms { try { Console.WriteLine( "Task {0} locks the semaphore" , Task.CurrentId); Thread.Sleep(2000); //2s后释放信号 } finally { Console.WriteLine( "Task {0} releases the semaphore" , Task.CurrentId); semaphore.Release(); //释放信号量 isCompleted = true ; } } else { //超过规定的等待时间,写入一条超时等待的信息 Console.WriteLine( "Timeout for task {0}; wait again" , Task.CurrentId); } } } |
以上方法中,信号量计数为3,因此最多只有三个任务可获得锁定,第4个及以后的任务必须等待。在解除锁定时,任何情况下一定要解除资源的锁定。
4、Events 类
事件也是一个系统范围内资源同步的方法。主要由以下几个类提供:ManualResetEvent、AutoResetEvent、ManualResetEventSlim、和CountdownEvent类。
ManualResetEventSlim类中,调用Set()方法可以发出信号;调用Reset()方法可以使重置为无信号状态。如果多个线程在等待向一个事件发出信号,并调用Set()方法,就释放所有等待线程。如果一个线程刚刚调用了WiatOne()方法,但事件已发出信号,等待的线程就可以继续等待。
AutoResetEvent类中,同样可以通过Set()方法发出信号、Reset()方法重置信号,但是该类是自动重置信号。如果一个线程在等待自动重置的事件发信号,当第一个线程的等待状态结束时,该事件会自动变为不发信号的状态。即:如果多个线程在等待向事件发信号,只有一个线程结束其等待状态,它不是等待事件最长的线程,而是优先级最高的线程。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
|
//计算数据的类,使用ManualResetEventSlim类的示例 public class Calculator { private ManualResetEventSlim mEvent; public int Result { get ; private set ; } public Calculator(ManualResetEventSlim ev) { this .mEvent = ev; } public void Calculation( int x, int y) { Console.WriteLine( "Task {0} starts calculation" , Task.CurrentId); Thread.Sleep( new Random().Next(3000)); //随机等待事件 Result = x + y; //计算结果 Console.WriteLine( "Task {0} is ready" , Task.CurrentId); mEvent.Set(); //发出完成信号 } } //外部调用的示例: static void Main() { const int taskCount = 10; ManualResetEventSlim[] mEvents = new ManualResetEventSlim[taskCount]; WaitHandle[] waitHandles = new WaitHandle[taskCount]; var calcs = new Calculator[taskCount]; for ( int i = 0; i < taskCount; i++) { int i1 = i; //目的是使后面要执行的Task不必等待执行完后才释放i,让for继续 mEvents[i] = new ManualResetEventSlim( false ); //对应任务的事件对象发出信号 waitHandles[i] = mEvents[i].WaitHandle; //ManualResetEvent类派生自WaitHandle类,但ManualResetEventSlim并不是,因此需要保存其WaitHandle对象 calcs[i] = new Calculator(mEvents[i]); Task.Run(() => calcs[i1].Calculation(i1 + 1, i1 + 3)); } for ( int i = 0; i < taskCount; i++) { int index = WaitHandle.WaitAny(waitHandles); //等待任何一个发出信号,并返回发出信号的索引 if (index == WaitHandle.WaitTimeout) { Console.WriteLine( "Timeout!!" ); } else { mEvents[index].Reset(); //重新设置为无信号状态 Console.WriteLine( "finished task for {0}, result: {1}" , index, calcs[index].Result); } } } |
CountdownEvent类适用于:需要把一个工作任务分配到多个任务中,然后在各个任务结束后合并结果(不需要为每个任务单独创建事件对象)。每个任务不需要同步。CountdownEvent类为所有设置了事件的任务定义了一个初始数字,达到该计数后,就发出信号。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
|
//修改计算类 public class Calculator { private CountdownEvent cEvent; public int Result { get ; private set ; } public Calculator(CountdownEvent ev) { this .cEvent = ev; } public void Calculation( int x, int y) { Console.WriteLine( "Task {0} starts calculation" , Task.CurrentId); Thread.Sleep( new Random().Next(3000)); //随机等待事件 Result = x + y; //计算结果 // signal the event—completed! Console.WriteLine( "Task {0} is ready" , Task.CurrentId); cEvent.Signal(); //发出完成信号 } } //修改方法调用 static void Main() { const int taskCount = 10; CountdownEvent cEvent = new CountdownEvent(taskCount); WaitHandle[] waitHandles = new WaitHandle[taskCount]; var calcs = new Calculator[taskCount]; for ( int i = 0; i < taskCount; i++) { int i1 = i; //目的是使后面要执行的Task不必等待执行后才释放i,让for可以继续 calcs[i] = new Calculator(cEvent); //为每个任务都赋该事件通知类 Task.Run(() => calcs[i1].Calculation(i1 + 1, i1 + 3)); } cEvent.Wait(); //等待一个事件的信号 Console.WriteLine( "all finished" ); for ( int i = 0; i < taskCount; i++) { Console.WriteLine( "task for {0}, result: {1}" , i, calcs[i].Result); } } |
5、Barrier 类
Barrier类适用于:工作有多个任务分支,并且在所有任务执行完后需要合并的工作情况。与CountdownEvent不同于,该类用于需要同步的参与者。在激活一个任务后,可以动态的添加其他参与者。在主参与者继续之前,可以等待所有其他参与者完成工作。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
|
static void Main() { const int numberTasks = 2; const int partitionSize = 1000000; var data = new List< string >(FillData(partitionSize * numberTasks)); var barrier = new Barrier(numberTasks + 1); //定义三个参与者:一个主参与者(分配任务者),两个子参与者(被分配任务者) var tasks = new Task< int []>[numberTasks]; //两个子参与者 for ( int i = 0; i < numberTasks; i++) { int jobNumber = i; tasks[i] = Task.Run(() => CalculationInTask(jobNumber, partitionSize, barrier, data)); //启动计算任务:可以分开写,以执行多个不同的任务。 } barrier.SignalAndWait(); // 主参与者以完成,等待子参与者全部完成。 //合并两个结果(LINQ) IEnumerable< int > resultCollection = tasks[0].Result.Zip(tasks[1].Result, (c1, c2) => { return c1 + c2; }).ToList(); //立即求和 char ch = 'a' ; int sum = 0; foreach (var x in resultCollection) { Console.WriteLine( "{0}, count: {1}" , ch++, x); //输出结果 sum += x; } Console.WriteLine( "main finished {0}" , sum); //统计过的字符串数量 Console.WriteLine( "remaining {0}, phase {1}" , barrier.ParticipantsRemaining, barrier.CurrentPhaseNumber); //当前参与者信息 } static int [] CalculationInTask( int jobNumber, int partitionSize, Barrier barrier, IList< string > coll) { var data = new List< string >(coll); int start = jobNumber * partitionSize; //计算其实下标 int end = start + partitionSize; //计算结束的位置 Console.WriteLine( "Task {0}: partition from {1} to {2}" , Task.CurrentId, start, end); int [] charCount = new int [26]; for ( int j = start; j < end; j++) { char c = data[j][0]; //获取当前字符串的第一个字符 charCount[c - 97]++; //对应字符的数量+1; } Console.WriteLine( "Calculation completed from task {0}. {1} times a, {2} times z" , Task.CurrentId, charCount[0], charCount[25]); //告知计算完成 barrier.RemoveParticipant(); //告知,减少一个参数者 Console.WriteLine( "Task {0} removed from barrier, remaining participants {1}" , Task.CurrentId, barrier.ParticipantsRemaining); return charCount; //返回统计的结果 } //用于填充一个字符串链表 public static IEnumerable< string > FillData( int size) { var data = new List< string >(size); var r = new Random(); for ( int i = 0; i < size; i++) { data.Add(GetString(r)); //获得一个随机的字符串 } return data; } private static string GetString(Random r) { var sb = new StringBuilder(6); for ( int i = 0; i < 6; i++) { sb.Append(( char )(r.Next(26) + 97)); //创建一个6个字符的随机字符串 } return sb.ToString(); } |
6、ReaderWriterLockSlim 类
该类是使锁定机制允许锁定多个读取器(而不是一个写入器)访问某个资源:如果没有写入器锁定资源,那么允许多个读取器访问资源,但只能有一个写入器锁定该资源。
由它的属性可以读取是否处于堵塞或不堵塞的锁定,如EnterReadLock()和TryEnterReadLock()方法。也可以获得其是否处于写入锁定或非锁定状态,如EnterWriteLock()和TryEnterWriteLock()方法。如果任务需要先读取资源,之后写入资源,可以使用EnterUpgradeableReadLock()或TryEnterUpgradeableReadLock()方法获取可升级的读取锁定。该锁定可以获取写入锁定,而不需要释放读取锁定。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
class Program { private static List< int > items = new List< int >() { 0, 1, 2, 3, 4, 5 }; private static ReaderWriterLockSlim rwl = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion); static void ReaderMethod( object reader) { try { rwl.EnterReadLock(); for ( int i = 0; i < items.Count; i++) { Console.WriteLine( "reader {0}, loop: {1}, item: {2}" , reader, i, items[i]); Thread.Sleep(40); } } finally { rwl.ExitReadLock(); } } static void WriterMethod( object writer) { try { while (!rwl.TryEnterWriteLock(50)) { Console.WriteLine( "Writer {0} waiting for the write lock" , writer); Console.WriteLine( "current reader count: {0}" , rwl.CurrentReadCount); } Console.WriteLine( "Writer {0} acquired the lock" , writer); for ( int i = 0; i < items.Count; i++) { items[i]++; Thread.Sleep(50); } Console.WriteLine( "Writer {0} finished" , writer); } finally { rwl.ExitWriteLock(); } } static void Main() { var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None); var tasks = new Task[6]; tasks[0] = taskFactory.StartNew(WriterMethod, 1); tasks[1] = taskFactory.StartNew(ReaderMethod, 1); tasks[2] = taskFactory.StartNew(ReaderMethod, 2); tasks[3] = taskFactory.StartNew(WriterMethod, 2); tasks[4] = taskFactory.StartNew(ReaderMethod, 3); tasks[5] = taskFactory.StartNew(ReaderMethod, 4); for ( int i = 0; i < 6; i++) { tasks[i].Wait(); } } } |
以上就是c# 进程之间的线程同步的详细内容,更多关于c# 线程同步的资料请关注服务器之家其它相关文章!
原文链接:https://www.cnblogs.com/pilgrim/p/9250052.html