Python探索之自定义实现线程池

什么是线程池？

线程池是一种线程管理的方法，它可限制线程的数量，避免线程过多消耗过多的系统资源，提供对线程的增加、删除等操作进行优化，以提升系统性能。

自定义线程池的实现

Step 1：导入所需模块

import threading
import queue

Step 2：定义线程池类

class ThreadPool:
    def __init__(self, size):
        self.size = size
        self.tasks = queue.Queue()
        self.threads = []
        self.is_closed = False
        self.__create_threads()

    def __create_threads(self):
        for i in range(self.size):
            thread = threading.Thread(target=self.__worker, name='pool_thread_{}'.format(i))
            thread.start()
            self.threads.append(thread)

    def __worker(self):
        while True:
            try:
                if self.is_closed:
                    break
                task = self.tasks.get(block=True, timeout=1)
                task()
                self.tasks.task_done()
            except queue.Empty:
                pass

    def add_task(self, task):
        if self.is_closed:
            raise ValueError('ThreadPool is already closed.')
        self.tasks.put(task)

    def close(self):
        self.is_closed = True
        for thread in self.threads:
            thread.join()

Step 3：使用自定义线程池

def task1():
    print('Task 1 executing...')

def task2():
    print('Task 2 executing...')

pool = ThreadPool(5)
pool.add_task(task1)
pool.add_task(task2)
pool.close()

示例说明

示例 1：计算斐波那契数列

使用自定义线程池计算斐波那契数列，代码如下：

def fibonacci(n):
    if n == 0 or n == 1:
        return 1
    return fibonacci(n-1) + fibonacci(n-2)

def print_fibonacci(n):
    print('Fibonacci({})={}'.format(n, fibonacci(n)))

pool = ThreadPool(5)
for i in range(35, 41):
    pool.add_task(lambda: print_fibonacci(i))
pool.close()

运行结果：

Fibonacci(35)=14930352
Fibonacci(37)=24157817
Fibonacci(39)=63245986
Fibonacci(36)=9227465
Fibonacci(38)=39088169
Fibonacci(40)=102334155

示例 2：爬取网站数据

使用自定义线程池爬取网站数据，代码如下：

import requests
import re

def download(url):
    try:
        response = requests.get(url)
        if response.status_code == 200:
            return response.text
    except Exception as e:
        print('Error: ', e)

def crawl(url):
    html = download(url)
    if html is None:
        return
    pattern = '<title>(.*?)</title>'
    title = re.findall(pattern, html, re.IGNORECASE)
    if title:
        print('{}: {}'.format(url, title[0]))

pool = ThreadPool(5)
urls = ['https://cn.bing.com/', 'https://www.sogou.com/', 'https://www.baidu.com/', 'https://www.hao123.com/', 'https://www.163.com/']
for url in urls:
    pool.add_task(lambda: crawl(url))
pool.close()

运行结果：

https://www.163.com/: 网易
https://www.sogou.com/: 搜狗搜索引擎 - 上网从搜狗开始
https://www.baidu.com/: 百度一下，你就知道
https://cn.bing.com/: Bing
https://www.hao123.com/: hao123上网导航-用最简单的方式让你上网更方便！