c#:ThreadPool实现并行分析，并实现线程同步结束

• 背景：

一般情况下，经常会遇到一个单线程程序时执行对CPU,MEMORY,IO利用率上不来，且速度慢下问题；那么，怎么解决这些问题呢？

据我个人经验来说有以下两种方式：

1、并行、多线程（Parallel、Task、ThreadPool）

2、多进程MultipleProcess

恰好工作中又一次遇到单线程程序性能低的问题，本次我主要想尝试使用ThreadPool来实现多线程，并且在实现多线程任务同步结束。

• ThreadPool线程同步结束示例一：

一个ManualResetEvent结合Interlocked来实现线程同步结束。

 static void Main(string[] args)
        {
            using (ManualResetEvent finish = new ManualResetEvent(false))
            {
                int maxThreadCount = 100;
                for (var i = 0; i < 100; i++) {
                    ThreadPool.QueueUserWorkItem((Object state)=> {
                        Console.WriteLine("task:{0}",state);

                        // 以原子操作的形式递减指定变量的值并存储结果。
                        if (Interlocked.Decrement(ref maxThreadCount) == 0) {
                            // 将事件状态设置为有信号，从而允许一个或多个等待线程继续执行。
                            finish.Set();
                        }                        
                    }, i);
                }

                // 阻止当前线程，直到当前 System.Threading.WaitHandle 收到信号。
                finish.WaitOne();
            }

            Console.WriteLine("Complete!");
            Console.ReadKey();

上边的代码是可行性，当系统线程数超过系统允许最大数时，线程会被在线程池中排队等待。

• ThreadPool线程同步结束示例二： 

ManualResetEvent集合（每一个线程由集合中的唯一一个ManualResetEvent对象来实现线程的同步跟踪）结合WaitHandle.WaitAll(ManualResetEvent集合)来实现线程同步结束。

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Text;
using System.Threading;

namespace ThreadPoolTest
{
    class MyTask
    {
        private ManualResetEvent finish = null;
        public MyTask(ManualResetEvent finish)
        {
            this.finish = finish;
        }

        public void MyTaskThreadPoolCallback(Object state)
        {
            Console.WriteLine("task:{0}", state);

            // 将事件状态设置为有信号，从而允许一个或多个等待线程继续执行。
            this.finish.Set();
        }
    }


    class Program
    {
        static void Main(string[] args)
        {
            const int maxThreadCount = 64;
            ManualResetEvent[] finishItems = new ManualResetEvent[maxThreadCount];
            MyTask[] myTaskItems = new MyTask[maxThreadCount]
                ;
            for (var i = 0; i < maxThreadCount; i++)
            {
                finishItems[i] = new ManualResetEvent(false);

                MyTask myTask = new MyTask(finishItems[i]);
                myTaskItems[i] = myTask;

                ThreadPool.QueueUserWorkItem(myTask.MyTaskThreadPoolCallback, i);
            }

            // 等待指定数组中的所有元素都收到信号。
            WaitHandle.WaitAll(finishItems);

            Console.WriteLine("Complete!");
            Console.ReadKey();
        }


    }
}

尽管这种想法不错，但是存在一些问题:比如ManualResetEvent集合数量不允许超过系统允许的最大数量，我的计算机系统允许的最大数量是64，当我把配置超过64时（const int maxThreadCount = 65;），就会抛出异常。

• 实现多线程时，需要注意事项：

可是一般情况下遇到这种业务的情况下，只要修改多线程，必然会遇到某个对象不允许被多个线程操作的问题。

比如：

1、多个线程同时向一个文件中写入内容，这种情况一般使用锁来包成被访问对象的安全性。比如：互斥锁（lock、Mutex）、读写锁（ReadWriteLock）、Monitor、Semaphore(信号灯)、Interlocked（内存共享）等。

2、多个线程同时修改一个非线程安全集合对象（List,Collection,Dictionary,Bag,Queue,Stack,ArrayList,Array,HashTable等）时，往往会抛出异常。针对这种情况，需要使用命名空间System.Collections.Concurrent.*下支持线程安全的集合、字典、队列、栈等对象来替代。

• 业务场景：

我们需要对一个多行文本文件进行解析，根据具体地址解析其中的经纬度信息。如果解析过程中解析失败的行，需要记录到一个_error.txt；解析成功的记录行，记录到_result.txt。使用单线程分析过程中已经遇到了性能低问题，需求解决方案是使用ThreadPool技术。

• 业务实现:

        private static int maxThreadCount = 0;
        private static int fakeMaxThreadCount = int.MaxValue;
        private static ManualResetEvent finish = new ManualResetEvent(false);
        private static object errorLocker = new object();
        private static object resultLocker = new object();
        private static object maxThreadCountLcker = new object();

        public void ParserFile(string filePath)
        {
            using (StreamWriter writerError = new StreamWriter(filePath + "_error"))
            {
                using (StreamWriter writerResult = new StreamWriter(filePath + "_result"))
                {
                    finish = new ManualResetEvent(false);
                    using (StreamReader reader = new StreamReader(filePath))
                    {
                        string line = reader.ReadLine();
                        while (line != null)
                        {
                            maxThreadCount++;
                            ThreadPool.QueueUserWorkItem(DoWork, new object[] { line, writerError, writerResult
});

                            line = reader.ReadLine();
                        }
                    }

                    maxThreadCount++;
                    lock (maxThreadCountLcker)
                    {
                        fakeMaxThreadCount = maxThreadCount;
                    }

                    ThreadPool.QueueUserWorkItem(DoWork, new object[] { });

                    finish.WaitOne();
                    finish.Close();
                    finish.Dispose();
                }
            }
        }



        private void DoWork(object state)
        {
            object[] objectItem = state as object[];
            if (objectItem.Length != 3)
            {
                if (Interlocked.Decrement(ref fakeMaxThreadCount) == 0)
                {
                    finish.Set();
                }
                return;
            }
            string line = objectItem[0].ToString();
            StreamWriter writerError = objectItem[1] as StreamWriter;
            StreamWriter writerResult = objectItem[2] as StreamWriter;

            try
            {
                string[] fields = line.Split(new char[] { '|' });

                string imsi = fields[0];
                string city = fields[1];
                string county = fields[2];
                string address = fields[3];

                // http://restapi.amap.com/v3/geocode/geo?key=7de8697669288fc848e12a08f58d995e&s=rsv3&city=**市&address=**省**市**区**路23号
                string uri = " http://restapi.amap.com/v3/geocode/geo";
                string parameter = string.Format("key={0}&s={1}&city={2}&address={3}", "7de8697669288fc848e12a08f58d995e", "rsv3", "**(市名称)", address);

                // {"status":"1","info":"OK","infocode":"10000","count":"1","geocodes":[{"formatted_address":"***省**市**区***路|23号","province":"***","citycode":"***","city":"***市","district":"***区","township":[],"neighborhood":{"name":[],"type":[]},"building":{"name":[],"type":[]},"adcode":"330105","street":[],"number":[],"location":"120.151367,30.362293","level":"门牌号"}]}
                string result = GetRequesetContext(uri, parameter);
                if (string.IsNullOrEmpty(result) || result.IndexOf("location") == -1)
                {
                    lock (errorLocker)
                    {
                        writerError.WriteLine(result);
                    }
                }
                else
                {
                    int indexCount = 0;
                    List<string> lnglatItems = new List<string>();
                    foreach (string resultItem in result.Split(new string[] { "","", ","" }, StringSplitOptions.RemoveEmptyEntries))
                    {
                        if (resultItem.IndexOf("location") != -1)
                        {
                            indexCount++;
                            lnglatItems.Add(resultItem.Split(new char[] { ':' })[1].Replace(""", string.Empty));
                        }
                    }
                    if (indexCount == 1)
                    {
                        lock (resultLocker)
                        {
                            writerResult.WriteLine(address + "|" + lnglatItems[0] + "|" + imsi);
                        }
                    }
                    else
                    {
                        lock (resultLocker)
                        {                            
                            writerError.WriteLine(address + "|" + string.Join(",", lnglatItems) + "|" + imsi);
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                logger.Error("{0}
{1}", ex.Message, ex.StackTrace);
                lock (errorLocker)
                {
                    writerError.WriteLine(line);
                }
            }
            finally
            {
                lock (maxThreadCountLcker)
                {
                    if (Interlocked.Decrement(ref fakeMaxThreadCount) == 0)
                    {
                        finish.Set();
                    }
                }
            }
        }

 备注：

关于ThreadPool线程池内最大线程控制函数：SetMaxThreads 设置可以同时处于活动状态的线程池的请求数目。 所有大于此数目的请求将保持排队状态，直到线程池线程变为可用。

[SecurityPermissionAttribute(SecurityAction.Demand, ControlThread = true)]
public static bool SetMaxThreads(
    int workerThreads,
    int completionPortThreads
)

workerThreads：线程池中辅助线程的最大数目。

completionPortThreads:线程池中异步 I/O 线程的最大数目。

但是，需要注意事项：

不能将辅助线程的数目或 I/O 完成线程的数目设置为小于计算机的处理器数目。

如果承载了公共语言运行时，例如由 Internet 信息服务 (IIS) 或 SQL Server 承载，主机可能会限制或禁止更改线程池大小。

更改线程池中的最大线程数时需谨慎。 虽然这类更改可能对您的代码有益，但对您使用的代码库可能会有不利的影响。

将线程池大小设置得太大可能导致性能问题。 如果同时执行的线程太多，任务切换开销就成为影响性能的一个主要因素。