聊聊多线程那一些事儿（task）之三异步取消和异步方法

程序员文章站 2023-11-10 21:39:40

了解了多线程的基本使用后，现在我们来聊聊多线的实际项目场景：多线程的取与舍，也就是取在指定时间有响应结果的任务，舍弃未响应的任务。我们也遇到根据实际需要处理的业务量动态创建线程并控制等场景。 ......

hello task,咱们又见面啦！！前面已经通过三篇简单的文章，对多线程的创建、运行、阻塞、等待、取消、延迟操作、异步方法等相关的知识点，通过这一些介绍，现在上手写一个多线程就是分分钟的小事件。如果需要看前三排文章的小伙伴，可以点击下面链接快速阅读谢谢！

说了那么多后，我仔细想了一下，还是要来点实际的项目用例比较实在，那么我现在就讲我平时在项目中用常用的一些业务梳理处理，以供参考，写到不好勿喷，有更好的解决方式，欢迎交流。谢谢！

应用一、多线程的中的取与舍

还是用上几篇文章中的关于酒店客房的数据来为例分析，假设系统同时对接了x程、y龙、q哪三家接口数据，用户进入到某一个酒店预订页面，系统需要实时的到第三方取该酒店对应房间的实时动态数据呈现给用户，但是在这个过程中，不能让用户等待的太久，并且能够尽可能多的提供多渠道给用户选择，那么这个时候该如何去实现这个需求呢？

做过聚合平台的同学，无论是酒店、机票、咨询等等，或多或少都会遇到这样类似的业务场景，下面分享一下，我们平时是怎么实现。

简单的说就是一个取与舍的逻辑，你想啊，不同的接口方，接口的查询效率不尽相同，并且同一接口在不同时间处理时间也不尽相同，遇到某一些特殊情况，有可能一个接口数据需要等待5-10s甚至更久才能取出数据，这个时候你不可能让用户也等待这么久吧，如果是这样，估计用户早吓跑了。况且这样等待下去，也会对系统带来很大的压力，尤其是在做活动高峰期时，那直接就是系统的一个瓶颈了。

为了到达这一目的，那么在实现上，我们首先应该想到的是，在约定时间内未返回数据的接口，那么我们直接放弃，只展示给用户已取到的数据即可。那在技术上该如何实现呢？

在实现上，有了task，一切都变得那么轻松啦，因为task.wait方法已经为我们考虑到了这样的场景，直接使用即可。

我们看看task.wait的几个重载吧：

聊聊多线程那一些事儿（task）之三异步取消和异步方法

看到了吧，在重载方法中，有一个timeout字段，该字段就是用于舍弃超时未处理完成的线程任务。那具体该怎么用呢？直接上代码吧！

static void main(string[] args)
{
    console.writeline("开始获取酒店数据......");
// 获取最新的客房信息(假设最大等待时间为1s)
    list<string> listhotelroominfro = gethotelroominfro(1000);
    console.writeline("获取酒店数据获取完毕,获取到的数据为：");
foreach (var item in listhotelroominfro)
    {
        console.writeline(item);
    }
    console.readline();
 }
 
/// <summary>
/// 获取最新的客房信息
/// </summary>
/// <param name="maxwaittimes">最大获取时间（也就是舍弃时间），单位毫秒</param>
/// <returns>客房信息集合</returns>
private static list<string> gethotelroominfro(int maxwaittimes)
 {
// 模拟存储获取到的酒店客房数据集合
     list<string> listhotelroominfro = new list<string>();
// 在此我也分别对3种不同渠道，采用3种不同的方式来实现
// 其一、通过传统的 new 方式来实例化一个task对象，获取 携程 的客房数据
     task newctriptask = new task(() =>
    {
// 具体获取业务逻辑处理...
         thread.sleep(new random().next(100, 999));
        listhotelroominfro.add("我是来自 携程 的最新客房信息");
    });
// 启动 tsak
     newctriptask.start();
// 其二、通过工厂 factory 来生成一个task对象，并自启动：获取 艺龙 的客房数据
     task factoryelongtask = task.factory.startnew(() =>
     {
// 具体获取业务逻辑处理...
         thread.sleep(new random().next(555, 1500));
         listhotelroominfro.add("我是来自 艺龙 的最新客房信息");
     });
// 其三、通过 task.run(action action) 来创建一个自启动task：获取 去哪儿网 的客房数据
     task runqunartask = task.run(() =>
     {
// 具体获取业务逻辑处理...
         thread.sleep(new random().next(1100, 2000));
         listhotelroominfro.add("我是来自 去哪儿网 的最新客房信息");
     });
// 等待获取接口，阻塞主流程，如果 在 指定的时间 maxwaittimes未返回数据的接口方直接舍弃掉
     task.waitall(new task[] { newctriptask, factoryelongtask, runqunartask }, maxwaittimes);
return listhotelroominfro;
 }

运行结果：

聊聊多线程那一些事儿（task）之三异步取消和异步方法

通过上面的运行结果，我们发现在1s内，只有携程返回了数据，那么最终也就只返回给用户携程的数据。这个实例就这样实现了，当然，wi相信你或许会有更好的实现，欢迎一起交流与学习，谢谢

应用二：动态创建多线程均批处理

在实际项目中，我们会遇到需要根据待处理任务数量，动态创建多线程来最优化分批处理。哈哈哈是不是说的有点空洞，云里雾里的，不急不急，下面来一个实际场景，一看你就明明白白啦。

需求：还是以酒店数据同步为例。具体需求是这样的：

1.手动批选择指定酒店，单次最多可选择500个

2.系统需要以最快的方式同步会接口方的最新数据

哈哈哈看了需求是不是觉得很简单，就两句话，那该如何来实现这个需求呢？

也许你会说，简单啊，根据所选择的酒店以次排队同步数据就对了嘛！这样可不行哦，如果用户选择了500个酒店，假设每个酒店数据同步需要2秒，那么500条数据都需要1000秒，那需要接近20分钟才处理完，这可不是时间最优啊！

那你也要你还会说那就直接每一个酒店都开一个线程来处理，这样速度是不是够快啦！嗯你说的没错，这样速度是上去了，但是呢，如果用户选择500个酒店，那么就需要创建500个线程，也就是500个并发，这样会有什么问题呢？第一.你自己的服务器扛的住吗？第二.如果你服务器扛的住，那么接口允许你这么干嘛？估计早都把你拉黑名单了。所以这也不是最优方案。

哈哈，说了那么多，那么具体该如何来实现呢？其实也简单，我想说的方案就去以上两两种方案的一个折中方案。动态的根据资源等多因素动态创建合理的线程，然后在并行处理。来来，直接上代码！

class program
 {
     /// <summary>
     /// 最多允许创建的线程数，可以通过配置文件来配置
     /// 具体的值也该是由：服务器资源+接口限制等多因数来确定的
     /// </summary>
     public static int maxthread = 10;
     static void main(string[] args)
     {
         // 假设选择处理20条酒店数据
         list<string> listhotel = new list<string>();
         for (int i = 0; i < 20; i++)
         {
             listhotel.add($"我是酒店{(i + 1).tostring().padleft(2, '0')}");
         }

         // 创建tsak处理酒店数据同步
         asyncdynamicsynchronouslyhotel(listhotel);

         console.readline();
     }

     /// <summary>
     /// 根据酒店数据量，动态创建tsak来处理酒店数据同步
     /// </summary>
     /// <param name="listhotel">待处理的酒店数据列表</param>
     private static void asyncdynamicsynchronouslyhotel(list<string> listhotel)
     {
         object hoteltasklock = new object();

         // 先做一个非空判断
         if (listhotel == null || listhotel.count < 1)
         {
             return;
         }

         // task线程数组
         list<task> tasklist = new list<task>();

         // 首先根据资源数据量+最大允许线程数来确定需要开启的线程
         int taskcount = listhotel.count < maxthread ? listhotel.count : maxthread;

         // 创建指定是task线程数
         for (int i = 0; i < taskcount; i++)
         {
             // 创建一个task线程
             tasklist.add(new task(() =>
             {
                 while (listhotel != null && listhotel.count > 0)
                 {
                     // 给该线程分配一个酒店处理任务
                     string hotelinfro = string.empty;

                     // 线程通过，加一个资源锁
                     lock (hoteltasklock)
                     {
                         // 在获取到锁后，还需要做一个资源判断，避免获取到锁后，资源以及被消耗完毕
                         if (listhotel != null && listhotel.count > 0)
                         {
                             hotelinfro = listhotel[0];
                             listhotel.remove(hotelinfro);
                         }
                     }

                     // 开始模拟真正的数据同步操作
                     if (!string.isnullorempty(hotelinfro))
                     {
                         thread.sleep(1000);
                         console.writeline($"我是线程id{thread.currentthread.managedthreadid.tostring()  }，完成酒店【{hotelinfro}的数据同步处理");
                     }
                 }
             }));

             // 启动线程
             tasklist[i].start();
         }
     }
 }

运行结果：

设置最多开5个线程

聊聊多线程那一些事儿（task）之三异步取消和异步方法