文档章节

C#综合揭秘——细说多线程(二)

深圳大道
 深圳大道
发布于 2016/12/29 15:37
字数 3132
阅读 44
收藏 0
/*
异步写入
FileStream中包含BeginWrite、EndWrite 方法可以启动I/O线程进行异步写入。
public override IAsyncResult BeginWrite ( byte[] array, int offset, int numBytes, AsyncCallback userCallback, Object stateObject )
public override void EndWrite (IAsyncResult asyncResult )
BeginWrite 返回值为IAsyncResult, 使用方式与委托的BeginInvoke方法相似,最好就是使用回调函数,避免线程阻塞。在最后两个参数中,参数AsyncCallback用于绑定回调函数; 参数Object用于传递外部数据。要注意一点:AsyncCallback所绑定的回调函数必须是带单个 IAsyncResult 参数的无返回值方法。
在例子中,把FileStream作为外部数据传递到回调函数当中,然后在回调函数中利用IAsyncResult.AsyncState获取FileStream对象,最后通过FileStream.EndWrite(IAsyncResult)结束写入。由输出结果可以看到,在使用FileStream.BeginWrite方法后,系统将自动启动CLR线程池中I/O线程。
*/

class Program
{
	static void Main(string[] args)
	{
		//把线程池的最大值设置为1000
		ThreadPool.SetMaxThreads(1000, 1000);
		ThreadPoolMessage("Start");

		//新立文件File.sour
		FileStream stream = new FileStream("File.sour", FileMode.OpenOrCreate, 
								   FileAccess.ReadWrite,FileShare.ReadWrite,1024,true);
		byte[] bytes = new byte[16384];
		string message = "An operating-system ThreadId has no fixed relationship........";
		bytes = Encoding.Unicode.GetBytes(message);

		//启动异步写入
		stream.BeginWrite(bytes, 0, (int)bytes.Length,new AsyncCallback(Callback),stream);
		stream.Flush();
		
		Console.ReadKey();
	}

	static void Callback(IAsyncResult result)
	{
		//显示线程池现状
		Thread.Sleep(200);
		ThreadPoolMessage("AsyncCallback");
		//结束异步写入
		FileStream stream = (FileStream)result.AsyncState;
		stream.EndWrite(result);
		stream.Close();
	}

	//显示线程池现状
	static void ThreadPoolMessage(string data)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
			  "WorkerThreads is:{2}  CompletionPortThreads is :{3}",
			  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
		Console.WriteLine(message);
	}
}

/*
异步读取
FileStream 中包含 BeginRead 与 EndRead 可以异步调用I/O线程进行读取。
public override IAsyncResult BeginRead ( byte[] array,int offset,int numBytes, AsyncCallback userCallback,Object stateObject)
public override int EndRead(IAsyncResult asyncResult)
其使用方式与BeginWrite和EndWrite相似,AsyncCallback用于绑定回调函数; Object用于传递外部数据。在回调函数只需要使用IAsyncResut.AsyncState就可获取外部数据。EndWrite 方法会返回从流读取到的字节数量。
首先定义 FileData 类,里面包含FileStream对象,byte[] 数组和长度。然后把FileData对象作为外部数据传到回调函数,在回调函数中,把IAsyncResult.AsyncState强制转换为FileData,然后通过FileStream.EndRead(IAsyncResult)结束读取。最后比较一下长度,若读取到的长度与输入的数据长度不一至,则抛出异常。
*/

class Program
{
	 public class FileData
	 {
		 public FileStream Stream;
		 public int Length;
		 public byte[] ByteData;
	 }

	 static void Main(string[] args)
	 {       
		 //把线程池的最大值设置为1000
		 ThreadPool.SetMaxThreads(1000, 1000);
		 ThreadPoolMessage("Start");
		 ReadFile();

		 Console.ReadKey();
	 }

	 static void ReadFile()
	 {
		 byte[] byteData=new byte[80961024];
		 FileStream stream = new FileStream("File1.sour", FileMode.OpenOrCreate, 
								 FileAccess.ReadWrite, FileShare.ReadWrite, 1024, true);
		 
		 //把FileStream对象,byte[]对象,长度等有关数据绑定到FileData对象中,以附带属性方式送到回调函数
		 FileData fileData = new FileData();
		 fileData.Stream = stream;
		 fileData.Length = (int)stream.Length;
		 fileData.ByteData = byteData;
		 
		 //启动异步读取
		 stream.BeginRead(byteData, 0, fileData.Length, new AsyncCallback(Completed), fileData);
	 }

	 static void Completed(IAsyncResult result)
	 {
		 ThreadPoolMessage("Completed");

		 //把AsyncResult.AsyncState转换为FileData对象,以FileStream.EndRead完成异步读取
		 FileData fileData = (FileData)result.AsyncState;
		 int length=fileData.Stream.EndRead(result);
		 fileData.Stream.Close();

		 //如果读取到的长度与输入长度不一致,则抛出异常
		 if (length != fileData.Length)
			 throw new Exception("Stream is not complete!");

		 string data=Encoding.ASCII.GetString(fileData.ByteData, 0, fileData.Length);
		 Console.WriteLine(data.Substring(2,22));
	 }

	 //显示线程池现状
	 static void ThreadPoolMessage(string data)
	 {
		 int a, b;
		 ThreadPool.GetAvailableThreads(out a, out b);
		 string message = string.Format("{0}\n  CurrentThreadId is {1}\n  "+
					  "WorkerThreads is:{2}  CompletionPortThreads is :{3}",
					  data, Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
		 Console.WriteLine(message);      
	 }
		 
}

/*
数据并行
数据并行的核心类就是System.Threading.Tasks.Parallel,它包含两个静态方法 Parallel.For 与 Parallel.ForEach, 使用方式与for、foreach相仿。通过这两个方法可以并行处理System.Func<>、System.Action<>委托。
以下一个例子就是利用 public static ParallelLoopResult For( int from, int max, Action<int>) 方法对List<Person>进行并行查询。
假设使用单线程方式查询3个Person对象,需要用时大约6秒,在使用并行方式,只需使用2秒就能完成查询,而且能够避开Thread的繁琐处理。
观察运行结果,对象并非按照原排列顺序进行查询,而是使用并行方式查询。
*/

class Program
{
	static void Main(string[] args)
	{
		//设置最大线程数
		ThreadPool.SetMaxThreads(1000, 1000);
		//并行查询
		Parallel.For(0, 3,n =>
			{
				Thread.Sleep(2000);  //模拟查询
				ThreadPoolMessage(GetPersonList()[n]);
			});
		Console.ReadKey();
	}

	//模拟源数据
	static IList<Person> GetPersonList()
	{
		var personList = new List<Person>();

		var person1 = new Person();
		person1.ID = 1;
		person1.Name = "Leslie";
		person1.Age = 30;
		personList.Add(person1);
		...........
		return personList;
	}

	//显示线程池现状
	static void ThreadPoolMessage(Person person)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
			  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
			  "  CompletionPortThreads is :{5}\n",
			  person.ID, person.Name, person.Age,
			  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

		Console.WriteLine(message);
	}
}

/*
数据并行
若想停止操作,可以利用ParallelLoopState参数,下面以ForEach作为例子。
public static ParallelLoopResult ForEach<TSource>( IEnumerable<TSource> source, Action<TSource, ParallelLoopState> action)
其中source为数据集,在Action<TSource,ParallelLoopState>委托的ParallelLoopState参数当中包含有Break()和 Stop()两个方法都可以使迭代停止。Break的使用跟传统for里面的使用方式相似,但因为处于并行处理当中,使用Break并不能保证所有运行能立即停止,在当前迭代之前的迭代会继续执行。若想立即停止操作,可以使用Stop方法,它能保证立即终止所有的操作,无论它们是处于当前迭代的之前还是之后。观察运行结果,当Person的ID等于2时,运行将会停止
*/

class Program
{
	 static void Main(string[] args)
	 {
		 //设置最大线程数
		 ThreadPool.SetMaxThreads(1000, 1000);

		 //并行查询
		 Parallel.ForEach(GetPersonList(), (person, state) =>
			 {
				 if (person.ID == 2)
					 state.Stop();
				 ThreadPoolMessage(person);
			 });
		 Console.ReadKey();
	 }

	 //模拟源数据
	 static IList<Person> GetPersonList()
	 {
		 var personList = new List<Person>();

		 var person1 = new Person();
		 person1.ID = 1;
		 person1.Name = "Leslie";
		 person1.Age = 30;
		 personList.Add(person1);
		 ..........
		 return personList;
	 }

	 //显示线程池现状
	 static void ThreadPoolMessage(Person person)
	 {
		 int a, b;
		 ThreadPool.GetAvailableThreads(out a, out b);
		 string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
			   "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
			   "  CompletionPortThreads is :{5}\n",
			   person.ID, person.Name, person.Age,
			   Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

		 Console.WriteLine(message);
	 }
 }

/*
任务并行
在TPL当中还可以使用Parallel.Invoke方法触发多个异步任务,其中 actions 中可以包含多个方法或者委托,parallelOptions用于配置Parallel类的操作。
public static void Invoke(Action[] actions )
public static void Invoke(ParallelOptions parallelOptions, Action[] actions )
下面例子中利用了Parallet.Invoke并行查询多个Person,actions当中可以绑定方法、lambda表达式或者委托,注意绑定方法时必须是返回值为void的无参数方法。
*/

class Program
{
	static void Main(string[] args)
	{
		//设置最大线程数
		ThreadPool.SetMaxThreads(1000, 1000);
		
		//任务并行
		Parallel.Invoke(option,
			PersonMessage, 
			()=>ThreadPoolMessage(GetPersonList()[1]),  
			delegate(){
				ThreadPoolMessage(GetPersonList()[2]);
			});
		Console.ReadKey();
	}

	static void PersonMessage()
	{
		ThreadPoolMessage(GetPersonList()[0]);
	}

	//显示线程池现状
	static void ThreadPoolMessage(Person person)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
			  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
			  "  CompletionPortThreads is :{5}\n",
			  person.ID, person.Name, person.Age,
			  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());

		Console.WriteLine(message);
	}

	//模拟源数据
	static IList<Person> GetPersonList()
	{
		var personList = new List<Person>();

		var person1 = new Person();
		person1.ID = 1;
		person1.Name = "Leslie";
		person1.Age = 30;
		personList.Add(person1);
		..........
		return personList;
	}
}

/*
Task简介
以Thread创建的线程被默认为前台线程,当然你可以把线程IsBackground属性设置为true,但TPL为此提供了一个更简单的类Task。
Task存在于System.Threading.Tasks命名空间当中,它可以作为异步委托的简单替代品。
通过Task的Factory属性将返回TaskFactory类,以TaskFactory.StartNew(Action)方法可以创建一个新线程,所创建的线程默认为后台线程。
*/

class Program
{
	static void Main(string[] args)
	{
		ThreadPool.SetMaxThreads(1000, 1000);
		Task.Factory.StartNew(() => ThreadPoolMessage());
		Console.ReadKey();
	}

	//显示线程池现状
	static void ThreadPoolMessage()
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("CurrentThreadId is:{0}\n" +
			"CurrentThread IsBackground:{1}\n" +
			"WorkerThreads is:{2}\nCompletionPortThreads is:{3}\n",
			 Thread.CurrentThread.ManagedThreadId,
			 Thread.CurrentThread.IsBackground.ToString(),
			 a.ToString(), b.ToString());
		Console.WriteLine(message);
	}
}

/*
AsParallel
通常想要实现并行查询,只需向数据源添加 AsParallel 查询操作即可。
*/

class Program
{
	static void Main(string[] args)
	{
		var personList=GetPersonList().AsParallel() 
			   .Where(x=>x.Age>30);
		Console.ReadKey();
	}

	//模拟源数据
	static IList<Person> GetPersonList()
	{
		var personList = new List<Person>();

		var person1 = new Person();
		person1.ID = 1;
		person1.Name = "Leslie";
		person1.Age = 30;
		personList.Add(person1);
		...........
		return personList;
	}
}

/*
AsOrdered
若要使查询结果必须保留源序列排序方式,可以使用AsOrdered方法。 
AsOrdered依然使用并行方式,只是在查询过程加入额外信息,在并行结束后把查询结果再次进行排列。
*/

class Program
{
	static void Main(string[] args)
	{
		var personList=GetPersonList().AsParallel().AsOrdered()
			.Where(x=>x.Age<30);
		Console.ReadKey();
	}

	static IList<Person> GetPersonList()
	{......}
}

/*
WithDegreeOfParallelism
默认情况下,PLINQ 使用主机上的所有处理器,这些处理器的数量最多可达 64 个。
通过使用 WithDegreeOfParallelism(Of TSource) 方法,可以指示 PLINQ 使用不多于指定数量的处理器。
*/

class Program
{
	static void Main(string[] args)
	{
		var personList=GetPersonList().AsParallel().WithDegreeOfParallelism(2)
			.Where(x=>x.Age<30);
		Console.ReadKey();
	}

	static IList<Person> GetPersonList()
	{.........}
}

/*
ForAll
如果要对并行查询结果进行操作,一般会在for或foreach中执行,执行枚举操作时会使用同步方式。
有见及此,PLINQ中包含了ForAll方法,它可以使用并行方式对数据集进行操作。
*/
class Program
{
	static void Main(string[] args)
	{
		ThreadPool.SetMaxThreads(1000, 1000);
		GetPersonList().AsParallel().ForAll(person =>{
			ThreadPoolMessage(person);
		});
		Console.ReadKey();
	}

	static IList<Person> GetPersonList()
	{.......}

	 //显示线程池现状
	static void ThreadPoolMessage(Person person)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
			  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
			  "  CompletionPortThreads is :{5}\n",
			  person.ID, person.Name, person.Age,
			  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
		Console.WriteLine(message);
	}
}

/*
WithCancellation
如果需要停止查询,可以使用 WithCancellation(Of TSource) 运算符并提供 CancellationToken 实例作为参数。 
与第三节Task的例子相似,如果标记上的 IsCancellationRequested 属性设置为 true,则 PLINQ 将会注意到它,并停止所有线程上的处理,然后引发 OperationCanceledException。这可以保证并行查询能够立即停止。
*/
class Program
{
	static CancellationTokenSource tokenSource = new CancellationTokenSource();

	static void Main(string[] args)
	{
		Task.Factory.StartNew(Cancel);
		try
		{
			GetPersonList().AsParallel().WithCancellation(tokenSource.Token)
				.ForAll(person =>
				{
					ThreadPoolMessage(person);
				});
		}
		catch (OperationCanceledException ex)
		{ }
		Console.ReadKey();
	}

	//在10~50毫秒内发出停止信号
	static void Cancel()
	{
		Random random = new Random();
		Thread.Sleep(random.Next(10,50));
		tokenSource.Cancel();
	}

	static IList<Person> GetPersonList()
	{......}

	//显示线程池现状
	static void ThreadPoolMessage(Person person)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("Person  ID:{0} Name:{1} Age:{2}\n" +
			  "  CurrentThreadId is {3}\n  WorkerThreads is:{4}" +
			  "  CompletionPortThreads is :{5}\n",
			  person.ID, person.Name, person.Age,
			  Thread.CurrentThread.ManagedThreadId, a.ToString(), b.ToString());
		Console.WriteLine(message);
	}
}
/*
定时器
若要长期定时进行一些工作,比如像邮箱更新,实时收听信息等等,可以利用定时器Timer进行操作。
在System.Threading命名空间中存在Timer类与对应的TimerCallback委托,它可以在后台线程中执行一些长期的定时操作,使主线程不受干扰。
Timer类中最常用的构造函数为 public Timer( timerCallback , object , int , int )
timerCallback委托可以绑定执行方法,执行方法必须返回void,它可以是无参数方法,也可以带一个object参数的方法。
第二个参数是为 timerCallback 委托输入的参数对象。
第三个参数是开始执行前等待的时间。
第四个参数是每次执行之间的等待时间。
注意观察运行结果,每次调用Timer绑定的方法时不一定是使用同一线程,但线程都会是来自工作者线程的后台线程。
*/

class Program
{
	static void Main(string[] args)
	{
		ThreadPool.SetMaxThreads(1000, 1000);

		TimerCallback callback = new TimerCallback(ThreadPoolMessage);
		Timer t = new Timer(callback,"Hello Jack! ", 0, 1000);
		Console.ReadKey();
	}

	//显示线程池现状
	static void ThreadPoolMessage(object data)
	{
		int a, b;
		ThreadPool.GetAvailableThreads(out a, out b);
		string message = string.Format("{0}\n   CurrentThreadId is:{1}\n" +
			"   CurrentThread IsBackground:{2}\n" +
			"   WorkerThreads is:{3}\n   CompletionPortThreads is:{4}\n",
			 data + "Time now is " + DateTime.Now.ToLongTimeString(),
			 Thread.CurrentThread.ManagedThreadId,
			 Thread.CurrentThread.IsBackground.ToString(),
			 a.ToString(), b.ToString());
		Console.WriteLine(message);
	}
}

/*
锁
在使用多线程开发时,存在一定的共用数据,为了避免多线程同时操作同一数据,.NET提供了lock、Monitor、Interlocked等多个锁定数据的方式。
lock
lock的使用比较简单,如果需要锁定某个对象时,可以直接使用lock(this)的方式。
*/

private void Method()
{
      lock(this)
      {
          //在此进行的操作能保证在同一时间内只有一个线程对此对象操作
      }
}

class Control
{
      private object obj=new object();
      
      public void Method()
      {
            lock(obj)
            {.......}
      }
}

/*
Montior
Montior存在于System.Thread命名空间内,相比lock,Montior使用更灵活。
它存在 Enter, Exit 两个方法,它可以对对象进行锁定与解锁,比lock使用更灵活。
使用try的方式,能确保程序不会因死锁而释放出异常!
而且在finally中释放obj对象能够确保无论是否出现死锁状态,系统都会释放obj对象。
而且Monitor中还存在Wait方法可以让线程等待一段时间,然后在完成时使用Pulse、PulseAll等方法通知等待线程。
*/

class Control
{
      private object obj=new object();
 
      public void Method()
      {
            Monitor.Enter(obj);
            try
            {......}
            catch(Excetion ex)
            {......}
            finally
            {
                Monitor.Exit(obj);
            }
      }
}

/*
Interlocked
Interlocked存在于System.Thread命名空间内,它的操作比Monitor使用更简单。
它存在CompareExchange、Decrement、Exchange、Increment等常用方法让参数在安全的情况进行数据交换。
Increment、Decrement 可以使参数安全地加1或减1并返回递增后的新值。
*/

class Example
{
      private int a=1;

      public void AddOne()
      {
             int newA=Interlocked.Increment(ref a);
      }
}

//Exchange可以安全地变量赋值。
public void SetData()
{
      Interlocked.Exchange(ref a,100);
}

//CompareExchange使用特别方便,它相当于if的用法,当a等于1时,则把100赋值给a。
public void CompareAndExchange()
{
    Interlocked.CompareExchange(ref a,100,1);
}

本文转载自:http://blog.csdn.net/smartsmile2012/article/details/44902173

深圳大道
粉丝 3
博文 877
码字总数 0
作品 0
深圳
架构师
私信 提问
5天玩转C#并行和多线程编程 —— 第二天 并行集合和PLinq

5天玩转C#并行和多线程编程系列文章目录 5天玩转C#并行和多线程编程 —— 第一天 认识Parallel 5天玩转C#并行和多线程编程 —— 第二天 并行集合和PLinq 5天玩转C#并行和多线程编程 —— 第三...

雲霏霏
2014/09/28
0
0
5天玩转C#并行和多线程编程 —— 第四天 Task进阶

5天玩转C#并行和多线程编程系列文章目录 5天玩转C#并行和多线程编程 —— 第一天 认识Parallel 5天玩转C#并行和多线程编程 —— 第二天 并行集合和PLinq 5天玩转C#并行和多线程编程 —— 第三...

雲霏霏
2014/11/25
0
0
C#多线程学习(五) 多线程的自动管理(定时器)

本系列文章导航 C#多线程学习(一) 多线程的相关概念 C#多线程学习(二) 如何操纵一个线程 C#多线程学习(三) 生产者和消费者 C#多线程学习(四) 多线程的自动管理(线程池) C#多线程学习(五) 多线...

Yamazaki
2012/03/29
23
0
5天玩转C#并行和多线程编程 —— 第五天 多线程编程大总结

5天玩转C#并行和多线程编程系列文章目录 5天玩转C#并行和多线程编程 —— 第一天 认识Parallel 5天玩转C#并行和多线程编程 —— 第二天 并行集合和PLinq 5天玩转C#并行和多线程编程 —— 第三...

雲霏霏
2014/11/26
0
0
c# 温故而知新: 线程篇(一)

c# 温故而知新: 线程篇(一) Thread 目录: 目录: 1 线程基础的简单介绍 2 线程同步与线程异步的简单介绍 3 前台线程与后台线程的简单介绍 4 细说下Thread 最为关键的构造函数 5 细说下Thr...

付翔
2013/01/08
0
0

没有更多内容

加载失败,请刷新页面

加载更多

面试官,Java8 JVM内存结构变了,永久代到元空间

在文章《JVM之内存结构详解》中我们描述了Java7以前的JVM内存结构,但在Java8和以后版本中JVM的内存结构慢慢发生了变化。作为面试官如果你还不知道,那么面试过程中是不是有些露怯?作为面试...

程序新视界
25分钟前
15
0
读书笔记:深入理解ES6 (八)

第八章 迭代器(Iterator)与生成器(Generator) 第1节 循环语句的问题   在循环、多重循环中,通过变量来跟踪数组索引的行为容易导致程序出错。迭代器的出现旨在消除这种复杂性,并减少循...

张森ZS
26分钟前
13
0
Elasticsearch 实战(一) - 简介

官腔 Elasticsearch,分布式,高性能,高可用,可伸缩的搜索和分析系统 基本等于没说,咱们慢慢看 1 概述 百度:我们比如说想找寻任何的信息的时候,就会上百度去搜索一下,比如说找一部自己喜...

JavaEdge
30分钟前
13
0
【jQuery基础学习】11 jQuery性能简单优化

本文转载于:专业的前端网站➦【jQuery基础学习】11 jQuery性能简单优化 关于性能优化 合适的选择器 $("#id")会直接调用底层方法,所以这是最快的。如果这样不能直接找到,也可以用find方法继...

前端老手
39分钟前
12
0
重磅发布 | 全球首个云原生应用标准定义与架构模型 OAM 正式开源

导读:2019 年 10 月 17 日,阿里巴巴合伙人、阿里云智能基础产品事业部总经理蒋江伟(花名:小邪)在 Qcon 上海重磅宣布,阿里云与微软联合推出开放应用模型 Open Application Model (OAM...

阿里云官方博客
41分钟前
15
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部