C#实现控制线程池最大数并发线程

1. 实验目的:

使用线程池的时候,有时候需要考虑服务器的最大线程数目和程序最快执行所有业务逻辑的取舍。
并非逻辑线程越多也好,而且新的逻辑线程必须会在线程池的等待队列中等待 ,直到线程池中工作的线程执行完毕,
才会有系统线程取出等待队列中的逻辑线程,进行CPU运算。

2.  解决问题:

<a>如果不考虑服务器实际可支持的最大并行线程个数,程序不停往线程池申请新的逻辑线程,这个时候我们可以发现CPU的使用率会不断飙升,并且内存、网络带宽占用也会随着逻辑线程在CPU队列中堆积,而不断增大。

<b>如果我们想在主程序有200个http网络通讯需要执行,如何每次循环用10个线程并发处理10个网络http通讯回话,下一次循环只有在上一次循环的10个线程都执行完毕后才会执行下一次循环,并且主程序监听和等待200个http网络通讯都在CPU线程池中执行完毕后,才会退出主程序。

3.  实现逻辑:

我们通过两个AutoResetEvent和线程监听器Monitor,分别实现:

<a>wait_sync:   任务线程的 并发执行,每次循环只处理最大10个线程分别对网络做http通讯回话。并且当前循环的10个线程都执行完毕后,才会进行下一次循环处理。
       <b> wait_main: 主程序线程的监听和等待,只有所有任务线程都执行完毕后,主程序线程才会退出程序。
       <c> list_Thread: 负责记录每次循环,CPU实际分配的系统线程的个数。和Monitor配合使用,Monitor.Enter(list_Thread)=占用共享线程资源的占用锁,Monitor.Exit(list_Thread)释放共享线程资源的占用锁。  
       <d> n_total_thread: 配合wait_main使用,记录全部逻辑线程,已经执行完毕的当前总个数,用来判断主线程是否还需要继续等待,还是可以结束主程序运行。

4. 主要代码:

<a> 线程池控制代码,如下:

/// <summary>
/// 多线程调用WCF
/// </summary>
/// <param name="select">调用WCF的方式,1=Restful,2=Tcp</param>
/// <param name="num"></param>
static void DoTest_MultiThread(string select, long num)
{
  int n_max_thread = 10; // 设置并行最大为10个线程
  int n_total_thread = 0; // 用来控制:主程序的结束执行,当所有任务线程执行完毕

  ILog log_add = new LogHelper("Add_Thread");
  ILog log_del = new LogHelper("Del_Thread");
  ILog log_wait = new LogHelper("Wait_Thread");
  ILog log_set = new LogHelper("Set_Thread");
  ILog log_for = new LogHelper("For_Thread");

  Console.Title = string.Format("调用WCF的方式 => {0}, 调用次数=> {1}"
    , select == "1" ? "Restful" : "Socket"
    , num);

  List<int> list_Thread = new List<int>();

  System.Threading.AutoResetEvent wait_sync = new System.Threading.AutoResetEvent(false); // 用来控制:并发最大个数线程=n_max_thread
  System.Threading.AutoResetEvent wait_main = new System.Threading.AutoResetEvent(false); // 用来控制:主程序的结束执行,当所有任务线程执行完毕

  DateTime date_step = DateTime.Now;
  for (long i = 0; i < num; i++)
  {
    Num_Query_Static++;
    if (i >0 && (i+1-1) % n_max_thread == 0) // -1 表示第max个线程尚未开始
    {
      //log_wait.Info(string.Format("thread n= {0},for i= {1}", dic_Thread.Count, i + 1));
      wait_sync.WaitOne(); // 每次并发10个线程,等待处理完毕后,在发送下一次并发线程
    }
    log_for.Info(string.Format("thread n= {0},for i= {1}", list_Thread.Count, i + 1));

    System.Threading.ThreadPool.QueueUserWorkItem
      ((data) =>
      {
        int id = System.Threading.Thread.CurrentThread.ManagedThreadId;
        System.Threading.Monitor.Enter(list_Thread);
        list_Thread.Add(id);
        System.Threading.Monitor.Exit(list_Thread);

        log_add.Info(string.Format("id={0}, count={1}", id, list_Thread.Count)); // 日志

        if (select == "1") // Restful方式调用
        {
          Query_Htty();
        }
        else
        {
          Query_Socket();
        }

        n_total_thread += 1;
        if (list_Thread.Count == (n_max_thread) || n_total_thread == num)
        {
          list_Thread.Clear();
          //log_set.Info(string.Format("thread n= {0},for i= {1}", dic_Thread.Count, i + 1));
          //wait_sync.Set();
          if (n_total_thread != num)
          {
            wait_sync.Set(); // 任务线程,继续执行
          }
          else
          {
            wait_main.Set(); // 主程序线程,继续执行
          }
        }
      }, list_Thread);
  }

  wait_main.WaitOne();

  Console.WriteLine(string.Format("总测试{0}次,总耗时{1}, 平均耗时{2}"
    , num
    , (DateTime.Now - date_step).ToString()
    , (DateTime.Now - date_step).TotalMilliseconds / num));

  Query_Thread();
}

 <b> WCF后台服务代码

private static ILog log = new LogHelper("SeqService"); // 日志
private static Dictionary<int, DateTime> dic_thread = new Dictionary<int, DateTime>(); // 线程列表

private static long Num = 0; // 线程个数
private static object lock_Num = 0; // 共享数据-锁

/// <summary>
/// 在线申请流水号
/// </summary>
/// <returns></returns>
[WebGet(UriTemplate = "GetSeqNum/Json", ResponseFormat = WebMessageFormat.Json)]
public string GetSeqNumber()
{
  lock (lock_Num)
  {
    Num++;
    int id_thread = System.Threading.Thread.CurrentThread.ManagedThreadId;
    DateTime now = DateTime.Now;
    if (!dic_thread.TryGetValue(id_thread, out now))
    {
      dic_thread.Add(id_thread, DateTime.Now);
    }

  }
  string ret = DateTime.Now.ToString("yyyyMMdd") + Num.ToString(new string('0', 9)); 

  log.Info(string.Format("{0}, Thread={1}/{2}", ret, System.Threading.Thread.CurrentThread.ManagedThreadId, dic_thread.Count));
  return ret;
}
  

5.  实验结果

1. 10000个WCF网络http请求,CPU分成每次10个(10可以按需求调整)线程并发执行,并且主程序在所有请求都执行完毕后,才退出主程序。

1. 前端日志:LogFile\Add_Thread\Info

2. WCF日志:LogFile\SeqService\Info

时间: 2016-07-30

C#控制台下测试多线程的方法

本文实例讲述了C#控制台下多线程实现方法.分享给大家供大家参考.具体如下: class Program { static void Main(string[] args) { ThreadStart num = new ThreadStart(PrintNum); Thread ConstrolNum = new Thread(num); ThreadStart str = new ThreadStart(PrintStr); Thread ConstrolStr = new Thread(st

C#多线程之线程控制详解

本文为大家分享了C#多线程之线程控制,供大家参考,具体内容如下 方案一: 调用线程控制方法.启动:Thread.Start();停止:Thread.Abort();暂停:Thread.Suspend();继续:Thread.Resume(); private void btn_Start_Click(object sender, EventArgs e) { mThread.Start(); // 开始 } private void btn_Stop_Click(object sender, E

C#实现的Win32控制台线程计时器功能示例

本文实例讲述了C#实现的Win32控制台线程计时器功能.分享给大家供大家参考,具体如下: 在C#中提供了三种类型的计时器: 1.基于 Windows 的标准计时器(System.Windows.Forms.Timer) 2.基于服务器的计时器(System.Timers.Timer) 3.线程计时器(System.Threading.Timer) 一.基于 Windows 的标准计时器(System.Windows.Forms.Timer) 首先注意一点就是:Windows 计时器是为单线程环境

C#通过Semaphore类控制线程队列的方法

本文实例讲述了C#通过Semaphore类控制线程队列的方法.分享给大家供大家参考.具体实现方法如下: using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.IO; using System.Diagnostics; using System.Threading; using System.ComponentModel; using System.Col

C#线程执行超时处理与并发线程数控制实例

本文实例讲述了C#线程执行超时处理与并发线程数控制的方法.分享给大家供大家参考.具体实现方法如下: 特别说明: 1.为了测试方便,这里对存储过程的执行是模拟的 2.这里限制了并发执行存储过程的最大个数,但并没有对并发线程数进行控制,与文章标题略有不符,但程序稍做改动即可控制并发线程数 代码如下: 复制代码 代码如下: using System; using System.Collections.Generic; using System.ComponentModel; using System.

Shell多线程操作及线程数控制实例

前言 在业务开发过程中,经常会在后台写一些shell脚本处理数据,但估计很多人不知道shell脚本也可以支持多线程,而且非常简单.本篇文章主要就是介绍shell实现多进程以及进程数量控制. 需求 为了更好的说明问题,我们结合例子讲解,假设需求就是扫描url.txt文件,然后判断里面的URL是否失效.url.txt文件的内容是一行一个URL,如: 复制代码 代码如下: http://www.baidu.com http://www.google.com http://www.jb51.net 单进

Java并发编程示例(六):等待线程执行终止

在某些场景下,我们必须等待线程执行完成才能进行下一步工作.例如,某些程序在开始执行之前,需要先初始化一些资源.这时,我们可以启动一个线程专门来做初始化任务,等到线程任务完成后,再去执行其他部分. 为此,Thread类为我们提供了join()方法.当我们使用线程对象调用此方法时,正在掉调用的线程对象将被推迟到被调用对象执行完成后再开始执行. 在本节,示例程序演示等待初始化方法完成后,再去执行其他任务. 知其然 按照下面所示步骤,完成示例程序. 1.创建一个名为DataSourcesLoader的类

.net让线程支持超时的方法实例和线程在执行结束后销毁的方法

.net让线程支持超时 使用 CancellationTokenSource 复制代码 代码如下: private static void TimeoutTest1()        {            var cts = new CancellationTokenSource(); var thread = new Thread(() =>            {                Console.WriteLine(String.Format("线程{0}执行中&q

Java并发 线程间的等待与通知

前言: 前面讲完了一些并发编程的原理,现在我们要来学习的是线程之间的协作.通俗来说就是,当前线程在某个条件下需要等待,不需要使用太多系统资源.在某个条件下我们需要去唤醒它,分配给它一定的系统资源,让它继续工作.这样能更好的节约资源. 一.Object的wait()与notify() 基本概念: 一个线程因执行目标动作的条件未能满足而被要求暂停就是wait,而一个线程满足执行目标动作的条件之后唤醒被暂停的线程就是notify. 基本模板: synchronized (obj){ //保护条件不成立

IOS 创建并发线程的实例详解

IOS 创建并发线程的实例详解 创建并发线程 主线程一般都是处理UI界面及用户交互的事儿的.其他的事一般就要另外的线程去处理,如下载,计算等... 现在先简单创建3个线程,分别打印出1-1000,,为了方便,线程3就放在主线程中执行. - (void) firstCounter{ @autoreleasepool { NSUInteger counter = 0; for (counter = 0; counter < 1000; counter++){ NSLog(@"First Cou

iOS应用程序中通过dispatch队列控制线程执行的方法

GCD编程的核心就是dispatch队列,dispatch block的执行最终都会放进某个队列中去进行,它类似NSOperationQueue但更复杂也更强大,并且可以嵌套使用.所以说,结合block实现的GCD,把函数闭包(Closure)的特性发挥得淋漓尽致. dispatch队列的生成可以有这几种方式: 1. dispatch_queue_t queue = dispatch_queue_create("com.dispatch.serial", DISPATCH_QUEUE_

Java多线程--让主线程等待所有子线程执行完毕在执行

朋友让我帮忙写个程序从文本文档中导入数据到oracle数据库中,技术上没有什么难度,文档的格式都是固定的只要对应数据库中的字段解析就行了,关键在于性能. 数据量很大百万条记录,因此考虑到要用多线程并发执行,在写的过程中又遇到问题,我想统计所有子进程执行完毕总共的耗时,在第一个子进程创建前记录当前时间用System.currentTimeMillis()在最后一个子进程结束后记录当前时间,两次一减得到的时间差即为总共的用时,代码如下 long tStart = System.currentTime

Java中一个线程执行死循环有什么后果

假设有两个线程在并发运行,一个线程执行的代码中含有一个死循环如:while(true)....当该线程在执行while(true)中代码时,另一个线程会有机会执行吗? 示例代码(代码来源于互联网) public class Service { Object object1 = new Object(); public void methodA() { synchronized (object1) { System.out.println("methodA begin"); boolea