浅谈C#多线程下的调优

目录
  • 一、原子操作
    • 1.基于Lock实现
    • 2.基于CAS实现
    • 3.自旋锁SpinLock
    • 4.读写锁ReaderWriterLockSlim
  • 二、线程安全
    • 1.线程安全集合
    • 2.线程安全字典
  • 三、线程池
    • 1.通过QueueUserWorkItem启动工作者线程
    • 2.线程池等待(信号量)
    • 3.Task
    • 4.线程池调度原理
  • 四、并行
  • 五、异步IO
    • 1.异步IO于同步IO比较
    • 2.异步读写文件

一、原子操作

先看一段问题代码

/// <summary>
        /// 获取自增
        /// </summary>
        public static void GetIncrement()
        {
            long result = 0;
            Console.WriteLine("开始计算");
            //10个并发执行
            Parallel.For(0, 10, (i) =>
            {
                    for (int j = 0; j < 10000; j++)
                    {
                        result++;
                    }
            });
            Console.WriteLine("结束计算");
            Console.WriteLine($"result正确值应为:{10000 * 10}");
            Console.WriteLine($"result    现值为:{result}");
            Console.ReadLine();
        }

这是多线程下,result的值不同步的原因。

1.基于Lock实现

平时大家用的最多的应该就是加锁了,同一时间,只有一个线程进入代码块。

实现代码:

private static Object _obj = new object();
     /// <summary>
        /// 原子操作基于Lock实现
        /// </summary>
        public static void AtomicityForLock()
        {
            long result = 0;
            Console.WriteLine("开始计算");
            //10个并发执行
           Parallel.For(0, 10, (i) =>
            {
                //lock锁
                lock (_obj)
                {
                    for (int j = 0; j < 10000; j++)
                    {
                        result++;
                    }
                }
            });
            Console.WriteLine("结束计算");
            Console.WriteLine($"result正确值应为:{10000 * 10}");
            Console.WriteLine($"result    现值为:{result}");
            Console.ReadLine();

        }

结果:

2.基于CAS实现

  CAS是一种有名的无锁算法。无锁编程,即不适用锁的情况下实现多线程之间的变量同步,也就是在没有线程被阻塞的情况下实现变量的同步。
  CAS在.NET中的实现类是Interlocked,内部提供很多原子操作的方法,最终都是调用Interlocked.CompareExchange(ref out,更新值,期望值) //基于内存屏障的方式操作 (七个步骤)
说到线程安全,不要一下子就想到加锁,尤其是可能会调用频繁或者是要求高性能的场合。

  • CAS(Compare And Swap)比较并替换,是线程并发运行时用到的一种技术
  • CAS是原子操作,保证并发安全,而不能保证并发同步
  • CAS是CPU的一个指令(需要JNI调用Native方法,才能调用CPU的指令)
  • CAS是非阻塞的、轻量级的乐观锁

CAS的适用场景

  读多写少:如果有大量的写操作,CPU开销可能会过大,因为冲突失败后会不断重试(自旋),这个过程中会消耗CPU
  单个变量原子操作:CAS机制所保证的只是一个变量的原子操作,而不能保证整个代码块的原子性,比如需要保证三个变量共同进行原子性的更新,就不得不使用悲观锁了

Interlocked主要函数如下:
Interlocked.Increment    原子操作,递增指定变量的值并存储结果。
Interlocked.Decrement   原子操作,递减指定变量的值并存储结果。
Interlocked.Add        原子操作,添加两个整数并用两者的和替换第一个整数
Interlocked.Exchange  原子操作,赋值
Interlocked.CompareExchange(ref a, b, c); 原子操作,a参数和c参数比较, 相等b替换a,不相等不替换。方法返回值始终是第一个参数的原值,也就是内存里的值

用Interlocked.Increment实现上面自增功能

代码:

/// <summary>
        /// 自增CAS实现
        /// </summary>
        public static void AtomicityForInterLock()
        {
            long result = 0;
            Console.WriteLine("开始计算");
           Parallel.For(0, 10, (i) =>
            {
                for (int j = 0; j < 10000; j++)
                {
                    //自增
                    Interlocked.Increment(ref result);
                }
            });
            Console.WriteLine($"结束计算");
            Console.WriteLine($"result正确值应为:{10000 * 10}");
            Console.WriteLine($"result    现值为:{result}");
            Console.ReadLine();
        }

结果:

Interlocked下原子操作的方法最终都是调用Interlocked.CompareExchange(ref a, b, c)实现的,现在我们利用CompareExchange自己实现一个原子操作功能

实现“一个变量自增到10000,然后又初始化到1开始自增的功能“

代码:

/// <summary>
        /// 基于CAS原子操作自己写
        /// </summary>
        public static void AtomicityForMyCalc()
        {
            long result = 0;
            Console.WriteLine("开始计算");

            Parallel.For(0, 10, (i) =>
            {
                long init = 0;
                long incrementNum = 0;
                for (int j = 0; j < 10000; j++)
                {
                    do
                    {
                        init = result;
                        incrementNum = result + 1;
                        incrementNum= incrementNum > 10000 ? 1 : incrementNum; //自增到10000后初始化成1

                    }
                    //如果result=init,则result的值被incrementNum替换,否则result不变,返回的是result的原始值
                    while (init != Interlocked.CompareExchange(ref result, incrementNum, init));
                    if(incrementNum==10000)
                    {
                        Console.WriteLine($"自增到达10000啦!值被初始化为1");
                    }
                }
            });
            Console.WriteLine($"结束计算");

            Console.WriteLine($"result正确值应为:{10000}");
            Console.WriteLine($"result    现值为:{result}");
            Console.ReadLine();

        }

结果:

3.自旋锁SpinLock

自旋锁(spinlock):

是指当一个线程在获取锁的时候,如果锁已经被其它线程获取,那么该线程将循环等待,然后不断的判断锁是否能够被成功获取,直到获取到锁才会退出循环。

什么情况下使用自旋锁:

自旋锁非常有助于避免阻塞,但是如果预期有大量阻塞,由于旋转过多,您可能不应该使用自旋锁。当锁是细粒度的并且数量巨大(例如链接的列表中每个节点一个锁)时以及锁保持时间总是非常短时,旋转可能非常有帮助。

短时间锁定的情况下,自旋锁(spinlock)更快。(因为自旋锁本质上不会让线程休眠,而是一直循环尝试对资源访问,直到可用。所以自旋锁线程被阻塞时,不进行线程上下文切换,而是空转等待。对于多核CPU而言,减少了切换线程上下文的开销,从而提高了性能。)如果机器单核或锁定时间长的要避免使用,因为占有着逻辑核心会导致其他的线程也不可用。

SpinLock和Lock的区别:
SpinLock,自旋锁。尝试获取该锁的线程持续不断的check是否可以获得。此时线程仍然是激活状态,只是在空转,浪费cpu而已。但是spinlock避免了线程调度和上下文切换,如果锁的时间极短的话,使用该锁反而效率会高。
而lock是线程被block了。这将引起线程调度和上下文切换等行为。

示例:

//创建自旋锁
        private static SpinLock spin = new SpinLock();
        public static void Spinklock()
        {
            Action action = () =>
            {
                bool lockTaken = false;
                try
                {
                    //申请获取锁
                    spin.Enter(ref lockTaken);
                    //临界区
                    for (int i = 0; i < 10; i++)
                    {
                        Console.WriteLine($"当前线程{Thread.CurrentThread.ManagedThreadId.ToString()},输出:1");
                    }
                }
                finally
                {
                    //工作完毕,或者产生异常时,检测一下当前线程是否占有锁,如果有了锁释放它
                    //避免出行死锁
                    if(lockTaken)
                    {
                        spin.Exit();
                    }
                }
            };
            Action action2 = () =>
            {
                bool lockTaken = false;
                try
                {
                    //申请获取锁
                    spin.Enter(ref lockTaken);
                    //临界区
                    for (int i = 0; i < 10; i++)
                    {
                        Console.WriteLine($"当前线程{Thread.CurrentThread.ManagedThreadId.ToString()},输出:2");
                    }

                }
                finally
                {
                    //工作完毕,或者产生异常时,检测一下当前线程是否占有锁,如果有了锁释放它
                    //避免出行死锁
                    if (lockTaken)
                    {
                        spin.Exit();
                    }
                }

            };
            //并行执行2个action
            Parallel.Invoke(action, action2);

        }

结果:

申请锁下面的临界区保证是顺序执行的,不会因为多线程穿插输出。

4.读写锁ReaderWriterLockSlim

  • 读写锁是一个具有特殊用途的线程锁,适用于频繁读取且读取需要一定时间的场景,共享资源的读取操作通常是可以同时执行的,
  • 普通的互斥锁不管是获取还是修改操作无法同时执行,如果多个线程为了读取操作而获取互斥锁,那么同一时间只有一个线程可以执行读取操作,
  • 频繁读取的场景下会对吞吐量造成影响
  • 读写锁把锁分为读取锁和写入锁,线程可以根据对共享资源的操作类型获取读取锁还是写入锁,读取锁可以被多个线程同时获取,写入锁不可以被多个线程同时获取,且读取锁和写入锁不可以被不同的线同时获取。
操作 读取锁状态 写入锁状态 获取锁是否需要等待
获取读取锁 未被获取 未被获取 无需等待
获取读取锁 已被其他线程获取 未获取 无需等待
获取读取锁 未被获取 已被其他线程获取 需要等待其他线程释放
获取写入锁 未被获取 未被获取 无需等待
获取写入锁 已被其他线程获取 未被获取 需要等待其他线程释放
获取写入锁 未被获取 已被其他线程获取 需要等待其他线程释放

代码示例:

//读写锁, //策略支持递归
        private static ReaderWriterLockSlim rwl = new ReaderWriterLockSlim(LockRecursionPolicy.SupportsRecursion);
        private static int index = 0;
        static void read()
        {
            try
            {
                //进入读锁
                rwl.EnterReadLock();
                for (int i = 0; i < 5; i++)
                {
                    Console.WriteLine($"线程id:{Thread.CurrentThread.ManagedThreadId},读数据,读到index:{index}");
                }
            }
            finally
            {
                //退出读锁
                rwl.ExitReadLock();
            }
        }
        static void write()
        {
            try
            {
                //尝试获写锁
                while (!rwl.TryEnterWriteLock(50))
                {
                    Console.WriteLine($"线程id:{Thread.CurrentThread.ManagedThreadId},等待写锁");
                }
                Console.WriteLine($"线程id:{Thread.CurrentThread.ManagedThreadId},获取到写锁");
                for (int i = 0; i < 5; i++)
                {
                    index++;
                    Thread.Sleep(50);
                }
                Console.WriteLine($"线程id:{Thread.CurrentThread.ManagedThreadId},写操作完成");
            }
            finally
            {
                //退出写锁
                rwl.ExitWriteLock();
            }
        }

        /// <summary>
        /// 执行多线程读写
        /// </summary>
        public static void test()
        {
            var taskFactory = new TaskFactory(TaskCreationOptions.LongRunning, TaskContinuationOptions.None);
            Task[] task = new Task[6];
            task[1] = taskFactory.StartNew(write); //写
            task[0] = taskFactory.StartNew(read); //读
            task[2] = taskFactory.StartNew(read); //读
            task[3] = taskFactory.StartNew(write); //写
            task[4] = taskFactory.StartNew(read); //读
            task[5] = taskFactory.StartNew(read); //读

            for (var i=0; i<6; i++)
            {
                task[i].Wait();
            }

        }

可以看到,在线程4写入期间,线程7是等待线程4写万再写,线程7写期间其它线程也没有操场,等写完后,读的操作是多个线程交叉的。

适合的场景举例:

  • 多线程写文件,多线程并发写文件时,会报资源被占用错误,用这里的写锁就可以独占资源写完再到下一个线程写。
  • 本地缓存的读写操作,几个缓存值写完才能读出来,防止读到不完整数据。

二、线程安全

1.线程安全集合

BlockingCollection:一个支持界限和阻塞的容器(线程安全集合),与队列结构相似,常用函数如下

Add :向容器中插入元素

TryTake:从容器中取出元素并删除

TryPeek:从容器中取出元素,但不删除。

CompleteAdding:告诉容器,添加元素完成。此时如果还想继续添加会发生异常。

IsCompleted:告诉消费线程,生产者线程还在继续运行中,任务还未完成。

普通用法示例:

/// <summary>
        /// 线程安全集合用法
        /// </summary>
        public static void BC()
        {
            //线程安全集合
            using (BlockingCollection<int> blocking = new BlockingCollection<int>())
            {
                int NUMITEMS = 10000;

                for (int i = 1; i < NUMITEMS; i++)
                {
                    blocking.Add(i);
                }
                //完成添加
                blocking.CompleteAdding();

                int outerSum = 0;

                // 定义一个委托方法取出集合元素
                Action action = () =>
                {
                    int localItem;
                    int localSum = 0;

                    //取出并删除元素,先进先出
                    while (blocking.TryTake(out localItem))
                    {
                        localSum += localItem;
                    }
                    //两数相加替换第一个值
                    Interlocked.Add(ref outerSum, localSum);
                };

                //并行3个线程执行,多个线程同时取集合的数据
                Parallel.Invoke(action, action, action);

                Console.WriteLine($"0+...{NUMITEMS-1} = {((NUMITEMS * (NUMITEMS - 1)) / 2)},输出结果:{outerSum}");
                //此集合是否已标记为已完成添加且为空
                Console.WriteLine($"线程安全集合.IsCompleted={blocking.IsCompleted}");
            }
        }

结果:

限制集合长度示例:

/// <summary>
        /// 限制集合长度
        /// </summary>
        public static void BCLimtLength()
        {
            //限制集合长度为5个,后面进的会阻塞等集合少于5个再进来
            BlockingCollection<int> blocking = new BlockingCollection<int>(5);

            var task1= Task.Run(() =>
            {
                for (int i = 0; i < 20; i++)
                {
                    blocking.Add(i);
                    Console.WriteLine($"集合添加:({i})");
                }

                blocking.CompleteAdding();
                Console.WriteLine("完成添加");
            });

            // 延迟500ms执行等待先生产数据
            var task2 = Task.Delay(500).ContinueWith((t) =>
            {
                while (!blocking.IsCompleted)
                {
                    var n = 0;
                    if (blocking.TryTake(out n))
                    {
                        Console.WriteLine($"取出:({n})");
                    }
                }

                Console.WriteLine("IsCompleted = true");
            });

            Task.WaitAll(task1, task2);
        }

结果:

在BlockingCollection中使用Stack(栈,先进后出)示例:

/// <summary>
        /// 线程安全集合,先进后出
        /// </summary>
        public static void BCStack()
        {
            //线程安全集合,参数表明栈标识,队列长度为5
            BlockingCollection<int> blocking = new BlockingCollection<int>(new ConcurrentStack<int>(), 5);

            var task1 = Task.Run(() =>
            {
                for (int i = 0; i < 20; i++)
                {
                    blocking.Add(i);
                    Console.WriteLine($"集合添加:({i})");
                }

                blocking.CompleteAdding();
                Console.WriteLine("完成添加");
            });

            // 等待先生产数据
            var task2 = Task.Delay(500).ContinueWith((t) =>
            {
                while (!blocking.IsCompleted)
                {
                    var n = 0;
                    if (blocking.TryTake(out n))
                    {
                        Console.WriteLine($"取出:({n})");
                    }
                }

                Console.WriteLine("IsCompleted = true");
            });

            Task.WaitAll(task1, task2);
        }

一开始入了0-4,从最后的4开始取。

2.线程安全字典

ConcurrentDictionary :这个比较好理解,普通字典多线程并发时添加时会报错,而这个则是线程安全的,不会报错。

普通字典示例:

//普通字典
        private static IDictionary<string, string> Dictionaries { get; set; } = new Dictionary<string, string>();
      /// <summary>
        /// 字典增加值
        /// </summary>
       public static void AddDictionaries()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            //并发1000个线程写
            Parallel.For(0, 1000, (i) =>
            {
                var key = $"key-{i}";
                var value = $"value-{i}";

                // 不加锁会报错
               // lock (Dictionaries)
               // {
                    Dictionaries.Add(key, value);
               // }
            });
            sw.Stop();
            Console.WriteLine("Dictionaries 当前数据量为: {0}", Dictionaries.Count);
            Console.WriteLine("Dictionaries 执行时间为: {0} ms", sw.ElapsedMilliseconds);
        }

不加锁时结果:

加锁后:

线程安全字典示例:

//线程安全字典
        private static IDictionary<string, string> ConcurrentDictionaries { get; set; } = new ConcurrentDictionary<string, string>();

      /// <summary>
        /// 线程安全字典添加值
        /// </summary>
        public static void AddConcurrentDictionaries()
        {
            Stopwatch sw = new Stopwatch();
            sw.Start();
            //并发1000个线程写
            Parallel.For(0, 1000, (i) =>
            {
                var key = $"key-{i}";
                var value = $"value-{i}";

                // 无须加锁
                ConcurrentDictionaries.Add(key, value);

            });
            sw.Stop();
            Console.WriteLine("ConcurrentDictionaries 当前数据量为: {0}", ConcurrentDictionaries.Count);
            Console.WriteLine("ConcurrentDictionaries 执行时间为: {0} ms", sw.ElapsedMilliseconds);
        }

可以看到,线程安全字典比普通字典性能略好,且线程安全字典无需加锁。

三、线程池

1.通过QueueUserWorkItem启动工作者线程

ThreadPool线程池中有两个重载的静态方法可以直接启动工作者线程:

  • ThreadPool.QueueUserWorkItem(waitCallback);
  • ThreadPool.QueueUserWorkItem(waitCallback,Object);

先把WaitCallback委托指向一个带有Object参数的无返回值方法,再使用ThreadPool.QueueUserWorkItem(WaitCallback)就可以一步启动此方法,此时异步方法的参数被视为null。

示例1:

public class ThreadLoopDemo
    {

        /// <summary>
        /// 回调方法
        /// </summary>
        /// <param name="obj"></param>
        static void CallMethod(object state)
        {
            Console.WriteLine("RunWorkerThread开始工作");
            Console.WriteLine("工作者线程启动成功!");
        }

        public static void Test()
        {
          //工作者线程最大数目,I/O线程的最大数目
            ThreadPool.SetMaxThreads(1000, 1000);
            //启动工作者线程
            ThreadPool.QueueUserWorkItem(new WaitCallback(CallMethod!));
            Console.ReadKey();
        }
    }

执行Test方法,结果:

示例2:

使用第二个重载方法ThreadPool.QueueUserWorkItem(WaitCallback,object)方法可以把object对象作为参数传送到回调函数中。

public class ThreadLoopDemo
    {
        /// <summary>
        /// 回调方法
        /// </summary>
        /// <param name="obj"></param>
        static void CallMethod(object state)
        {
            Console.WriteLine("RunWorkerThread开始工作");
            Order order=state as Order;
            Console.WriteLine($"orderName:{order.orderName},price:{order.price}");
            Console.WriteLine("工作者线程启动成功!");
        }

        public static void Test()
        {
            //工作者线程最大数目,I/O线程的最大数目
            ThreadPool.SetMaxThreads(1000, 1000);
            Order order = new Order()
            {
                orderName = "冰箱",
                price = 1888
            };
            //启动工作者线程
            ThreadPool.QueueUserWorkItem(new WaitCallback(CallMethod!),order);
            Console.ReadKey();
        }

    }
    public class Order
    {
        public string orderName { get; set; }
        public decimal price { get; set; }
    }

执行Test方法,结果:

通过ThreadPool.QueueUserWork启动工作者线程非常方便,但是WaitCallback委托指向的必须是一个带有object参数的无返回值方法。

线程池还可以重用线程,比喻可以吧线程池大小设为5个,去执行一批任务,防止大量创建新线程损耗大量cpu。

所以这个方法启动的工作者线程仅仅适合于带单个参数和无返回值的情况。

2.线程池等待(信号量)

ThreadPool并没有Thread的Join等待接口,那么想让ThreadPool等待要这么做呢?

ManualResetEvent:通知一个或多个正在等待的线程已发生的事件,相当于发送了一个信号

mre.WaitOne:卡住当前主线程,一直等到信号修改为true的时候,才会接着往下跑

public class ThreadLoopDemo
    {
        /// <summary>
        /// 执行
        /// </summary>
        public static void Test()
        {
            //用来控制线程等待,false默认为关闭状态
            ManualResetEvent mre = new ManualResetEvent(false);
            ThreadPool.QueueUserWorkItem(p =>
            {

                Console.WriteLine("线程1开始");
                Thread.Sleep(1000);
                Console.WriteLine($"线程1结束,带参数,{Thread.CurrentThread.ManagedThreadId}");
                //通知线程,修改信号为true
                mre.Set();
            });

            //阻止当前线程,直到等到信号为true在继续执行
            mre.WaitOne();

            //关闭线程,相当于设置成false
            mre.Reset();
            Console.WriteLine("信号被关闭了");

            ThreadPool.QueueUserWorkItem(p =>
            {
                Console.WriteLine("线程2开始");
                Thread.Sleep(2000);
                Console.WriteLine("线程2结束");
                mre.Set();

            });

            mre.WaitOne();
            Console.WriteLine("主线程结束");

        }
    }

执行Test,结果:

3.Task

Task.Factory.StartNew:创建一个新的线程,Task的线程也是从线程池中拿的(ThreadPool),官方建议用Task而不是直接使用ThreadPool,因为Task是对ThreadPool的封装和优化。

Task.WaitAny:等待一群线程中的其中一个完成,这里是卡主线程,一直等到一群线程中的最快的一个完成,才能继续往下执行,打个简单的比方:从三个地方获取配置信息(数据库,config,IO),同时开启三个线程来访问,谁快谁来执行。

Task.WaitAll:等待所有线程完成,这里也是卡主线程,一直等待所有子线程完成任务,才能继续往下执行。

Task.WhenAll:等待所有线程完成,这里不卡主线程,一直等待所有子线程完成任务,才能继续往下执行。

Task.ContinueWhenAny:回调形式的,任意一个线程完成后执行的后续动作,这个就跟WaitAny差不多,只不是是回调形式的。

Task.ContinueWhenAll:回调形式的,所有线程完成后执行的后续动作,理解同上

代码示例:

public class TaskDemo
    {
        /// <summary>
        /// 一个比较耗时的方法,循环1000W次
        /// </summary>
        /// <param name="name"></param>
        public static void DoSomething(string name)
        {
            int iResult = 0;
            for (int i = 0; i < 1000000000; i++)
            {
                iResult += i;
            }

            Console.WriteLine($"{name},线程Id:{Thread.CurrentThread.ManagedThreadId},{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}"+Environment.NewLine);
        }

        public static void Test()
        {
            //线程容器
            List<Task> taskList = new List<Task>();
            Stopwatch watch = new Stopwatch();
            watch.Start();
            Console.WriteLine("************任务开始**************");

            //启动5个线程
            for (int i = 0; i < 5; i++)
            {
                string name = $"Task:{i}";
                Task task = Task.Factory.StartNew(() =>
                {
                    DoSomething(name);
                });

                taskList.Add(task);
            }

            //回调形式的,任意一个完成后执行的后续动作
            Task any = Task.Factory.ContinueWhenAny(taskList.ToArray(), task =>
            {
                Console.WriteLine($"ContinueWhenAny,当前线程Id:{Thread.CurrentThread.ManagedThreadId},一个线程执行完的回调,继续执行后面动作,{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}" + Environment.NewLine);
            });

            //回调形式的,全部任务完成后执行的后续动作
            Task all = Task.Factory.ContinueWhenAll(taskList.ToArray(), tasks =>
            {
                Console.WriteLine($"ContinueWhenAll,当前线程Id:{Thread.CurrentThread.ManagedThreadId},全部线程执行完的回调,线程数:{tasks.Length},{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}" + Environment.NewLine);
            });

            //把两个回调也放到容器里面,包含回调一起等待
            taskList.Add(any);
            taskList.Add(all);

            //WaitAny:线程等待,当前线程等待其中一个线程完成
            Task.WaitAny(taskList.ToArray());
            Console.WriteLine($"WaitAny,当前线程Id:{Thread.CurrentThread.ManagedThreadId},其中一个完成已完成,{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}" + Environment.NewLine);

            //WaitAll:线程等待,当前线程等待所有线程的完成
            Task.WaitAll(taskList.ToArray());
            Console.WriteLine($"WaitAll,当前线程Id:{Thread.CurrentThread.ManagedThreadId},全部线程完成,{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}" + Environment.NewLine);

            Console.WriteLine($"************任务结束**************耗时:{watch.ElapsedMilliseconds}ms,{Thread.CurrentThread.ManagedThreadId},{DateTime.Now}");
        }
    }

结果:

4.线程池调度原理

根据 TaskCreationOptions 的不同,出现了三个分支

LongRunning:独立线程,和线程池无关包含 PreferFairness时:preferLocal=false,进入全局队列不包含 PreferFairness时:preferLocal=ture,进入本地队列

进入全局队列的任务能够公平地被各个线程池中的线程领取执行。

下图中 Task1先进入全局队列,随后被 Thread1 领走。Thread3 通过 WorkStealing 机制窃取了 Thread2 中的 Task3。

1、线程任务1先去到线程池调度队列。

2、线程池队列根据参数去到公共队列或线程的本地队列。

3、线程池线程1从公共线程池中取Task1到本地队列执行

4、Thead3发现自己的队列为空,公共队列也为空,会从其它的线程中窃取任务执行。

四、并行

Parallel:是并行编程,在Task的基础上做了封装,.NET FrameWork 4.5之后的版本可用,调用Parallel线程参与执行任务。

与Task区别: 使用Task开启子线程的时候,主线程是属于空闲状态,并不参与执行;Parallel开启子线程的时候,主线程也会参与计算

示例1:

/// <summary>
        /// 一个比较耗时的方法,循环1000W次
        /// </summary>
        /// <param name="name"></param>
        public static void DoSomething(string name)
        {
            int iResult = 0;
            for (int i = 0; i < 1000000000; i++)
            {
                iResult += i;
            }

            Console.WriteLine($"{name},线程Id:{Thread.CurrentThread.ManagedThreadId},{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss ffff")}" + Environment.NewLine);
        }

        public static void Test1()
        {
            //并行编程
            Console.WriteLine($"并行编程开始,主线程Id:{Thread.CurrentThread.ManagedThreadId}");
            Console.WriteLine("【示例1】");
            //示例1:
            //一次性执行1个或多个线程,效果类似:Task WaitAll,只不过Parallel的主线程也参与了计算
            Parallel.Invoke(
                () => { DoSomething("并行1-1"); },
                () => { DoSomething("并行1-2"); },
                () => { DoSomething("并行1-3"); },
                () => { DoSomething("并行1-4"); },
                () => { DoSomething("并行1-5"); });
            Console.WriteLine("*************并行结束************");
            Console.ReadLine();
        }

执行Test1结果:

示例2:

public static void Test2()
        {
            //并行编程
            Console.WriteLine($"并行编程开始,主线程Id:{Thread.CurrentThread.ManagedThreadId}");
            Console.WriteLine("【示例2】");
            //示例2:
            //定义要执行的线程数量
            Parallel.For(0, 5, t =>
            {
                int a = t;
                DoSomething($"并行2-{a}");

            });
            Console.WriteLine("*************并行结束************");
            Console.ReadLine();
        }

结果:

示例3:

public static void Test3()
        {
            //并行编程
            Console.WriteLine($"并行编程开始,主线程Id:{Thread.CurrentThread.ManagedThreadId}");
            Console.WriteLine("【示例3】");
            ParallelOptions options = new ParallelOptions()
            {
                MaxDegreeOfParallelism = 3//执行线程的最大并发数量,执行完成一个,就接着开启一个
            };

            //遍历集合,根据集合数量执行线程数量
            Parallel.ForEach(new List<string> { "a", "b", "c", "d", "e", "f", "g" }, options, (t, status) =>

            {
                //status.Break();//这一次结束。
                //status.Stop();//整个Parallel结束掉,Break和Stop不可以共存
                DoSomething($"并行4-{t}");
            });
        }

结果:分成3次并行

五、异步IO

1.异步IO于同步IO比较

异步IO在数据准备阶段不会阻塞主线程,而同步IO则会阻塞主线程。

2.异步读写文件

这里使用 FileStream 类,它带有一个参数 useAsync,可以避免在许多情况下阻塞线程池的线程。可以通过 useAsync = true 来进行启用或在构造函数中进行参数调用。

但是我们不能对 StreamReader 和 StreamWriter 中的参数进行设置。但是,如果你想使用该参数 useAsync,则需要自己新建一个 FileStream 对象。

请注意,异步调用是在 UI 中的,即使线程池线程阻塞,在 await 期间,用户界面线程也不会被阻塞。

异步写入文本

/// <summary>
        /// 异步写入文件
        /// </summary>
        /// <returns></returns>
        public async Task WriteTextAsync()
        {
            var path = "temp.txt"; //文件名
            var content = Guid.NewGuid().ToString(); //写入内容

            using (var fs = new FileStream(path, FileMode.OpenOrCreate, FileAccess.ReadWrite, FileShare.None, bufferSize: 4096, useAsync: true))
            {
                var buffer = Encoding.UTF8.GetBytes(content);
                await fs.WriteAsync(buffer, 0, buffer.Length);
            }
        }

执行完查看根目录文件结果:

异步读取文件

/// <summary>
        /// 异步读取文本
        /// </summary>
        /// <returns></returns>
        public static async Task ReadTextAsync()
        {
            var fileName = "temp.txt"; //文件名
            using (var fs = new FileStream(fileName,
                                           FileMode.OpenOrCreate, FileAccess.Read, FileShare.None, bufferSize: 4096, useAsync: true))
            {
                var sb = new StringBuilder();
                var buffer = new byte[4096];
                var readLength = 0;

                while ((readLength = await fs.ReadAsync(buffer, 0, buffer.Length)) != 0)
                {
                    var text = Encoding.UTF8.GetString(buffer, 0, readLength);
                    sb.Append(text);
                }

                Console.WriteLine("读取文件内容:"+sb.ToString());
            }
        }

执行结果:

到此这篇关于浅谈C#多线程下的调优的文章就介绍到这了,更多相关C#多线程调优内容请搜索我们以前的文章或继续浏览下面的相关文章希望大家以后多多支持我们!

(0)

相关推荐

  • C#简单多线程同步和优先权用法实例

    本文实例讲述了C#简单多线程同步和优先权用法.分享给大家供大家参考.具体分析如下: 本文实例题目如下: 麦当劳有两个做汉堡的厨师(工号:11,12)和三个销售人员(工号:21,22,23). 厨师生产汉堡,并负责将做好的汉堡放入货架,货架台大小有限,最多放6个汉堡,11和12不能同时往货架台上放汉堡,11具有优先权. 销售人员负责销售食品,三个销售人员取食品时,货架不能为空,三人不能同时取,23优先权最高,21最低.21卖的最快,取得频率最高,22次之. 一天的工作量是销售70个汉堡. 这里先来

  • 浅谈C#多线程下的调优

    目录 一.原子操作 1.基于Lock实现 2.基于CAS实现 3.自旋锁SpinLock 4.读写锁ReaderWriterLockSlim 二.线程安全 1.线程安全集合 2.线程安全字典 三.线程池 1.通过QueueUserWorkItem启动工作者线程 2.线程池等待(信号量) 3.Task 4.线程池调度原理 四.并行 五.异步IO 1.异步IO于同步IO比较 2.异步读写文件 一.原子操作 先看一段问题代码 /// <summary> /// 获取自增 /// </summa

  • 浅谈Linux环境下gcc优化级别

    代码优化可以说是一个非常复杂而又非常重要的问题,以笔者多年的linux c开发经验来说优化通常分为两个方面,一是人为优化,也就是基于编程经验采用更简易的数据结构函数等来降低编译器负担,二是采用系统自带的优化模式,也就是gcc - o系列,下面我将简述一下各级优化的过程以及实现. gcc - o1 首先o1上面还有一个o0,那个是不提供任何优化,项目中几乎不会使用,而o1使用就非常广泛了,o1是最基本的优化,主要对代码的分支,表达式,常量来进行优化,编译器会在较短的时间下将代码变得更加短小,这样体

  • 浅谈Java多线程处理中Future的妙用(附源码)

    java 中Future是一个未来对象,里面保存这线程处理结果,它像一个提货凭证,拿着它你可以随时去提取结果.在两种情况下,离开Future几乎很难办.一种情况是拆分订单,比如你的应用收到一个批量订单,此时如果要求最快的处理订单,那么需要并发处理,并发的结果如果收集,这个问题如果自己去编程将非常繁琐,此时可以使用CompletionService解决这个问题.CompletionService将Future收集到一个队列里,可以按结果处理完成的先后顺序进队.另外一种情况是,如果你需要并发去查询一

  • 浅谈Java多线程的优点及代码示例

    尽管面临很多挑战,多线程有一些优点使得它一直被使用.这些优点是: 资源利用率更好 程序设计在某些情况下更简单 程序响应更快 资源利用率更好 想象一下,一个应用程序需要从本地文件系统中读取和处理文件的情景.比方说,从磁盘读取一个文件需要5秒,处理一个文件需要2秒.处理两个文件则需要: 5秒读取文件A 2秒处理文件A 5秒读取文件B 2秒处理文件B --------------------- 总共需要14秒 从磁盘中读取文件的时候,大部分的CPU时间用于等待磁盘去读取数据.在这段时间里,CPU非常的

  • 浅谈java多线程编程

    一.多线程的优缺点 多线程的优点: 1)资源利用率更好 2)程序设计在某些情况下更简单 3)程序响应更快 多线程的代价: 1)设计更复杂 虽然有一些多线程应用程序比单线程的应用程序要简单,但其他的一般都更复杂.在多线程访问共享数据的时候,这部分代码需要特别的注意.线程之间的交互往往非常复杂.不正确的线程同步产生的错误非常难以被发现,并且重现以修复. 2)上下文切换的开销 当CPU从执行一个线程切换到执行另外一个线程的时候,它需要先存储当前线程的本地的数据,程序指针等,然后载入另一个线程的本地数据

  • 浅谈Java多线程实现及同步互斥通讯

    Java多线程深入理解本文主要从三个方面了解和掌握多线程: 1. 多线程的实现方式,通过继承Thread类和通过实现Runnable接口的方式以及异同点. 2. 多线程的同步与互斥中synchronized的使用方法. 3. 多线程的通讯中的notify(),notifyAll(),及wait(),的使用方法,以及简单的生成者和消费者的代码实现. 下面来具体的讲解Java中的多线程: 一:多线程的实现方式 通过继承Threa类来实现多线程主要分为以下三步: 第一步:继承 Thread,实现Thr

  • 浅谈java多线程 join方法以及优先级方法

    join: 当A线程执行到了B线程的.join()方法时,A就会等待.等B线程都执行完,A才会执行. join可以用来临时加入线程执行. 1.线程使用join方法,主线程就停下,等它执行完,那么如果该线程冻结了,主线程就挂了,这也是为什么线程要抛异常的原因 2.当两个或以上线程开启了,这个A线程才使用join方法,那么主线程还是停下,这几个个线程交替进行,直到A执行完,主线程才复活 1. tostring(),方法,获取线程具体的名字,优先级 2. 优先级代表抢资源的频率 3. java中设置有

  • 浅谈find命令下的atime,ctime,mtime

    Linux下的find命令在目录结构中搜索文件,并执行指定的操作.linux下的find命令提供了相当多的查找条件,功能很强大,由于find的功能很强大,所以他的选项也很多,今天我们来细说一下find命令下的三个时间参数atime,ctime,mtime. atime:access time,最近一次访问时间,当使用这个文件的时候就会更新这个时间. mtime:modify time,最近一次文件内容更改时间,当修改文件内容数据时,就会更改这个时间. ctime:change time,最近一次

  • 浅谈Java向下转型的意义

    一开始学习 Java 时不重视向下转型.一直搞不清楚向下转型的意义和用途,不清楚其实就是不会,那开发的过程肯定也想不到用向下转型. 其实向上转型和向下转型都是很重要的,可能我们平时见向上转型多一点,向上转型也比较好理解. 但是向下转型,会不会觉得很傻,我是要用子类实例对象,先是生成子类实例赋值给父类引用,在将父类引用向下强转给子类 引用,这不是多此一举吗?我不向上转型也不向下转型,直接用子类实例就行了. 我开始学习Java时也是这么想的,这误区导致我觉得向下转型就是没用的. 随着技术的提升,我在

  • 浅谈在JupyterNotebook下导入自己的模块的问题

    在jupyternotebook下导入自己写的模块,有两点需要注意: 1.要将自己写的模块编程xxx.py的形式,而不是.ipynb文件 2.当更改自己的模块的内容后,要Restart内核,才能反映到使用该模块的.ipynb文件中. 补充知识:Jupyter 重新导入修改后的自定义包 Jupyter 经常遇到这样一个问题,就是在已有的 notebook 中导入了自定义的 itools.py 包文件,但是在编辑 notebook 中发现需要对 itools.py 文件某些函数进行修改,但是保存后再

随机推荐