进程之间的线程同步

　　Mutex类、Event类、SemaphoreSlim类和ReaderWriterLockSlim类等提供了多个进程之间的线程同步。

 一、WaitHandle 基类

　　WaitHandle抽象类，用于等待一个信号的设置。能够根据其派生类的不一样，等待不一样的信号。异步委托的BeginInvoke()方法返回一个实现了IAsycResult接口的对象。使用IAsycResult接口能够用AsycWaitHandle属性访问WaitHandle基类。在调用WaitOne()方法时，线程会等待接收一个和等待句柄相关的信号：

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操做，或者自定义的句柄。

二、Mutex 类

　　Mutex类继承自WaitHandle类，提供跨多个进程同步访问的一个类。相似于Monitor类，只能有一个线程拥有锁定。在Mutex类的构造函数各参数含义：

• initiallyOwned: 若是为 true，则给予调用线程已命名的系统互斥体的初始所属权（若是已命名的系统互斥体是经过此调用建立的）；不然为 false。
• name:系统互斥体的名称。 若是值为 null，则 System.Threading.Mutex 是未命名的。
• createdNew: 在此方法返回时，若是建立了局部互斥体（即，若是 name 为 null 或空字符串）或指定的命名系统互斥体，则包含布尔值 true；若是指定的命名系统互斥体已存在，则为false。 该参数未经初始化即被传递。
• mutexSecurity: 一个 System.Security.AccessControl.MutexSecurity 对象，表示应用于已命名的系统互斥体的访问控制安全性。

　　互斥也能够在另外一个进程中定义，操做系统可以识别有名称的互斥，它由进程之间共享。若是没有指定互斥的名称，则不在不一样的进程之间共享。该方法能够检测程序是否已运行，能够禁止程序启动两次。

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()方法释放互斥：

if(mutex.WaitOne())//设置互斥锁定
{
    try
    {
        //执行代码
    }
    finally {
        mutex.ReleaseMutex();//释放互斥
    }
}
else
{
    //出现问题
}

 三、Semaphore 类

　　信号量是一种计数的互斥锁定，能够同时由多个线程使用。信号量可定义容许同时访问受旗语锁定保护的资源的线程个数。Semaphore和SemaphoreSlim两个类具备信号量功能。Semaphore类能够指定名称，让其在系统资源范围内查找到，容许在不一样的进程之间同步。Semaphore类是对较短等待时间进行优化了的轻型版本。

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个及之后的任务必须等待。在解除锁定时，任何状况下必定要解除资源的锁定。

四、Events 类

　　事件也是一个系统范围内资源同步的方法。主要由如下几个类提供：ManualResetEvent、AutoResetEvent、ManualResetEventSlim、和CountdownEvent类。

　　ManualResetEventSlim类中，调用Set()方法能够发出信号；调用Reset()方法可使重置为无信号状态。若是多个线程在等待向一个事件发出信号，并调用Set()方法，就释放全部等待线程。若是一个线程刚刚调用了WiatOne()方法，但事件已发出信号，等待的线程就能够继续等待。

　　AutoResetEvent类中，一样能够经过Set()方法发出信号、Reset()方法重置信号，可是该类是自动重置信号。若是一个线程在等待自动重置的事件发信号，当第一个线程的等待状态结束时，该事件会自动变为不发信号的状态。即：若是多个线程在等待向事件发信号，只有一个线程结束其等待状态，它不是等待事件最长的线程，而是优先级最高的线程。

//计算数据的类，使用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);
        }
    }
}

View Code

　　CountdownEvent类适用于：须要把一个工做任务分配到多个任务中，而后在各个任务结束后合并结果（不须要为每一个任务单首创建事件对象）。每一个任务不须要同步。CountdownEvent类为全部设置了事件的任务定义了一个初始数字，达到该计数后，就发出信号。

//修改计算类
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);
    }
}

View Code

 五、Barrier 类

　　Barrier类适用于：工做有多个任务分支，而且在全部任务执行完后须要合并的工做状况。与CountdownEvent不一样于，该类用于须要同步的参与者。在激活一个任务后，能够动态的添加其余参与者。在主参与者继续以前，能够等待全部其余参与者完成工做。

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();
}

View Code

 六、ReaderWriterLockSlim 类

　　该类是使锁定机制容许锁定多个读取器（而不是一个写入器）访问某个资源：若是没有写入器锁定资源，那么容许多个读取器访问资源，但只能有一个写入器锁定该资源。

　　由它的属性能够读取是否处于堵塞或不堵塞的锁定，如EnterReadLock()和TryEnterReadLock()方法。也能够得到其是否处于写入锁定或非锁定状态，如EnterWriteLock()和TryEnterWriteLock()方法。若是任务须要先读取资源，以后写入资源，可使用EnterUpgradeableReadLock()或TryEnterUpgradeableReadLock()方法获取可升级的读取锁定。该锁定能够获取写入锁定，而不须要释放读取锁定。

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();
        }
    }
}

View Code