Python开启线程和线程池的方法

简介: Python开启线程和线程池的方法

​### 一.最佳线程数的获取:

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

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

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

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

1.多线程中,线程的数量并非越多越好

2.节省每次开启线程的开销

三、如何实现线程池?

  • threadpool模块

  • concurrent.futures

  • 重写threadpool或者future的函数

  • vthread 模块

1、过去:

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

#! /usr/bin/env python
# -*- coding: utf-8 -*-

import threadpool
import time

def sayhello (a):
    print("hello: "+a)
    time.sleep(2)

def main():
    global result
    seed=["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-start
    print("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 time

def 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、自己构建一个线程池:

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

import threading
import Queue
import hashlib
import logging
from utils.progress import PrintProgress
from utils.save import SaveToSqlite


class ThreadPool(object):
    def __init__(self, thread_num, args):

        self.args = args
        self.work_queue = Queue.Queue()
        self.save_queue = Queue.Queue()
        self.threads = []
        self.running = 0
        self.failure = 0
        self.success = 0
        self.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] = url
            self.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 task

    def 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 += 1

    def increase_failure(self):
        self.failure += 1

    def increase_running(self):
        self.running += 1

    def decrease_running(self):
        self.running -= 1

    def 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.success
        progress_info['failure'] = self.failure

        return progress_info

    def 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_task

    def 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,deep
                do, url, deep = self.thread_pool.get_task()
                self.thread_pool.increase_running()

                # 判断deep,是否获取新的链接
                flag_get_new_link = True
                if deep >= self.thread_pool.args.deep:
                    flag_get_new_link = False
                #学习中遇到问题没人解答?小编创建了一个Python学习交流群:711312441

                # 此处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:
                    break
            except Exception, e:
                self.thread_pool.decrease_running()
                # print str(e)
                break

安装vthread函数库
系统命令行下执行:

pip install vthread

一句话实现简单多线程

import vthread,requests

@vthread.thread(5)     #开5个线程执行同一个函数
def compete(url):
    r = requests.get(url)
    if r.status_code == 200 :
        print("[*]Success")
    else:
        print("[*]Fail. Retrying...")

compete("http://www.baidu.com/")

相同效果:

import vthread,requests

@vthread.thread
def compete(url):
    r = requests.get(url)
    if r.status_code == 200 :
        print("[*]Success")
    else:
        print("[*]Fail. Retrying...")

for i in range(5):  #线程数
    compete("http://www.baidu.com/")

线程池包装

import vthread,requests

@vthread.pool(10) #包装10条线程池
def compete(url):
    r = requests.get(url)
    if r.status_code == 200 :
        print("[*]Success")
    else:
        print("[*]Fail. Retrying...")

for i in range(20):  #20线程
    compete("http://www.baidu.com/")
相关文章
|
23天前
|
JSON 数据可视化 API
Python 中调用 DeepSeek-R1 API的方法介绍,图文教程
本教程详细介绍了如何使用 Python 调用 DeepSeek 的 R1 大模型 API,适合编程新手。首先登录 DeepSeek 控制台获取 API Key,安装 Python 和 requests 库后,编写基础调用代码并运行。文末包含常见问题解答和更简单的可视化调用方法,建议收藏备用。 原文链接:[如何使用 Python 调用 DeepSeek-R1 API?](https://apifox.com/apiskills/how-to-call-the-deepseek-r1-api-using-python/)
|
5天前
|
存储 监控 Java
【Java并发】【线程池】带你从0-1入门线程池
欢迎来到我的技术博客!我是一名热爱编程的开发者,梦想是编写高端CRUD应用。2025年我正在沉淀中,博客更新速度加快,期待与你一起成长。 线程池是一种复用线程资源的机制,通过预先创建一定数量的线程并管理其生命周期,避免频繁创建/销毁线程带来的性能开销。它解决了线程创建成本高、资源耗尽风险、响应速度慢和任务执行缺乏管理等问题。
106 60
【Java并发】【线程池】带你从0-1入门线程池
|
2月前
|
监控 Kubernetes Java
阿里面试:5000qps访问一个500ms的接口,如何设计线程池的核心线程数、最大线程数? 需要多少台机器?
本文由40岁老架构师尼恩撰写,针对一线互联网企业的高频面试题“如何确定系统的最佳线程数”进行系统化梳理。文章详细介绍了线程池设计的三个核心步骤:理论预估、压测验证和监控调整,并结合实际案例(5000qps、500ms响应时间、4核8G机器)给出具体参数设置建议。此外,还提供了《尼恩Java面试宝典PDF》等资源,帮助读者提升技术能力,顺利通过大厂面试。关注【技术自由圈】公众号,回复“领电子书”获取更多学习资料。
|
2月前
|
并行计算 安全 Java
Python GIL(全局解释器锁)机制对多线程性能影响的深度分析
在Python开发中,GIL(全局解释器锁)一直备受关注。本文基于CPython解释器,探讨GIL的技术本质及其对程序性能的影响。GIL确保同一时刻只有一个线程执行代码,以保护内存管理的安全性,但也限制了多线程并行计算的效率。文章分析了GIL的必要性、局限性,并介绍了多进程、异步编程等替代方案。尽管Python 3.13计划移除GIL,但该特性至少要到2028年才会默认禁用,因此理解GIL仍至关重要。
134 16
Python GIL(全局解释器锁)机制对多线程性能影响的深度分析
|
18天前
|
Python
python3多线程中使用线程睡眠
本文详细介绍了Python3多线程编程中使用线程睡眠的基本方法和应用场景。通过 `time.sleep()`函数,可以使线程暂停执行一段指定的时间,从而控制线程的执行节奏。通过实际示例演示了如何在多线程中使用线程睡眠来实现计数器和下载器功能。希望本文能帮助您更好地理解和应用Python多线程编程,提高程序的并发能力和执行效率。
43 20
|
2月前
|
人工智能 自然语言处理 算法
随机的暴力美学蒙特卡洛方法 | python小知识
蒙特卡洛方法是一种基于随机采样的计算算法,广泛应用于物理学、金融、工程等领域。它通过重复随机采样来解决复杂问题,尤其适用于难以用解析方法求解的情况。该方法起源于二战期间的曼哈顿计划,由斯坦尼斯拉夫·乌拉姆等人提出。核心思想是通过大量随机样本来近似真实结果,如估算π值的经典示例。蒙特卡洛树搜索(MCTS)是其高级应用,常用于游戏AI和决策优化。Python中可通过简单代码实现蒙特卡洛方法,展示其在文本生成等领域的潜力。随着计算能力提升,蒙特卡洛方法的应用范围不断扩大,成为处理不确定性和复杂系统的重要工具。
77 21
|
23天前
|
安全 Java C#
Unity多线程使用(线程池)
在C#中使用线程池需引用`System.Threading`。创建单个线程时,务必在Unity程序停止前关闭线程(如使用`Thread.Abort()`),否则可能导致崩溃。示例代码展示了如何创建和管理线程,确保在线程中执行任务并在主线程中处理结果。完整代码包括线程池队列、主线程检查及线程安全的操作队列管理,确保多线程操作的稳定性和安全性。
|
4天前
|
数据采集 Java 数据处理
Python实用技巧:轻松驾驭多线程与多进程,加速任务执行
在Python编程中,多线程和多进程是提升程序效率的关键工具。多线程适用于I/O密集型任务,如文件读写、网络请求;多进程则适合CPU密集型任务,如科学计算、图像处理。本文详细介绍这两种并发编程方式的基本用法及应用场景,并通过实例代码展示如何使用threading、multiprocessing模块及线程池、进程池来优化程序性能。结合实际案例,帮助读者掌握并发编程技巧,提高程序执行速度和资源利用率。
15 0
|
2月前
|
数据挖掘 数据处理 开发者
Python3 自定义排序详解:方法与示例
Python的排序功能强大且灵活,主要通过`sorted()`函数和列表的`sort()`方法实现。两者均支持`key`参数自定义排序规则。本文详细介绍了基础排序、按字符串长度或元组元素排序、降序排序、多条件排序及使用`lambda`表达式和`functools.cmp_to_key`进行复杂排序。通过示例展示了如何对简单数据类型、字典、类对象及复杂数据结构(如列车信息)进行排序。掌握这些技巧可以显著提升数据处理能力,为编程提供更强大的支持。
41 10
|
11天前
|
SQL 关系型数据库 MySQL
Python中使用MySQL模糊查询的方法
本文介绍了两种使用Python进行MySQL模糊查询的方法:一是使用`pymysql`库,二是使用`mysql-connector-python`库。通过这两种方法,可以连接MySQL数据库并执行模糊查询。具体步骤包括安装库、配置数据库连接参数、编写SQL查询语句以及处理查询结果。文中详细展示了代码示例,并提供了注意事项,如替换数据库连接信息、正确使用通配符和关闭数据库连接等。确保在实际应用中注意SQL注入风险,使用参数化查询以保障安全性。

热门文章

最新文章