一：监视锁

Monitor:限定线程个数的一把锁

Monitor.Enter(b);锁住某一个资源

Monitor.Exit(b);释放某一个资源

static object lockMe = new object();
       static int num = 0;
       static void Main(string[] args)
       {
           for (int i = 0; i < 5; i++)
           {
               Task.Factory.StartNew(()=> {
                   Run();
               });
           }
Console.ReadKey();
   }
   static void Run()
   {
       for (int i = 0; i < 100; i++)
       {
           var b = false;
           try
           {
               Monitor.Enter(lockMe, ref b);
               Console.WriteLine(num++);
           }
           catch (Exception ex)
           {
               Console.WriteLine(ex.Message);
           }
           finally
           {
               if (b)
               {
                   Monitor.Exit(lockMe);
               }
           }
       }
   }

简化上面的代码：

   static object lockMe = new object();
    static int num = 0;
    static void Main(string[] args)
    {
        for (int i = 0; i < 5; i++)
        {
            Task.Factory.StartNew(()=> {
                Run();
            });
        }Console.ReadKey();
}
static void Run()
{
    for (int i = 0; i < 100; i++)
    {
        lock (lockMe)
        {
            Console.WriteLine(num++);
        }
    }
}

二：Lock/Monitor的内部机制

因为众多的锁机制中，唯独只有Monitor有专有的语法糖，所以比较重视！

一：ReaderWriteLock

从读写的角度进行功能分区

Sqllite:库锁

SqlServer:行锁【锁住行】

多个线程可以一起度，只能让一个线程去写。

读写是8/2开

读和写的时间如果时间比较长，都会影响他们。

static ReaderWriterLock reader = new ReaderWriterLock();
    static void Main(string[] args)
    {
        for (int i = 0; i < 5; i++)
        {
            Task.Factory.StartNew(() => {
                Read();
            });
        }
        Task.Factory.StartNew(() => {
            Write();
        });
        Console.ReadKey();
    }
/// <summary>
/// 线程读
/// </summary>
static void Read()
{
    while (true)
    {
        Thread.Sleep(10);
        reader.AcquireReaderLock(int.MaxValue);
        Thread.Sleep(10);
        Console.WriteLine("当前t={0}进行读取{1}",Thread.CurrentContext.ContextID,DateTime.Now);
        reader.ReleaseReaderLock();
    }
}

/// <summary>
/// 线程写
/// </summary>
static void Write()
{
    while (true)
    {
        Thread.Sleep(3000);

        reader.AcquireWriterLock(int.MaxValue);

        Thread.Sleep(3000);

        Console.WriteLine("当前t={0}进行写入........................................{1}", Thread.CurrentContext.ContextID,DateTime.Now);

        reader.ReleaseWriterLock();
    }
}

二：CountdownEvent

1

static CountdownEvent countdown = new CountdownEvent(10);

限制线程书的一个机制。

  static CountdownEvent countdown = new CountdownEvent(10);
 static void Main(string[] args)
 {
     countdown.Reset(10);
for (int i = 0; i < 10; i++)
 {
     Task.Factory.StartNew(()=> {
         LoadOrders();
     });

 }
 countdown.Wait();
 Console.WriteLine("所有Orders表执行完毕，恭喜！");
 countdown.Reset(5);

 for (int i = 0; i < 5; i++)
 {
     Task.Factory.StartNew(() => {
         LoadProducts();
     });
 }
 countdown.Wait();
 Console.WriteLine("所有Products表执行完毕，恭喜！");
 countdown.Reset(2);

 for (int i = 0; i < 2; i++)
 {
     Task.Factory.StartNew(() => {
         LoadUsers();
     });

 }
 countdown.Wait();
 Console.WriteLine("所有Users表执行完毕，恭喜！");
 Console.WriteLine("所有的表执行完毕，恭喜！");
      Console.ReadKey();
 }
 /// <summary>
 /// 加载Order表
 /// </summary>
 static void LoadOrders()
 {
     //将当前的countdown减减操作

     Console.WriteLine("当前Order正在加载中。。。{0}",Thread.CurrentThread.ManagedThreadId);
     countdown.Signal();
 }
 /// <summary>
 /// 加载Products表
 /// </summary>
 static void LoadProducts()
 {
     Console.WriteLine("当前Products正在加载中。。。{0}", Thread.CurrentThread.ManagedThreadId);
     countdown.Signal();
 }

 /// <summary>
 /// 加载Users表
 /// </summary>
 static void LoadUsers()
 {
     Console.WriteLine("当前Users正在加载中。。。{0}", Thread.CurrentThread.ManagedThreadId);
     countdown.Signal();

 }

一：锁机制的内核模式

1.在万不得已的情况下，不要使用内核模式的锁，因为代价太大。我们可以有其他方法代替，例如：混合锁机制，lock

2.事件锁

3.信号量

4.互斥锁

二：事件锁【开关锁，true、false的变量来进行控制】

1.自动事件锁【AutoResetEvent】

场景：可以用此所实现多线程环境下某个变量的自增。【true：表示终止状态，false：表示非终止】

private static AutoResetEvent auto = new AutoResetEvent(true);
static void Main(string[] args)
{
while (true)
{
Thread.Sleep(1000);
Console.WriteLine(“开始检票”);
auto.WaitOne();
Thread.Sleep(1000);
Console.WriteLine(“检票ing”);
auto.Set();
Thread.Sleep(1000);
Console.WriteLine(“检票结束”);
}

}

2.手动时间锁[ManualResetEvent]

private static ManualResetEvent manual = new ManualResetEvent(true);
       manual.WaitOne();
       manual.Set();

3.注意点：两者ManualResetEvent和AutoResetEvent是不一样的，所以不能混用。

4.Semaphore:[通过int整数来控制线程个数]

static Semaphore semaphore = new Semaphore(1,1);当前只能是一个线程通过

5.Mutex:互斥锁，同一时间只有一个线程可以拥有它，该类还可用于进程间同步的同步基元。

1
2
3
4
5

private static Mutex mutex = new Mutex();

 mutex.WaitOne();

 mutex.ReleaseMutex();

一：锁机制

1.net锁机制

时间锁、信号锁、互斥锁、读写锁、互锁，易变构造

分类：

1. 用户模式锁：通过一些cpu指令或者一个死循环达到thread等待和休眠。
2. 内核模式锁：调用win32底层的代码，来实现thread的各种操作，如：Thread.Sleep
3. 混合锁：用户模式+内核模式

2.为什么要用锁？

多线程对一个共享资源进行操作的时候，容易出现共享资源混乱的问题

二：用户模式锁

1. 易变结构：一个线程读，一个线程写，z在release版本中会有bug;

解决问题：

• Thread.MemoryBarrier, Thread.VolatileRead

• 关键字：volatile

  1. 不可以底层对代码进行优化

  2. 我的Read和Write都是从memrory中读取，读取的数据是最新的。

     1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

     public static volatile bool isStop = false; static void Main(string[] args) { var t = new Thread(()=> { var isSuccess = false; while (!isStop) { isSuccess = true; } }); t.Start(); Thread.Sleep(1000); isStop = true; t.Join(); Console.WriteLine("主线程执行结束！"); Console.ReadKey(); }

2.互锁结构：Interlocked[还只能做一些简单类型计算]

Interlocked：

• Increment：自增

  var sum = 0;
      for (int i = 0; i < 10; i++)
      {
          Interlocked.Increment(ref sum);
      }
       //最后sum=10,自增单位为1

• Decrement：自减

  1 2 3 4 5 6

  var sum = 0; for (int i = 0; i < 10; i++) { Interlocked.Decrement(ref sum); } //最后sum=-10，自减单位为1

• Exchange：赋值

  1 2 3

  var sum = 5; Interlocked.Exchange(ref sum,10); //最好sum=10;

• Add:增加指定的值

  1 2 3

  var sum = 5; Interlocked.Add(ref sum,2); //最好sum=7

• CompareExchange：比较赋值

  var sum = 5;
  Interlocked.CompareExchange(ref sum, 6, 5);
  //如果sum==5,true：sum=6,false=5;

  3.旋转锁：Soinlock

特殊的业务逻辑让thread在用户模式下进行自选，欺骗cpu当前thread正在运行中

用户模式—>内核模式—>用户模式

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

public static SpinLock spinLock = new SpinLock();
   static void Main(string[] args)
        {
            for (int i = 0; i < 3; i++)
            {
                var task = Task.Factory.StartNew(()=> {
                    Run();
                });
            }
            Console.ReadKey();
        }
   static void Run()
        {
            for (int i = 0; i < 100; i++)
            {
                try
                {
                    var b = false;
                    spinLock.Enter(ref b); ;
                    num++;
                    Console.WriteLine(num);
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
                finally
                {
                    spinLock.Exit();
                }
            }
        }

SpinLock 仅当您确定这样做可以改进应用程序的性能之后才能使用。另外，务必请注意 SpinLock 是一个值类型（出于性能原因）。因此，您必须非常小心，不要意外复制了 SpinLock 实例，因为两个实例（原件和副本）之间完全独立，这可能会导致应用程序出现错误行为。如果必须传递 SpinLock 实例，则应该通过引用而不是通过值传递。

不要将 SpinLock 实例存储在只读字段中

当锁是细粒度的并且数量巨大（例如链接的列表中每个节点一个锁）时以及锁保持时间总是非常短时，旋转可能非常有帮助

一：async await

• 调用方法：该方法调用异步方法，然后在异步方法执行其任务的时候继续执行；
• 异步方法：该方法异步执行工作，然后立刻返回到调用方法；
• await 表达式：用于异步方法内部，指出需要异步执行的任务。一个异步方法可以包含多个 await 表达式（不存在 await 表达式的话 IDE 会发出警告）。

二、async/await 结构

• 同步方法：一个程序调用某个方法，等到其执行完成之后才进行下一步操作。这也是默认的形式。

• 异步方法：一个程序调用某个方法，在处理完成之前就返回该方法。通过 async/await 我们就可以实现这种类型的方法。

三： 异步方法

• 关键字：方法头使用 async 修饰。
• 要求：包含 N（N>0） 个 await 表达式（不存在 await 表达式的话 IDE 会发出警告），表示需要异步执行的任务。
• ​ （3）返回类型：只能返回 3 种类型（void、Task 和 Task）。Task 和 Task 标识返回的对象会在将来完成工作，表示调用方法和异步方法可以继续执行。
• 参数：数量不限，但不能使用 out 和 ref 关键字。
• 命名约定：方法后缀名应以 Async 结尾。
• 其它：匿名方法和 Lambda 表达式也可以作为异步对象；async 是一个上下文关键字；关键字 async 必须在返回类型前。

一：同步编程模型(SPM)

单线线程、串行开发模式。

二：异步编程模型（APM）

xxxbegin、xxxend的模式。

   FileStream file = new FileStream(Environment.CurrentDirectory+"//1.txt",FileMode.Open);
var bytes= new byte[file.Length];

   file.BeginRead(bytes,0,bytes.Length,(ary)=> {
       var nums = file.EndRead(ary);
       Console.WriteLine(nums);
   },string.Empty);

   Console.Read();

三：基于事件的编程模型（EAP）

xxAsync这样的事件模式。 eg：WebClient。

四：基于Task的编程模型（TAP）

APM和EAP都可以使用Task来实现，微软的初衷就是想通过Task大一统异步编程领域

使用Task封装APM模式：

代码量小

Task更简单

FileStream file = new FileStream(Environment.CurrentDirectory+"//1.txt",FileMode.Open);

       var bytes= new byte[file.Length];
        var task = Task.Factory.FromAsync(file.BeginRead, file.EndRead, bytes, 0, bytes.Length, string.Empty);

        var nums = task.Result;
        Console.WriteLine(nums);

使用Task包装EAP

TaskCompletionSource：包装器

  private static Task<int> GetTask(string url)
     {
         TaskCompletionSource<int> source = new TaskCompletionSource<int>();
WebClient client = new WebClient();
     client.DownloadDataCompleted += (sendr, e) =>
     {
         try
         {
             source.TrySetResult(e.Result.Length);
         }
         catch (Exception ex)
         {
             source.TrySetException(ex);
         }

     };
     client.DownloadDataAsync(new Uri(url));
     return source.Task;
 }

一：TaskScheduler

问：TaskScheduler是做什么的？

答：我们发现任务执行要经过Scheduler。Task的核心就是这个Scheduler，因为他把要执行任务安排在线程或者线程池中。TaskScheduler由Thread和ThreadPool组成。

二：在.net FrameWork中有两种TaskScheduler

1.ThreadPoolTaskScheduler

也就是Task的默认是调用ThreadPool

1
2
3
4
5
6
7
8
9
10
11
12
13

protected internal override void QueueTask(Task task)
		{
			if ((task.Options & TaskCreationOptions.LongRunning) != TaskCreationOptions.None)
			{
				new Thread(ThreadPoolTaskScheduler.s_longRunningThreadWork)
				{
					IsBackground = true
				}.Start(task);
				return;
			}
			bool forceGlobal = (task.Options & TaskCreationOptions.PreferFairness) > TaskCreationOptions.None;
			ThreadPool.UnsafeQueueCustomWorkItem(task, forceGlobal);
		}

2.SynchronizationContextTaskScheduler

1.同步上下文

Task task = new Task(() =>
{
    try
    {
        label1.Text = "das";
    }
    catch (Exception ex)
    {
        MessageBox.Show(ex.Message);
    }
});
task.Start(TaskScheduler.FromCurrentSynchronizationContext());

2.操作耗时线程

1
2
3
4
5
6
7
8

var task = Task.Factory.StartNew(() =>
           {
               Thread.Sleep(10000);
           });
           task.ContinueWith(t =>
           {
               label1.Text = "你好!";
           }, TaskScheduler.FromCurrentSynchronizationContext());

ContinueWith等待执行完耗时线程，在同步上下文

三：如何自定义Scheduler？

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

public class PerThreadTaskScheduler : TaskScheduler
   {
       /// <summary>
       /// 给debug
       /// </summary>
       /// <returns></returns>
       protected override IEnumerable<Task> GetScheduledTasks()
       {
           return Enumerable.Empty<Task>();
       }
       /// <summary>
       /// 执行Task
       /// </summary>
       /// <param name="task"></param>
       protected override void QueueTask(Task task)
       {
           var thread = new Thread(()=> {
               TryExecuteTask(task);
           });
           thread.Start();
       }
       /// <summary>
       /// 同步执行
       /// </summary>
       /// <param name="task"></param>
       /// <param name="taskWasPreviouslyQueued"></param>
       /// <returns></returns>
       protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
       {
           return true;
       }
   }

一：Plinq

问：为什么要使用Plinq？

答：为了能够达到最大的灵活度，Linq又了并行的版本

二：如何把Linq转换为lingq

1.AsParallel():将串行转换为并行

var nums = Enumerable.Range(0, 100).ToList();
            var query = from n in nums.AsParallel()
                        select new
                        {
                            thread = Thread.CurrentThread.ManagedThreadId,
                            nums = n
                        };

        foreach (var item in query)
        {
            Console.WriteLine(item);
        }

2.AsOrdered()

说明：就是将并行结果还是按照未排序的样式排序

var nums = Enumerable.Range(0, 100).ToList();
            nums[0] = 1000;
                var query = from n in nums.AsParallel().AsOrdered()
                            select new
                            {
                                thread = Thread.CurrentThread.ManagedThreadId,
                                nums = n
                            };
foreach (var item in query)
        {
            Console.WriteLine(item);
        }

可以看出，AsOrdered()是按照1-100的排序输出，如果在前面加1000，还是会输出，并不会对结果在排序。

3.AsUnordered就是AsOrdered的相反的意思

4.AsSequential 《==》AsParallel

前者将plinq转换为linq

后者将Linq转换为Plinq

var nums = Enumerable.Range(0, 100).ToList();
    nums[0] = 1000;
        var query = from n in nums.AsParallel().AsSequential()
                    select new
                    {
                        thread = Thread.CurrentThread.ManagedThreadId,
                        nums = n
                    };
foreach (var item in query)
{
    Console.WriteLine(item);
}

线程串行，只有一个线程！

5.plinq底层都是用task的，基6于task的一些编程模型，让我们快速进行计算的。

 var nums = Enumerable.Range(0, 100).ToList();
            nums[0] = 1000;
                var query = from n in nums.AsParallel()
                            select new
                            {
                                thread = GetThreadID(),
                                nums = n
                            };
foreach (var item in query)
        {
            Console.WriteLine(item);
        }
        static int GetThreadID()
        {
            Thread.Sleep(1000);
            return Thread.CurrentThread.ManagedThreadId;
        }

6.WithDegreeOfParallelism(Environment.ProcessorCount-1)

设置Task开启线程的数量

7.WithCancellation(source.Token)

 CancellationTokenSource source = new CancellationTokenSource();
    source.Cancel();
     var nums = Enumerable.Range(0, 100).ToList();
    nums[0] = 1000;
    var query = from n in nums.AsParallel().WithDegreeOfParallelism(Environment.ProcessorCount - 1)
                .WithCancellation(source.Token)
                    select new
                    {
                        thread = GetThreadID(),
                        nums = n
                    };
foreach (var item in query)
{
    Console.WriteLine(item);
}

如果执行之前被取消，那就不要执行，会以异常保存。

7.WithExecutionMode：此参数可以告诉系统当前是否强制并行

var query = from n in nums.AsParallel().WithDegreeOfParallelism(Environment.ProcessorCount - 1)
                  .WithCancellation(source.Token)
                  .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                        select new
                      {
                          thread = GetThreadID(),
                          nums = n
                      };

8.WithMergeOptions:

1
2
3
4
5
6
7
8
9

var query = from n in nums.AsParallel().WithDegreeOfParallelism(Environment.ProcessorCount - 1)
                        .WithCancellation(source.Token)
                        .WithExecutionMode(ParallelExecutionMode.ForceParallelism)
                        .WithMergeOptions(ParallelMergeOptions.Default)
                              select new
                            {
                                thread = GetThreadID(),
                                nums = n
                            };

聚合计算合并

一：Task中并行和串行

串行：

for (int i = 0; i < 100; i++)
{
    Console.WriteLine(i);
}

并行：

1
2
3
4

Parallel.For(0, 100, (item) =>
   {
       Console.WriteLine(item);
   });

二：Parallel.For的实现结构

1.可以设置指定的线程计算，不需要所有的线程参与计算。

1

int nNumExpectedWorkers = (parallelOptions.EffectiveMaxConcurrencyLevel == -1) ? PlatformHelper.ProcessorCount : parallelOptions.EffectiveMaxConcurrencyLevel;

2.分区函数，实现线程并行

1

RangeManager rangeManager = new RangeManager((long)fromInclusive,(long)toExclusive, 1L, nNumExpectedWorkers);

分区后，使用 ParallelForReplicatingTask处理线程，这个类继承Task

3.不要在Parallel.For中使用break()或Stop()，或许会给你引入不必要的bug；

ConcurrentStack<int> stack = new ConcurrentStack<int>();
      Parallel.For(0, 100, (item,loop) =>
      {
          Thread.Sleep(100000);
          stack.Push(item);
      });
      Console.WriteLine(string.Join(",", stack));

线程池中开启16个线程，使用了12个线程,还有4个线程是CLR为发生饱和，做备用线程

三.Parallel.For高级重载

Parallel.For解决问题是数组的遍历。

public static ParallelLoopResult For(int fromInclusive, int toExclusive, Func localInit, Func<int, ParallelLoopState, TLocal, TLocal> body, Action localFinally)

计算1-100的总和

var totalNums = 0;
       Parallel.For(1,100,()=> {
           return 0;
       },(current,loop,total)=> {
           total += (int)current;
           return total;
       },(total)=> {
           Interlocked.Add(ref totalNums,total);
       });
       Console.WriteLine(totalNums);

结果4950

四：Parallel.Foreach()

Parallel.Foreach()应对一些集合运算【非数组】

Dictionary<int, int> dic = new Dictionary<int, int>() { 
{ 1, 100 }, { 2, 200 }, { 3, 300 } };
       Parallel.ForEach(dic,(item)=> {
            Console.WriteLine(item.Key+":"+item.Value);
        });

上面是源码，分区函数，首先得分区。

五：Parallel.Invoke（）

Parallel.Invoke(() =>
{
    Console.WriteLine("我是并行计算{0}",Thread.CurrentThread.ManagedThreadId);
}, () =>
{
    Console.WriteLine("我是并行计算{0}", Thread.CurrentThread.ManagedThreadId);
}, () =>
{
    Console.WriteLine("我是并行计算{0}", Thread.CurrentThread.ManagedThreadId);
});

Parallel.Invoke()可以同时开始多个线程并行执行，CPU回分配不同的线程执行任务。

一：Task的返回值：

1.使用Task.Result

Task<int> task = Task.Factory.StartNew(()=> {
    int s = 0;
    for (int i = 0; i < 5; i++)
    {
        s += i;
    }
    return s;
});
Console.WriteLine(task.Result);

二：ContinueWith的返回值

  Task<int> task = Task.Factory.StartNew(()=> {
        int s = 0;
        for (int i = 0; i < 5; i++)
        {
            s += i;
        }
        return s;
    });
    var task2 = task.ContinueWith<string>(t =>
    {var num = t.Result;
    var sum = num + 10;
    return sum.ToString();
});
Console.WriteLine(task2.Result);
Console.ReadKey();

三：Task.WhenAll/WhenAny

   Task<int> task = Task.Factory.StartNew(() =>
      {
          return 6;
      });
Task<int> task2 = Task.Factory.StartNew(() =>
  {
      return 2;
  });
  int[] result = Task.WhenAll<int>(new Task<int>[2] { task, task2 }).Result;

  foreach (var item in result)
  {
      Console.WriteLine(item);
  }
  Console.ReadKey();

四：异常处理

AggregateException是一个集合，因为task中可能回抛出多个异常，所以我们需要一种新的类型把这些异常都追加到一个集合种

1。什么时候抛出异常？

Task.Wait() or Tresult

2.何时会有多个异常在AggregateException中？

    var task = Task.Factory.StartNew(()=> {
        var child = Task.Factory.StartNew(() => {
            throw new Exception("我是child的异常");
        },TaskCreationOptions.AttachedToParent);
 var child2 = Task.Factory.StartNew(() => {
        throw new Exception("我是child2的异常");
    },TaskCreationOptions.AttachedToParent);
});
try
{
    task.Wait();
}
catch (AggregateException ex)
{
    foreach (var item in ex.InnerExceptions)
    {
        Console.WriteLine(string.Format("message:{0},type={1}", item.InnerException.Message, item.GetType().ToString()));
    }
}

3.Handle方法，就是处理当前的异常数组，判断上一层我当前哪些已经处理好了，没有处理好的，还需要向上抛出的？

Handle是用来便利异常数组，如果又一个异常信息是这样的，我任务已经处理，如果你觉得异常还需要往上抛，请返回fasle

try
{
task.Wait();
}
catch (AggregateException ex)
{
ex.Handle(x=> {
return true;
});
return false;
}