python多线程并行计算通过向线程池ThreadPoolExecutor提交任务的实现方法

Posted 2023-04-23

参考技术A Python的线程池可以有效地控制系统中并发线程的数量。
当程序中需要创建许多生存期较短的线程执行运算任务时，首先考虑使用线程池。线程池任务启动时会创建出最大线程数参数 max_workers 指定数量的空闲线程，程序只要将执行函数提交给线程池，线程池就会启动一个空闲的线程来执行它。当该函数执行结束后，该线程并不会死亡，而是再次返回到线程池中变成空闲状态，等待执行下一个函数。配合使用 with 关键字实现任务队列完成后自动关闭线程池释放资源。

C#并行编程：.NET线程池

线程 Thread

在总结线程池之前，先来看一下.NET线程。

.NET线程与操作系统(Windows)线程有什么区别？

.NET利用Windows的线程处理功能。在C#程序编写中，我们首先会新建一个线程对象System.Threading.Thread，并为其指定一个回调方法；当我们调用线程对象的Start方法启动线程时，会创建一个操作系统线程来执行回调方法。.NET中的线程实际上等价于Windows系统线程，都是CPU调度和分配的对象。

前台线程和后台线程

.NET把线程分为前台线程和后台线程，两者几乎相同，唯一的区别是，前台线程会阻止进程的正常退出，后台线程则不会。下面用一个例子描述前、后台线程的区别：

class Program
{
    static void Main(string[] args)
 {
        ThreadDemo threadDemo = new ThreadDemo();
        
        threadDemo.RunBackgroundThread();

        {
            Thread.Sleep(5000);
            Thread.CurrentThread.Abort();
        }

        Console.ReadKey();
    }
}

public class ThreadDemo
{
    private readonly Thread _foregroundThread;
    private readonly Thread _backgroundThread;

    public ThreadDemo()
 {
        this._foregroundThread = new Thread(WriteNumberWorker) { Name = "ForegroundThread"};
        this._backgroundThread = new Thread(WriteNumberWorker) { Name = "BackgroundThread", IsBackground = true };
    }

    
    
    
    private static void WriteNumberWorker()
 {
        for (int i = 0; i < 20; i++)
        {
            Console.WriteLine($"{DateTime.Now}=> {Thread.CurrentThread.Name} writes {i + 1}.");
            Thread.Sleep(500);
        }
    }

    
    
    
    public void RunForegroundThread()
 {
        this._foregroundThread?.Start();
    }

    
    
    
    public void RunBackgroundThread()
 {
        this._backgroundThread?.Start();
    }
}

线程池 ThreadPool

线程的创建和销毁要耗费很多时间，而且过多的线程不仅会浪费内存空间，还会导致线程上下文切换频繁，影响程序性能。为改善这些问题，.NET运行时（CLR）会为每个进程开辟一个全局唯一的线程池来管理其线程。

线程池内部维护一个操作请求队列，程序执行异步操作时，添加目标操作到线程池的请求队列；线程池代码提取记录项并派发给线程池中的一个线程；如果线程池中没有可用线程，就创建一个新线程，创建的新线程不会随任务的完成而销毁，这样就可以避免线程的频繁创建和销毁。如果线程池中大量线程长时间无所事事，空闲线程会进行自我终结以释放资源。

线程池通过保持进程中线程的少量和高效来优化程序的性能。

C#中线程池是一个静态类，维护两种线程，工作线程和异步IO线程，这些线程都是后台线程。线程池不会影响进程的正常退出。

线程池的使用

线程池提供两个静态方法SetMaxThreads和SetMinThreads让我们设置线程池的最大线程数和最小线程数。最大线程数指的是，该线程池能够创建的最大线程数，当线程数达到设定值且忙碌，异步任务将进入请求队列，直到有线程空闲才会执行；最小线程数指的是，线程池优先尝试以设置数量的线程处理请求，当请求数达到一定量（未做深入研究）时，才会创建新的线程。

下面的例子展示了线程池的特性及常见使用方式。

class Program
{
    static void Main(string[] args)
 {
        
        RunCancellableWork();

        Console.ReadKey();
    }

    static void RunThreadPoolDemo()
 {
        ThreadPoolDemo.ThreadPoolDemo.ShowThreadPoolInfo();
        ThreadPool.SetMaxThreads(100, 100);
        ThreadPool.SetMinThreads(8, 8);
        ThreadPoolDemo.ThreadPoolDemo.ShowThreadPoolInfo();
        ThreadPoolDemo.ThreadPoolDemo.MakeThreadPoolDoSomeWork(100);
        ThreadPoolDemo.ThreadPoolDemo.MakeThreadPoolDoSomeIOWork();
    }

    static void RunCancellableWork()
 {
        Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] started a work");
        Console.WriteLine("Press 'Esc' to cancel the work.");
        Console.WriteLine();
        ThreadPoolDemo.ThreadPoolDemo.DoSomeWorkWithCancellation();
        if (Console.ReadKey(true).Key == ConsoleKey.Escape)
        {
            ThreadPoolDemo.ThreadPoolDemo.CTSource.Cancel();
        }
    }
}

public class ThreadPoolDemo
{
    
    
    
    public static void ShowThreadPoolInfo()
 {
        int workThreads, completionPortThreads;

        
        ThreadPool.GetAvailableThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetAvailableThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        
        ThreadPool.GetMaxThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetMaxThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        
        ThreadPool.GetMinThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"GetMinThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        Console.WriteLine();
    }

    
    
    
    public static void MakeThreadPoolDoSomeWork(int workCount = 10)
 {
        for (int i = 0; i < workCount; i++)
        {
            int index = i;
            
            ThreadPool.QueueUserWorkItem(s =>
            {
                Thread.Sleep(100);
                Debug.Print($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] is running. [{index}]");
                ShowAvailableThreads("WorkerThread");
            });
        }
    }

    
    
    
    public static void MakeThreadPoolDoSomeIOWork()
 {
        
        IList<string> uriList = new List<string>()
        {
            "http://news.baidu.com/",
            "https://www.hao123.com/",
            "https://map.baidu.com/",
            "https://tieba.baidu.com/",
            "https://wenku.baidu.com/",
            "http://fanyi-pro.baidu.com",
            "http://bit.baidu.com/",
            "http://xueshu.baidu.com/",
            "http://www.cnki.net/",
            "http://www.wanfangdata.com.cn",
        };

        foreach (string uri in uriList)
        {
            WebRequest request = WebRequest.Create(uri);
            request.BeginGetResponse(ac =>
            {
                try
                {
                    WebResponse response = request.EndGetResponse(ac);
                    ShowAvailableThreads("IOThread");
                    Debug.Print($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] is running. [{response.ContentLength}]");
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                }
            }, request);
        }
    }

    
    
    
    private static void ShowAvailableThreads(string sourceTag = null)
 {
        int workThreads, completionPortThreads;
        ThreadPool.GetAvailableThreads(out workThreads, out completionPortThreads);
        Console.WriteLine($"{sourceTag} GetAvailableThreads => workThreads:{workThreads};completionPortThreads:{completionPortThreads}");
        Console.WriteLine();
    }

    
    
    
    public static CancellationTokenSource CTSource { get; set; } = new CancellationTokenSource();

    
    
    
    public static void DoSomeWorkWithCancellation()
 {
        ThreadPool.QueueUserWorkItem(t =>
        {
            Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] begun running. [0 - 9999]");

            for (int i = 0; i < 10000; i++)
            {
                if (CTSource.Token.IsCancellationRequested)
                {
                    Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] recived the cancel token. [{i}]");
                    break;
                }
                Thread.Sleep(100);
            }

            Console.WriteLine($"{DateTime.Now}=> Thread-[{Thread.CurrentThread.ManagedThreadId}] was cancelled.");
        });
    }
}

线程池的调度

前面提到，线程池内部维护者一个工作项队列，这个队列指的是线程池全局队列。实际上，除了全局队列，线程池会给每个工作者线程维护一个本地队列。

当我们调用ThreadPool.QueueUserWorkItem方法时，工作项会被放入全局队列；使用定时器Timer的时候，也会将工作项放入全局队列；但是，当我们使用任务Task的时候，假如使用默认的任务调度器，任务会被调度到工作者线程的本地队列中。

工作者线程优先执行本地队列中最新进入的任务，如果本地队列中已经没有任务，线程会尝试从其他工作者线程任务队列的队尾取任务执行，这里需要进行同步。如果所有工作者线程的本地队列都没有任务可以执行，工作者线程才会从全局队列取最新的工作项来执行。所有任务执行完毕后，线程睡眠，睡眠一定时间后，线程醒来并销毁自己以释放资源。

线程池处理异步IO的内部原理

上面的例子中，从网站获取信息需要用到线程池的异步IO线程，线程池内部利用IOCP（IO完成端口）与硬件设备建立连接。异步IO实现过程如下：

1. 托管的IO请求线程调用Win32本地代码ReadFile方法

2. ReadFile方法分配IO请求包IRP并发送至Windows内核

3. Windows内核把收到的IRP放入对应设备驱动程序的IRP队列中，此时IO请求线程已经可以返回托管代码

4. 驱动程序处理IRP并将处理结果放入.NET线程池的IRP结果队列中

5. 线程池分配IO线程处理IRP结果

小结

.NET线程池是并发编程的主要实现方式。C#中Timer、Parallel、Task在内部都是利用线程池实现的异步功能，深入理解线程池在并行编程中十分重要。

原文地址：https://www.cnblogs.com/chenbaoshun/p/10566124.html