C#线程池

网友投稿 652 2022-08-31

C#线程池

C#线程池

class Program { // 使用BeginXXX/EndXXX和IAsyncResult对象的方式被称为异步编程模型（APM模式） delegate string RunOnThreadPool(out int threadId); static void Main(string[] args) { int threadId = 0; // 给委托变量赋值 RunOnThreadPool poolDelegate = Test; var t = new Thread(() => { Test(out threadId); }); t.Start(); t.Join(); Console.WriteLine($"1.返回的线程Id：{threadId}"); // 通过调用委托变量的BeginInvoke方法来运行委托（执行Test方法） IAsyncResult ar = poolDelegate.BeginInvoke(out threadId, Callback, "异步委托调用"); ar.AsyncWaitHandle.WaitOne(); // 调用委托的EndInvoke会等待异步操作（Test方法）完成 // 异步操作执行完成之后，开始执行回掉函数；异步操作和回掉函数很可能会被线程池中同一个工作线程执行 string result = poolDelegate.EndInvoke(out threadId, ar); Console.WriteLine($"2.返回的线程Id：{threadId}"); Console.WriteLine($"返回值：{result}"); Console.ReadKey(); } static void Callback(IAsyncResult ar) { Console.WriteLine("开始执行回掉函数..."); Console.WriteLine($"异步状态：{ar.AsyncState}"); Console.WriteLine($"是否为线程池中的线程：{Thread.CurrentThread.IsThreadPoolThread}"); Console.WriteLine($"线程池工作线程Id：{Thread.CurrentThread.ManagedThreadId}"); } static string Test(out int threadId) { Console.WriteLine("开始测试方法..."); Console.WriteLine($"是否为线程池中的线程：{Thread.CurrentThread.IsThreadPoolThread}"); Thread.Sleep(TimeSpan.FromSeconds(2)); threadId = Thread.CurrentThread.ManagedThreadId; return $"线程Id：{threadId}"; } }

向线程池中放入异步操作

class Program { static void Main(string[] args) { ThreadPool.QueueUserWorkItem(AsyncOperation); Thread.Sleep(TimeSpan.FromSeconds(2)); ThreadPool.QueueUserWorkItem(AsyncOperation, "async state"); Thread.Sleep(TimeSpan.FromSeconds(2)); ThreadPool.QueueUserWorkItem(state => { Console.WriteLine($"操作状态：{state}"); Console.WriteLine($"工作线程Id：{Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(TimeSpan.FromSeconds(2)); }, "lambda state"); const int x = 1; const int y = 2; const string lambdaState = "lambda state 2"; // 使用闭包机制，无需传递lambda表达式的状态参数 ThreadPool.QueueUserWorkItem(_ => { Console.WriteLine($"操作状态：{x + y},{lambdaState}"); Console.WriteLine($"工作线程Id：{Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(TimeSpan.FromSeconds(2)); }); Console.ReadKey(); } static void AsyncOperation(object state) { Console.WriteLine($"操作状态：{state ?? "(null)"}"); Console.WriteLine($"工作线程Id：{Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(TimeSpan.FromSeconds(2)); } }

线程池与并行度：

static void Main(string[] args) { int threadCount = 500; var sw = new Stopwatch(); sw.Start(); UseThreads(threadCount); sw.Stop(); Console.WriteLine($"手动创建线程执行操作耗时：{sw.ElapsedMilliseconds}"); sw.Reset(); sw.Start(); UseThreadPool(threadCount); sw.Stop(); Console.WriteLine($"使用线程池执行操作耗时：{sw.ElapsedMilliseconds}"); Console.ReadKey(); } static void UseThreads(int threadCount) { using (CountdownEvent cdEvt = new CountdownEvent(threadCount)) { Console.WriteLine("开始创建线程"); for (int i = 0; i < threadCount; i++) { var thread = new Thread(() => { Console.WriteLine($"是否为线程池中的线程：{Thread.CurrentThread.IsThreadPoolThread}，线程Id：{Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(TimeSpan.FromSeconds(0.1)); cdEvt.Signal(); }); thread.Start(); } cdEvt.Wait(); Console.WriteLine(); } } ///

/// 线程池的用途是执行时间短的操作，使用线程池可以减少并行度消耗及节省操作系统资源 /// /// static void UseThreadPool(int threadCount) { using (CountdownEvent cdEvt = new CountdownEvent(threadCount)) { Console.WriteLine("开始使用线程池"); for (int i = 0; i < threadCount; i++) { ThreadPool.QueueUserWorkItem(_ => { Console.WriteLine($"是否为线程池中的线程：{Thread.CurrentThread.IsThreadPoolThread}，线程Id：{Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(TimeSpan.FromSeconds(0.1)); cdEvt.Signal(); }); } cdEvt.Wait(); Console.WriteLine(); } } //在收到信号特定次数后取消阻止等待线程的同步基元 //可以使用取消令牌取消等待操作 //创建实例后，可以递增信号计数 //在 Reset 方法调用返回 Wait 后，可以重用实例 public void Test() { IEnumerable source = GetData(); using (CountdownEvent e = new CountdownEvent(1)) { foreach (Data element in source) { e.AddCount(); ThreadPool.QueueUserWorkItem(delegate (object state) { try { ProcessData(state); } finally { e.Signal(); } }, element); } } } static IEnumerable GetData() { return new List() { new Data(1), new Data(2), new Data(3), new Data(4), new Data(5) }; } static void ProcessData(object obj) { DataWithToken dataWithToken = (DataWithToken)obj; if (dataWithToken.Token.IsCancellationRequested) { Console.WriteLine("Canceled before starting {0}", dataWithToken.Data.Num); return; } for (int i = 0; i < 10000; i++) { if (dataWithToken.Token.IsCancellationRequested) { Console.WriteLine("Cancelling while executing {0}", dataWithToken.Data.Num); return; } Thread.SpinWait(100000); } Console.WriteLine("Processed {0}", dataWithToken.Data.Num); } static void EventWithCancel() { IEnumerable source = GetData(); CancellationTokenSource cts = new CancellationTokenSource(); Task.Factory.StartNew(() => { if (Console.ReadKey().KeyChar == 'c') cts.Cancel(); }); CountdownEvent e = new CountdownEvent(1); foreach (Data element in source) { DataWithToken item = new DataWithToken(element, cts.Token); e.AddCount(); ThreadPool.QueueUserWorkItem(delegate (object state) { ProcessData(state); if (!cts.Token.IsCancellationRequested) e.Signal(); }, item); } e.Signal(); try { e.Wait(cts.Token); } catch (OperationCanceledException oce) { if (oce.CancellationToken == cts.Token) { Console.WriteLine("User canceled."); } else { throw; //We don't know who canceled us! } } finally { e.Dispose(); cts.Dispose(); } } } class Data { public int Num { get; set; } public Data(int i) { Num = i; } public Data() { } } class DataWithToken { public CancellationToken Token { get; set; } public Data Data { get; private set; } public DataWithToken(Data data, CancellationToken ct) { this.Data = data; this.Token = ct; } }

取消异步操作：

static void Main(string[] args) { using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; ThreadPool.QueueUserWorkItem(_ => { AsyncOperation1(token); }); Thread.Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; ThreadPool.QueueUserWorkItem(_ => { AsyncOperation2(token); }); Thread.Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } using (var cts = new CancellationTokenSource()) { CancellationToken token = cts.Token; ThreadPool.QueueUserWorkItem(_ => { AsyncOperation3(token); }); Thread.Sleep(TimeSpan.FromSeconds(2)); cts.Cancel(); } Console.ReadKey(); } static void AsyncOperation1(CancellationToken token) { Console.WriteLine("启动第一个异步操作"); for (int i = 0; i < 5; i++) { // 轮询检查IsCancellationRequested属性 if (token.IsCancellationRequested) { Console.WriteLine("第一个异步操作被取消啦~~~"); return; } Thread.Sleep(TimeSpan.FromSeconds(1)); } Console.WriteLine("第一个异步操作完成"); } static void AsyncOperation2(CancellationToken token) { try { Console.WriteLine("启动第二个异步操作"); for (int i = 0; i < 5; i++) { // 抛出异常，取消操作时，通过操作之外的代码来处理 token.ThrowIfCancellationRequested(); Thread.Sleep(TimeSpan.FromSeconds(1)); } Console.WriteLine("第二个异步操作完成"); } catch (OperationCanceledException ex) { Console.WriteLine($"第二个异步操作被取消，异常消息：{ex.Message}"); } } static void AsyncOperation3(CancellationToken token) { bool cancellationFlag = false; // 注册回调函数，当操作被取消时，线程池将调用该回调函数 token.Register(() => { cancellationFlag = true; }); Console.WriteLine("启动第三个异步操作"); for (int i = 0; i < 5; i++) { if (cancellationFlag) { Console.WriteLine("第三个异步操作被取消啦~~~"); return; } Thread.Sleep(TimeSpan.FromSeconds(1)); } Console.WriteLine("第三个异步操作完成"); }

在线程池中使用等待事件处理器及超时

static void Main(string[] args) { RunOperation(TimeSpan.FromSeconds(5)); RunOperation(TimeSpan.FromSeconds(7)); Console.ReadKey(); } static void RunOperation(TimeSpan workerOperationTs) { using (var evt = new ManualResetEvent(false)) using (var cts = new CancellationTokenSource()) { Console.WriteLine("注册操作超时的方法..."); var worker = ThreadPool.RegisterWaitForSingleObject(evt, (state, isTimeout) => WorkerOperationWait(cts, isTimeout), null, workerOperationTs, true); Console.WriteLine("开始长时间操作..."); ThreadPool.QueueUserWorkItem(_ => { WorkerOperation(cts.Token, evt); }); Thread.Sleep(workerOperationTs.Add(TimeSpan.FromSeconds(2))); worker.Unregister(evt); } } static void WorkerOperation(CancellationToken token, ManualResetEvent evt) { for (int i = 0; i < 6; i++) { if(token.IsCancellationRequested) { Console.WriteLine("异步操作被取消"); return; } Thread.Sleep(TimeSpan.FromSeconds(1)); } evt.Set(); } static void WorkerOperationWait(CancellationTokenSource cts,bool isTimeout) { if(isTimeout) { cts.Cancel(); Console.WriteLine("异步操作超时并被取消了"); } else { Console.WriteLine("异步操作完成"); } }

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。

标签：代码
上一篇:从 0 到 1 学习 Go 语言
下一篇:Net作业调度-----Quartz.Net
相关文章

 发表评论

暂时没有评论，来抢沙发吧~