zoukankan      html  css  js  c++  java
  • Multithreading With C# Cookbook---Chapter4---使用任务并行库

    概念

    任务并行库(Task Parallel Library,TPL),在.NET Framework 4.0引入,4.5有所改进。我们可以把线程池认为是一个抽象层,其隐藏使用线程的细节;TPL可被认为是线程池上的又一个抽象层,其隐藏了与线程池交互的底层代码,并提供更方便的API。

    更多内容

    TPL核心概念是任务。一个任务代表一个异步操作,该操作可以通过多种方式运行,可以使用或不使用独立线程运行。TPL向用户隐藏任务的实现细节从而创建一个抽象层,我们无须知道任务实际是如何执行的。但如果不正确使用,将会导致诡异的错误。

    一个任务可以通过多种方式与其他任务组合,其用于组合任务的便利的API也是一个与其他模式(APM,EAP)的一个关键优势。

    处理任务中的异常:一个任务可能有多个任务组成,多个任务又有自己的子任务,通过AggregateException来捕获底层任务的所有异常,并允许单独处理这些异常。

    准备工作

    使用.NET Framework 4.5以上版本。

    使用using

    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Text;
    using System.Threading.Tasks;
    using static System.Threading.Thread;
    using System.Threading;
    View Code

    创建任务

    Task.Run创建任务,再使用Start开启任务;Task.Factory.StartNew创建任务后会立即启动。当创建任务时标记任务为长时间运行TaskCreationOptions.LongRunning,将不会使用线程池线程,而是另开线程。

        class Program
        {
            static void Main(string[] args)
            {
                Task t1 = new Task(() => TaskMethod("Task1"));
                Task t2 = new Task(() => TaskMethod("Task2"));
                t1.Start();
                t2.Start();
                Task.Run(() => TaskMethod("Task3"));
                Task.Factory.StartNew(() => TaskMethod("Task4"));
                Task.Factory.StartNew(() => TaskMethod("Task5"), TaskCreationOptions.LongRunning);//标记该任务为长时间运行任务,将不会使用线程池线程
    
                Console.ReadLine();
            }
    
            static void TaskMethod(string name)
            {
                Console.WriteLine($"Task {name} is running on a thread id: {CurrentThread.ManagedThreadId}.Is thread pool thread: {CurrentThread.IsThreadPoolThread}");
            }
        }
    View Code

    使用任务执行基本操作

    创建Task<int>任务,使用任务对象.Result属性来获取该任务返回值,在在任务返回值之前,主线程会阻塞等待完成。使用Task.IsCompleted来轮询任务状态,通过Task.RunAsynchornously()方法,来指定任务运行在主线程。

    class Program
        {
            static void Main(string[] args)
            {
                Console.WriteLine("Task1 result:"+ TaskMethod("Task1")); 
    
                Task<int> task = CreatTask("Task2");
                task.Start();
                Console.WriteLine("Task2 result:" + task.Result);
    
                task = CreatTask("Task3");
                task.RunSynchronously();//运行在主线程
                Console.WriteLine("Task3 result:" + task.Result);
    
                task = CreatTask("Task4");
                Console.WriteLine(task.Status) ;
                task.Start();
                while (!task.IsCompleted)
                {
                    Console.WriteLine(task.Status);
                    Thread.Sleep(TimeSpan.FromSeconds(1));
                }
                Console.WriteLine(task.Status);
                Console.WriteLine("Task4 result:" + task.Result);
    
                Console.ReadLine();
            }
    
            static Task<int> CreatTask(string name)
            {
                return new Task<int>(()=>TaskMethod(name));
            }
    
            static int TaskMethod(string name)
            {
                Console.WriteLine($"Task {name} is running on a thread id: {CurrentThread.ManagedThreadId}.Is thread pool thread: {CurrentThread.IsThreadPoolThread}");
                Thread.Sleep(TimeSpan.FromSeconds(2));
                return 42;
            }
        }
    View Code

    组合任务

    创建一个任务,再给他创建一个后续任务(Task.ContinueWith),先执行该任务,完成后才能执行后续任务。TaskContinuationOptions.OnlyOnRanToCompletion属性指定该延续任务只能在前面任务完成后才能执行;TaskContinuationOptions.ExecuteSynchronously属性指定该延续任务会工作在创建此延续任务的线程上执行。

    创建任务时指定TaskCreationOptions.AttachedToParent属性,指定该任务为父子任务,只有当所有子任务完成,父任务才算完成。

    class Program
        {
            static void Main(string[] args)
            {
                var t1 = new Task<int>(() => TaskMethod("task1", 3));
                var t2 = new Task<int>(() => TaskMethod("task2", 2));
    
                t1.ContinueWith(t => Console.WriteLine($"The first answer is {t.Result}. Thread id :{CurrentThread.ManagedThreadId},is thread pool thread:{CurrentThread.IsThreadPoolThread}"), TaskContinuationOptions.OnlyOnRanToCompletion);//该属性指定该延续任务只能在前面任务完成后才能执行
    
                t1.Start();
                t2.Start();
    
                Thread.Sleep(TimeSpan.FromSeconds(4));
    
                Task continuation = t2.ContinueWith(t => Console.WriteLine($"The first answer is {t.Result}. Thread id :{CurrentThread.ManagedThreadId},is thread pool thread:{CurrentThread.IsThreadPoolThread}"), TaskContinuationOptions.OnlyOnRanToCompletion | TaskContinuationOptions.ExecuteSynchronously);//ExecuteSynchronously属性指定该延续任务会工作在创建此延续任务的线程上执行
    
                continuation.GetAwaiter().OnCompleted(() => Console.WriteLine($"Continuation Task Completed! Thread id is: {CurrentThread.ManagedThreadId}. Is thread pool thread: {CurrentThread.IsThreadPoolThread}"));
    
                Thread.Sleep(TimeSpan.FromSeconds(2));
                Console.WriteLine();
    
                t1 = new Task<int>(() =>
                {
                    var innerTask = Task.Factory.StartNew(() => TaskMethod("second Task", 5), TaskCreationOptions.AttachedToParent);
                    innerTask.ContinueWith(t => TaskMethod("Third Task", 2), TaskContinuationOptions.AttachedToParent);
                    return TaskMethod("First Task", 2);
                });
    
                t1.Start();
    
                while (!t1.IsCompleted)
                {
                    Console.WriteLine(t1.Status);
                    Thread.Sleep(TimeSpan.FromSeconds(0.5));
                }
                Console.WriteLine(t1.Status);
    
                Console.ReadLine();
    
    
    
                Console.ReadLine();
            }
    
            static int TaskMethod(string name, int seconds)
            {
                Console.WriteLine($"Task {name} is running on a thread id: {CurrentThread.ManagedThreadId}.Is thread pool thread: {CurrentThread.IsThreadPoolThread}");
                Thread.Sleep(TimeSpan.FromSeconds(seconds));
                return seconds * 42;
            }
        }
    View Code

    APM(异步编程)模式转任务

     将APM API转换为任务。代码示例Task.Factory.FromAsync三种重载方法。

    class Program
        {
            static void Main(string[] args)
            {
                int threadId;
                AsynchronouTask a = Test;
                IncompatibleAsynchronousTask c = Test;
    
                Console.WriteLine("Option1");
                Task<string> task1 = Task<string>.Factory.FromAsync(a.BeginInvoke("Async 1",callBack,"async1 state"),a.EndInvoke);
                task1.ContinueWith(t => Console.WriteLine($"Callback is finished. now running a continuation!result:{t.Result}"));
                while (!task1.IsCompleted)
                {
                    //Console.WriteLine(task1.Status);
                }
                Console.WriteLine(task1.Status);
    
                Console.WriteLine("----------------------------");
                Console.WriteLine();
    
                Console.WriteLine("Option2");
                Task<string> task2 = Task<string>.Factory.FromAsync(a.BeginInvoke,a.EndInvoke,"Async 2","async2 state");
                task2.ContinueWith(t => Console.WriteLine($"Callback is finished. now running a continuation!result:{t.Result}"));
                while (!task2.IsCompleted)
                {
                    //Console.WriteLine(task2.Status);
                }
                Console.WriteLine(task2.Status);
    
                Console.WriteLine("----------------------------");
                Console.WriteLine();
    
                Console.WriteLine("Option3");
                IAsyncResult ar = c.BeginInvoke(out threadId,callBack,"async3 state");
                Task<string> task3 = Task<string>.Factory.FromAsync(ar,_=>c.EndInvoke(out threadId,ar));
                task2.ContinueWith(t => Console.WriteLine($"Callback is finished. now running a continuation!result:{t.Result}"));
                while (!task3.IsCompleted)
                {
                    //Console.WriteLine(task3.Status);
                }
                Console.WriteLine(task3.Status);
    
                Console.ReadLine();
            }
    
            delegate string AsynchronouTask(string threadName);
    
            delegate string IncompatibleAsynchronousTask(out int threadId);
    
            static void callBack(IAsyncResult ar)
            {
                Console.WriteLine("Starting callback……");
                Console.WriteLine("State paseed to a callback:"+ar.AsyncState);
                Console.WriteLine("Is thread pool thread:"+CurrentThread.IsThreadPoolThread);
            }
    
            static string Test(string threadName)
            {
                Console.WriteLine("TN: Starting Test……");
                Console.WriteLine("Is thread pool thread:"+CurrentThread.IsThreadPoolThread);
                Sleep(2000);
                CurrentThread.Name = threadName;
                return "Thread name is:" + CurrentThread.Name;
             }
    
            static string Test(out int threadId)
            {
                Console.WriteLine("TI: Starting Test……");
                Console.WriteLine("Is thread pool thread:" + CurrentThread.ManagedThreadId);
                Sleep(2000);
                threadId = CurrentThread.ManagedThreadId;
                return "Thread Id is:" + threadId;
            }
        }
    View Code

    EAP(基于事件的异步操作)模式转任务

    class Program
        {
            static void Main(string[] args)
            {
                var tcs = new TaskCompletionSource<int>();
    
                var worker = new BackgroundWorker();
                worker.DoWork += (sender, evenArgs) =>
                  {
                      evenArgs.Result = TaskMethod("BackgroundWorker",3);
                  };
                worker.RunWorkerCompleted += (sender, evenArgs) =>
                  {
                      if (evenArgs.Error != null)
                      {
                          tcs.SetException(evenArgs.Error);
                      }
                      else if (evenArgs.Cancelled)
                      {
                          tcs.SetCanceled();
                      }
                      else
                      {
                          tcs.SetResult((int)evenArgs.Result);
                      }
                  };
                worker.RunWorkerAsync();
                int result = tcs.Task.Result;
                Console.WriteLine(result);
    
                Console.ReadLine();
            }
    
            static int TaskMethod(string name,int seconds)
            {
                Console.WriteLine($"Task {name} on a thread id:{CurrentThread.ManagedThreadId},is thread pool thread:{CurrentThread.IsThreadPoolThread}");
                Sleep(TimeSpan.FromSeconds(seconds));
                return seconds * 42;
            }     
        }
    View Code

    实现取消选项

    取消基于任务的异步操作。

    class Program
        {
            static void Main(string[] args)
            {
                var cts = new CancellationTokenSource();
                var longTask = new Task<int>(() =>
                    TaskMethod("Task1", 10, cts.Token), cts.Token);
                Console.WriteLine(longTask.Status);
                cts.Cancel();
                //longTask.Start();//异常
                Console.WriteLine(longTask.Status);
                Console.WriteLine("First task hai been cancelled before execution.");
    
                cts = new CancellationTokenSource();
                longTask = new Task<int>(() =>
                      TaskMethod("Task2", 10, cts.Token), cts.Token);
                longTask.Start();
                for (int i = 0; i < 5; i++)
                {
                    Thread.Sleep(TimeSpan.FromSeconds(0.5));
                    Console.WriteLine(longTask.Status);
                }
                cts.Cancel();
                Console.WriteLine("Task2 is canceled.");
                for (int i = 0; i < 5; i++)
                {
                    Thread.Sleep(TimeSpan.FromSeconds(0.5));
                    Console.WriteLine(longTask.Status);
                }
                Console.WriteLine($"A task has been completed with result {longTask.Result}");
                Console.ReadLine();
            }
    
            static int TaskMethod(string name, int seconds, CancellationToken token)
            {
                Console.WriteLine($"Task {name} is running on a thread id: {CurrentThread.ManagedThreadId}.Is thread pool thread: {CurrentThread.IsThreadPoolThread}");
    
                for (int i = 0; i < seconds; i++)
                {
                    Thread.Sleep(TimeSpan.FromSeconds(1));
                    if (token.IsCancellationRequested)
                        return -1;
                }
    
                return seconds * 42;
            }
        }
    View Code

    并行运行任务

    借助Task.WhenAll方法,创建第三个方法,该方法在所有任务完成之后执行,并返回一个结果集;使用Task.WhenAny方法,任务列表中每完成一个任务,就从列表删除并等待其他任务完成。可以使用一个任务来计时,超时取消其他所有任务,来模拟超时管理。

    class Program
        {
            static void Main(string[] args)
            {
                var t1 = new Task<int>(() => TaskMethod("Task1", 2));
                var t2 = new Task<int>(() => TaskMethod("Task2", 3));
                var whenAllTask = Task.WhenAll(t1, t2);
    
                whenAllTask.ContinueWith(t => Console.WriteLine($"The first task answer is {t.Result[0]}. The second task answer is {t.Result[1]}"), TaskContinuationOptions.OnlyOnRanToCompletion);
                t1.Start();
                t2.Start();
    
                Thread.Sleep(TimeSpan.FromSeconds(5));
    
                var tasks = new List<Task<int>>();
                for (int i = 0; i < 4; i++)
                {
                    int counter = i;
                    var task = new Task<int>(() => TaskMethod("Tasks:Task " + counter, counter));
                    tasks.Add(task);
                    task.Start();
                }
    
                while (tasks.Count > 0)
                {
                    var completedTask = Task.WhenAny(tasks).Result;
                    tasks.Remove(completedTask);
                    Console.WriteLine($"A task has been completed with result {completedTask.Result}");
                }
    
                Console.ReadLine();
            }
    
            static int TaskMethod(string name, int seconds)
            {
                Console.WriteLine($"Task {name} is running on a thread id: {CurrentThread.ManagedThreadId}.Is thread pool thread: {CurrentThread.IsThreadPoolThread}");
                Thread.Sleep(TimeSpan.FromSeconds(seconds));
                return seconds * 42;
            }
        }
    View Code

    UI访问线程

    第一个按钮Sync点击后整个界面卡死,一段时间后返回错误“调用线程无法访问此对象。另一个线程拥有该对象。”,因为我们不允许从创建UI的线程之外的线程访问UI;第二个按钮Async点击后,界面没有卡死,但依然会报错;第三个按钮AsyncOk点击后,可以成功返回结果,并且等待过程中,界面依然可以相应其他事件。

    WPF XAML:

    <Window x:Class="TaskSchedulerTest.MainWindow"
            xmlns="http://schemas.microsoft.com/winfx/2006/xaml/presentation"
            xmlns:x="http://schemas.microsoft.com/winfx/2006/xaml"
            xmlns:d="http://schemas.microsoft.com/expression/blend/2008"
            xmlns:mc="http://schemas.openxmlformats.org/markup-compatibility/2006"
            xmlns:local="clr-namespace:TaskSchedulerTest"
            mc:Ignorable="d"
            Title="MainWindow" Height="350" Width="525">
        <Grid>
            <TextBlock Name="ContentTextBlock" HorizontalAlignment="Left" Margin="44,134,0,0" VerticalAlignment="Top"
                       Width="425" Height="40"/>
            <Button Name="btnSync" Content="Sync" HorizontalAlignment="Left" Margin="45,190,0,0" VerticalAlignment="Top" Width="75" Click="btnSync_Click"/>
            <Button Name="btnAsync" Content="Async" HorizontalAlignment="Left" Margin="165,190,0,0" VerticalAlignment="Top" Width="75" Click="btnAsync_Click"/>
            <Button Name="btnAsyncOk" Content="AsyncOk" HorizontalAlignment="Left" Margin="285,190,0,0" VerticalAlignment="Top" Width="75" Click="btnAsyncOk_Click"/>
        </Grid>
    </Window>
    View Code

    后台程序:

    public partial class MainWindow : Window
        {
            public MainWindow()
            {
                InitializeComponent();
            }
    
            private void btnSync_Click(object sender, RoutedEventArgs e)
            {
                ContentTextBlock.Text = string.Empty;
                try
                {
                    string result = TaskMethod().Result;
                    ContentTextBlock.Text = result;
                }
                catch (Exception ex)
                {
    
                    ContentTextBlock.Text = ex.InnerException.Message;
                }
            }
    
            private void btnAsync_Click(object sender, RoutedEventArgs e)
            {
                ContentTextBlock.Text = string.Empty;
                Mouse.OverrideCursor = Cursors.Wait;
                Task<string> task = TaskMethod();
                task.ContinueWith(t=>
                {
                    ContentTextBlock.Text = t.Exception.InnerException.Message;
                    Mouse.OverrideCursor = null;
                }, CancellationToken.None,
                TaskContinuationOptions.OnlyOnFaulted,TaskScheduler.FromCurrentSynchronizationContext());
                
            }
    
            private void btnAsyncOk_Click(object sender, RoutedEventArgs e)
            {
                ContentTextBlock.Text = string.Empty;
                Mouse.OverrideCursor = Cursors.Wait;
                Task<string> task = TaskMethod(TaskScheduler.FromCurrentSynchronizationContext());
    
                task.ContinueWith(t=>Mouse.OverrideCursor=null,CancellationToken.None,TaskContinuationOptions.None,TaskScheduler.FromCurrentSynchronizationContext());
            }
    
            private Task<string> TaskMethod()
            {
                return TaskMethod(TaskScheduler.Default);
            }
            private Task<string> TaskMethod(TaskScheduler scheduler)
            {
                Task delay = Task.Delay(TimeSpan.FromSeconds(5));
    
                return delay.ContinueWith(d=>
                {
                    string str = $"Task is running on thread id :{CurrentThread.ManagedThreadId}. Is thread pool: {CurrentThread.IsThreadPoolThread}";
                    ContentTextBlock.Text = str;
                    return str;
                },scheduler);
            }
        }
    View Code

    注:本文是在阅读《C#多线程编程实战》后所写,部分内容引用该书内容,这是一本不错的书,感谢! 

  • 相关阅读:
    Struts2 采用Convention-plugin 实现零配置
    xml中批量新增数据或者批量修改数据
    Linux基础笔记_05
    Linux基础笔记_03
    Linux基础笔记_02
    Linux基础笔记_01
    玩转spring boot——websocket
    获取一天中的起始时间与结束时间
    《浅谈架构之路:前后端分离模式》
    区域语言与区域语言包对照关系表
  • 原文地址:https://www.cnblogs.com/EasonDongH/p/8487911.html
Copyright © 2011-2022 走看看