c#多线程开发

  1

        //要执行的业务是从阿里云上下载将近40000条的音频到本地，单条下载忒慢，就想采用多线程，分配了二十个线程同时下载，省了很大部分的时间

          class program
  2     {
  3        
  4         static void main(string[] args) {
  5             string sql = "select en_audio,us_audio from t_audio limit 198 ";
  6             mysqldatareader mysqldatareader = dbhelper.executereader(sql);         
  7             list<string> slist = new list<string>();
  8             slist.add("https://qutifen-qudao.oss-cn-beijing.aliyuncs.com/mfg/audio/v3/1abacus_en.ogg");
  9             slist.add("https://qutifen-qudao.oss-cn-beijing.aliyuncs.com/mfg/audio/v3/2abacus_en.ogg");
 10             if (mysqldatareader.hasrows)
 11             {
 12                 while (mysqldatareader.read())
 13                 {                 
 14                     slist.add(mysqldatareader.getstring(0));
 15                     slist.add(mysqldatareader.getstring(1));
 16                 }
 17             }              
 18             console.writeline(slist.count);
 19             stopwatch stopwatch = new stopwatch();
 20             stopwatch.start();
 21             threadstart(slist);
 22             waithandle.waitall(waits);  //监听wait里面的所有的线程都已经set了 才执行下面的代码，否则一直在这里等待
 23             stopwatch.stop();
 24             console.writeline($"耗时{stopwatch.elapsedmilliseconds}毫秒");
 25             console.readkey();
 26 
 27 
 28         }
 29        static  thread[] threads = new thread[20];
 30         static waithandle[] waits = new waithandle[20];
 31         public  static void threadstart(list<string> nums) {
                  //分配线程
 32             for (int i=0;i<20;i++) {
 33                 threads[i] = new thread(downloadfile);
 34                 waits[i] = new autoresetevent(false);
 35 
 36             }
                 //为每个线程分配要执行的数据并开始执行
 37             for (int i = 0; i < 20; i++)
 38             {
 39                 if (i== threads.length-1) {
 40                     var retult = nums.skip(nums.count / 20 * i).take(nums.count- nums.count / 20*i).tolist();
 41                     threads[i].start(new objpt()
 42                     {
 43                         slist = retult,
 44                         waithandle = waits[i],
 45                         threadindex = i
 46                     });
 47                 }
 48                 else {
 49                   var retult= nums.skip(nums.count / 20 * i).take(nums.count / 20).tolist();
 50                     threads[i].start(new objpt() { 
 51                      slist= retult,
 52                      waithandle=waits[i],
 53                      threadindex=i
 54                     });
 55                 }
 56                
 57             }
 58         }
 59 
 60         public static void downloadfile(object obj)
 61         {
 62             int count = 0;
 63             objpt optobj = (obj as objpt);
 64             var slist = optobj.slist;
 65             console.writeline($"线程{optobj.threadindex}开始了");
 66             foreach (var url in slist)
 67             {
 68                 try
 69                 {
 70                     count++;
 71                     var arrs = url.split('/');
 72                     webrequest request = webrequest.create(url);
 73                     httpwebresponse res = (httpwebresponse)request.getresponse();
 74                     webresponse response = request.getresponse();           
 75                     if (res.statuscode.tostring() == "ok")
 76                     {
 77                         stream responsestream = response.getresponsestream();
 78                         using (filestream fswrite = new filestream($"f:/audio/v4/{arrs[arrs.length - 2]}/{arrs[arrs.length - 1]}", filemode.openorcreate, fileaccess.write))
 79                         {
 80                             byte[] buffer = new byte[response.contentlength];
 81                             while (true)
 82                             {
 83                                 ////返回本次实际读取到的字节数
 84                                 int r = responsestream.read(buffer, 0, buffer.length);
 85                                 if (r == 0)
 86                                 {
 87                                     break;
 88                                 }
 89                                 fswrite.write(buffer, 0, r);///写入
 90 
 91                             }
 92                         }
 93                     }
 94                     //if (count % 20 == 0 || count == slist.count)
 95                     //{
 96                     //    console.writeline($"线程{optobj.threadindex}已处理个数：{count}");
 97                     //}
 98 
 99                 }
100                 catch (exception ex)
101                 {
102                     string strerrorlogfile = system.appdomain.currentdomain.basedirectory + $"\\{optobj.threadindex}errorlog.log";
103                     if (!system.io.file.exists(strerrorlogfile))
104                         system.io.file.writealltext(strerrorlogfile, "//系统错误日志记录文件\r\n");
105                     object objsql ="线程"+ optobj.threadindex.tostring()+ ex.message;
106                     system.io.file.appendalltext(strerrorlogfile, datetime.now.tostring("yyyy-mm-dd hh:mm:ss") + "\t" + objsql.tostring() + url + "\r\n");
107 
108                 }
109 
110             }
111             console.writeline($"线程{optobj.threadindex}结束");
112             (optobj.waithandle as autoresetevent).set();  //set方法是当某个线程结束起做个标记的作用
113         }
114     }
115 
116      public class objpt {
117         public list<string> slist { get; set; }
118         public waithandle waithandle { get; set; }
119         public int threadindex { get; set; }
120 
121     }

 

using system;
using system.collections.generic;
using system.linq;
using system.text;
using system.threading;

namespace consoletest
{
    //下面代码说明在进程等待后台线程执行完时，如何使用等待句柄阻止进程终止。
    class program
    {
        static autoresetevent autoevent = new autoresetevent(false);

        static void main(string[] args)
        {
            console.writeline("main starting.");

            threadpool.queueuserworkitem(new waitcallback(workmethod), autoevent);

            //if (autoevent.waitone(1000, false))
            //    console.writeline("work method signaled.");
            //else
            //    console.writeline("time out waiting for work "+"method to signal.");

            //wait for work method to signal.
            autoevent.waitone();

            console.writeline("main ending.");
        }

        static void workmethod(object stateinfo)
        {
            console.writeline("work starting.");
            thread.sleep(new random().next(100, 2000));
            console.writeline("work ending.");
            ((autoresetevent)stateinfo).set();
        }
    }
}

执行结果：

main starting.
work starting.
work ending.
main ending.

autoresetevent和manualresetevent实例化，如果为 true，则将初始状态设置为终止（不阻塞）；如果为 false，则将初始状态设置为非终止（阻塞）。

//将事件状态设置为终止状态，从而允许继续执行一个或多个等待线程。
//public bool set();
//将事件状态设置为非终止，从而导致线程受阻。
//public bool reset();

waitone：在一个线程mainthread中开启一个新的线程newthread，在完成初始化并启动newthread的操作后，调用waitone，则mainthread堵塞，直到在newthread中调用set，mainthread才继续执行。

 

autoresetevent 允许线程通过发信号互相通信。通常，此通信涉及线程需要独占访问的资源。

线程通过调用 autoresetevent 上的 waitone 来等待信号。如果 autoresetevent 处于非终止状态，则该线程阻塞，并等待当前控制资源的线程通过调用 set 发出资源可用的信号。

调用 set 向 autoresetevent 发信号以释放等待线程。autoresetevent 将保持终止状态，直到一个正在等待的线程被释放，然后自动返回非终止状态。如果没有任何线程在等待，则状态将无限期地保持为终止状态。

可以通过将一个布尔值传递给构造函数来控制 autoresetevent 的初始状态，如果初始状态为终止状态，则为 true；否则为 false。

通俗的来讲只有等myreseteven.set()成功运行后,myreseteven.waitone()才能够获得运行机会;set是发信号，waitone是等待信号，只有发了信号，等待的才会执行。如果不发的话，waitone后面的程序就永远不会执行。