Python自定义线程池实现方法分析

什么是线程池？

线程池是一个线程队列，当有任务到来时，就会将任务加入队列中，线程池内的线程就会去队列中获取任务并执行。线程池的主要优势在于减少了线程的频繁创建和销毁的操作，提升了程序的效率。

Python线程池的实现

Python中线程池的实现需要用到两个模块，分别是threading和queue。

1. threading模块用来实现线程池中执行任务的线程。
2. queue模块用来实现线程池的任务队列。

以下是线程池的实现方法：

import queue
import threading

class ThreadPool:
    def __init__(self, max_workers):
        self._max_workers = max_workers
        self._job_queue = queue.Queue()
        self._workers = []

    def submit(self, func, *args, **kwargs):
        self._job_queue.put((func, args, kwargs))
        if len(self._workers) < self._max_workers:
            worker = threading.Thread(target=self._worker_thread)
            worker.start()
            self._workers.append(worker)

    def _worker_thread(self):
        while True:
            try:
                func, args, kwargs = self._job_queue.get(True)
            except:
                break
            func(*args, **kwargs)
            self._job_queue.task_done()

    def wait_completion(self):
        self._job_queue.join()

上述代码中，ThreadPool类维护了一个任务队列_job_queue和一组线程_workers。submit方法用来向线程池中提交任务，在队列中添加元组(func, args, kwargs)，其中func是待执行的函数，args和kwargs是用于调用函数的参数。如果线程池还有空余的线程，会将一个新线程加入到_workers列表中。_worker_thread方法是实际执行任务的方法，当线程从队列中取出元组，会调用其中的函数func(*args, **kwargs)。wait_completion方法用于等待所有任务执行完毕。

示例说明

示例1

以下代码是一个简单的例子，演示如何使用自定义的线程池执行多个任务。

from time import sleep

def task(num):
    sleep(1)
    print(f'Task {num} is executing...')

if __name__ == '__main__':
    thread_pool = ThreadPool(max_workers=3)
    for i in range(10):
        thread_pool.submit(task, i)
    thread_pool.wait_completion()

上述代码中，自定义了一个task函数，它会执行等待1秒钟后输出一个执行信息。在if __name__ == '__main__':判断中，创建了一个ThreadPool对象，最多同时执行3个任务。接着循环加入了10个任务，这些任务会被添加到任务队列中并且被线程池中的线程执行。最后等待所有的任务执行完成。

示例2

以下代码演示了如何使用自定义线程池爬取多个网站的内容。

import requests

URLS = [
    'http://www.baidu.com',
    'http://www.sohu.com',
    'http://www.163.com',
    'http://www.qq.com',
    'http://www.taobao.com',
    'http://www.jd.com'
]

def fetch_url(url):
    response = requests.get(url, timeout=10)
    print(f'{url}: {len(response.text)} bytes')

if __name__ == '__main__':
    thread_pool = ThreadPool(max_workers=3)
    for url in URLS:
        thread_pool.submit(fetch_url, url)
    thread_pool.wait_completion()

上述代码中，定义了一个fetch_url函数，它会根据指定的url发起网络请求，并输出响应的长度信息。在if __name__ == '__main__':判断中，创建了一个ThreadPool对象，最多同时执行3个任务。接着循环遍历网站列表并加入队列中，然后线程池中的线程会去获取这些任务并执行。输出每个网站的响应长度信息。最后等待所有任务执行完成。

总结

本文介绍了Python自定义线程池的实现方法，并且通过两个示例演示了如何使用。线程池的实现需要用到threading和queue两个模块。使用自定义线程池可以有效地减少线程的频繁创建和销毁的操作，提升程序的效率。