python 线程池

一、既然多线程可以缩短程序运行时间，那么，是不是线程数量越多越好呢？

显然，并不是，每一个线程的从生成到消亡也是需要时间和资源的，太多的线程会占用过多的系统资源（内存开销，cpu开销），而且生成太多的线程时间也是可观的，很可能会得不偿失，这里给出一个最佳线程数量的计算方式：

最佳线程数的获取：

1、通过用户慢慢递增来进行性能压测，观察QPS（即每秒的响应请求数，也即是最大吞吐能力。），响应时间

2、根据公式计算:服务器端最佳线程数量=((线程等待时间+线程cpu时间)/线程cpu时间) * cpu数量

3、单用户压测，查看CPU的消耗，然后直接乘以百分比，再进行压测，一般这个值的附近应该就是最佳线程数量。

二、为什么要使用线程池？

对于任务数量不断增加的程序，每有一个任务就生成一个线程，最终会导致线程数量的失控，例如，整站爬虫，假设初始只有一个链接a，那么，这个时候只启动一个线程，运行之后，得到这个链接对应页面上的b，c，d，，，等等新的链接，作为新任务，这个时候，就要为这些新的链接生成新的线程，线程数量暴涨。在之后的运行中，线程数量还会不停的增加，完全无法控制。所以，对于任务数量不端增加的程序，固定线程数量的线程池是必要的。

三、如何实现线程池？

这里，我分别介绍三种实现方式：

1、过去：

使用threadpool模块，这是个python的第三方模块，支持python2和python3，具体使用方式如下：

#! /usr/bin/env python
# -*- coding: utf-8 -*-import threadpool
import timedef sayhello (a):print("hello: "+a)time.sleep(2)def main():global resultseed=["a","b","c"]start=time.time()task_pool=threadpool.ThreadPool(5)requests=threadpool.makeRequests(sayhello,seed)for req in requests:task_pool.putRequest(req)task_pool.wait()end=time.time()time_m = end-startprint("time: "+str(time_m))start1=time.time()for each in seed:sayhello(each)end1=time.time()print("time1: "+str(end1-start1))if __name__ == '__main__':main()

运行结果如下：

threadpool是一个比较老的模块了，现在虽然还有一些人在用，但已经不再是主流了，关于python多线程，现在已经开始步入未来（future模块）了

2、未来：

使用concurrent.futures模块，这个模块是python3中自带的模块，但是，python2.7以上版本也可以安装使用，具体使用方式如下：

#! /usr/bin/env python
# -*- coding: utf-8 -*-from concurrent.futures import ThreadPoolExecutor
import timedef sayhello(a):print("hello: "+a)time.sleep(2)def main():seed=["a","b","c"]start1=time.time()for each in seed:sayhello(each)end1=time.time()print("time1: "+str(end1-start1))start2=time.time()with ThreadPoolExecutor(3) as executor:for each in seed:executor.submit(sayhello,each)end2=time.time()print("time2: "+str(end2-start2))start3=time.time()with ThreadPoolExecutor(3) as executor1:executor1.map(sayhello,seed)end3=time.time()print("time3: "+str(end3-start3))if __name__ == '__main__':main()

运行结果如下：

注意到一点：

concurrent.futures.ThreadPoolExecutor，在提交任务的时候，有两种方式，一种是submit（）函数，另一种是map（）函数，两者的主要区别在于：

2.1、map可以保证输出的顺序, submit输出的顺序是乱的

2.2、如果你要提交的任务的函数是一样的，就可以简化成map。但是假如提交的任务函数是不一样的，或者执行的过程之可能出现异常（使用map执行过程中发现问题会直接抛出错误）就要用到submit（）

2.3、submit和map的参数是不同的，submit每次都需要提交一个目标函数和对应的参数，map只需要提交一次目标函数，目标函数的参数放在一个迭代器（列表，字典）里就可以。

3.现在？

这里要考虑一个问题，以上两种线程池的实现都是封装好的，任务只能在线程池初始化的时候添加一次，那么，假设我现在有这样一个需求，需要在线程池运行时，再往里面添加新的任务（注意，是新任务，不是新线程），那么要怎么办？

其实有两种方式：

3.1、重写threadpool或者future的函数：

这个方法需要阅读源模块的源码，必须搞清楚源模块线程池的实现机制才能正确的根据自己的需要重写其中的方法。

3.2、自己构建一个线程池：

这个方法就需要对线程池的有一个清晰的了解了，附上我自己构建的一个线程池：

#! /usr/bin/env python
# -*- coding: utf-8 -*-import threading
import Queue
import hashlib
import logging
from utils.progress import PrintProgress
from utils.save import SaveToSqliteclass ThreadPool(object):def __init__(self, thread_num, args):self.args = argsself.work_queue = Queue.Queue()self.save_queue = Queue.Queue()self.threads = []self.running = 0self.failure = 0self.success = 0self.tasks = {}self.thread_name = threading.current_thread().getName()self.__init_thread_pool(thread_num)# 线程池初始化def __init_thread_pool(self, thread_num):# 下载线程for i in range(thread_num):self.threads.append(WorkThread(self))# 打印进度信息线程self.threads.append(PrintProgress(self))# 保存线程self.threads.append(SaveToSqlite(self, self.args.dbfile))# 添加下载任务def add_task(self, func, url, deep):# 记录任务，判断是否已经下载过url_hash = hashlib.new('md5', url.encode("utf8")).hexdigest()if not url_hash in self.tasks:self.tasks[url_hash] = urlself.work_queue.put((func, url, deep))logging.info("{0} add task {1}".format(self.thread_name, url.encode("utf8")))# 获取下载任务def get_task(self):# 从队列里取元素，如果block=True,则一直阻塞到有可用元素为止。task = self.work_queue.get(block=False)return taskdef task_done(self):# 表示队列中的某个元素已经执行完毕。self.work_queue.task_done()# 开始任务def start_task(self):for item in self.threads:item.start()logging.debug("Work start")def increase_success(self):self.success += 1def increase_failure(self):self.failure += 1def increase_running(self):self.running += 1def decrease_running(self):self.running -= 1def get_running(self):return self.running# 打印执行信息def get_progress_info(self):progress_info = {}progress_info['work_queue_number'] = self.work_queue.qsize()progress_info['tasks_number'] = len(self.tasks)progress_info['save_queue_number'] = self.save_queue.qsize()progress_info['success'] = self.successprogress_info['failure'] = self.failurereturn progress_infodef add_save_task(self, url, html):self.save_queue.put((url, html))def get_save_task(self):save_task = self.save_queue.get(block=False)return save_taskdef wait_all_complete(self):for item in self.threads:if item.isAlive():# join函数的意义，只有当前执行join函数的线程结束，程序才能接着执行下去item.join()# WorkThread 继承自threading.Thread
class WorkThread(threading.Thread):# 这里的thread_pool就是上面的ThreadPool类def __init__(self, thread_pool):threading.Thread.__init__(self)self.thread_pool = thread_pool#定义线程功能方法，即，当thread_1，...，thread_n，调用start（）之后，执行的操作。def run(self):print (threading.current_thread().getName())while True:try:# get_task()获取从工作队列里获取当前正在下载的线程，格式为func,url,deepdo, url, deep = self.thread_pool.get_task()self.thread_pool.increase_running()# 判断deep，是否获取新的链接flag_get_new_link = Trueif deep >= self.thread_pool.args.deep:flag_get_new_link = False# 此处do为工作队列传过来的func，返回值为一个页面内容和这个页面上所有的新链接html, new_link = do(url, self.thread_pool.args, flag_get_new_link)if html == '':self.thread_pool.increase_failure()else:self.thread_pool.increase_success()# html添加到待保存队列self.thread_pool.add_save_task(url, html)# 添加新任务，即，将新页面上的不重复的链接加入工作队列。if new_link:for url in new_link:self.thread_pool.add_task(do, url, deep + 1)self.thread_pool.decrease_running()# self.thread_pool.task_done()except Queue.Empty:if self.thread_pool.get_running() <= 0:breakexcept Exception, e:self.thread_pool.decrease_running()# print str(e)break

python 线程池相关推荐

Python 线程池 ThreadPoolExecutor(二) - Python零基础入门教程
目录一.Python 线程池前言二.Python 线程池 ThreadPoolExecutor 常用函数 1.线程池 as_completed 函数使用 2.线程池 map 函数使用 3.线程池 ...
Python 线程池 ThreadPoolExecutor(一) - Python零基础入门教程
目录一.Python 线程池前言二.Python 线程池原理三.Python 线程池 ThreadPoolExecutor 函数介绍四.Python 线程池 ThreadPoolExecuto ...
python线程池原理及使用
python线程池及其原理和使用系统启动一个新线程的成本是比较高的,因为它涉及与操作系统的交互.在这种情形下,使用线程池可以很好地提升性能,尤其是当程序中需要创建大量生存期很短暂的线程时,更应该考虑 ...
python 线程池 concurrent.futures ThreadPoolExecutor
python 线程池 concurrent.futures ThreadPoolExecutor 步骤: 1,导包from concurrent.futures import ThreadPoolEx ...
Python线程池与进程池
Python线程池与进程池前言很多人学习python,不知道从何学起. 很多人学习python,掌握了基本语法过后,不知道在哪里寻找案例上手. 很多已经做案例的人,却不知道如何去学习更加高深的知识 ...
python线程池（threadpool）模块使用笔记详解
这篇文章主要介绍了python线程池(threadpool)模块使用笔记详解,小编觉得挺不错的,现在分享给大家,也给大家做个参考.一起跟随小编过来看看吧最近在做一个视频设备管理的项目,设备包括(摄像 ...
python线程池使用和问题记录
记录一次使用多线程的问题背景最近工作有个需求根据文件中的数据请求中台服务,然后解析返回值.文件中每行代表一个参数,使用post方式携带参数请求中台接口. 分析:需要处理的数据量非常大(近200w行 ...
浅谈python线程池
python线程池的使用 python的多线程管理一直很麻烦,可能是我基础不够好,这里记录并分享以下python的线程池管理在网上查了一个线程池的使用资料,个人感觉不是很清晰但是重点很到位, 原文 ...
python线程池wait_python线程池 ThreadPoolExecutor 的用法示例
前言从Python3.2开始,标准库为我们提供了 concurrent.futures 模块,它提供了 ThreadPoolExecutor (线程池)和ProcessPoolExecutor (进 ...
python3 线程池源码解析_5分钟看懂系列：Python 线程池原理及实现
概述传统多线程方案会使用"即时创建, 即时销毁"的策略.尽管与创建进程相比,创建线程的时间已经大大的缩短,但是如果提交给线程的任务是执行时间较短,而且执行次数极其频繁,那么服务器 ...

python 线程池

python 线程池相关推荐

最新文章

热门文章