Python实现自定义多线程与多进程处理类
发布时间：2026-10-13| 4人浏览过
本文介绍了一个基于Python实现的自定义多线程与多进程处理类，旨在简化大量文件操作的并发处理流程。该类封装了线程池与进程池的核心逻辑，支持设置线程延迟、最大线程数及进程数，并提供纯多进程和多进程结合多线程两种运行模式。通过示例展示了如何快速复用并发逻辑，避免重复编写底层代码，适用于需要批量处理文件或执行耗时任务的场景。
本文展示了一个使用Python编写的自定义多线程与多进程类示例，供读者参考和借鉴。
在处理大量文件时，以往每次编写程序都需要在函数内部重新实现多线程或多进程逻辑，导致大量重复工作且新任务需重新编码。为此，本文将多线程与多进程的基础功能封装为一个独立的类，以便后续直接调用，简化开发流程。
若在使用过程中发现Bug，代码将进行后续更新。
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2017/5/10 12:47
# @Author : zhaowen.zhu
# @Site :
# @File : MultiThread.py
# @Software: Python Idle
import threading,time,sys,multiprocessing
from multiprocessing import Pool
class MyTMultithread(threading.Thread):
'''''
自定义的线程函数,
功能:使用多线程运行函数,函数的参数只有一个file,并且未实现结果值的返回
args:
filelist 函数的参数为列表格式，
funname 函数的名字为字符串，函数仅有一个参数为file
delay 每个线程之间的延迟，
max_threads 线程的最大值
'''
def __init__(self,filelist,delay,funname,max_threads = 50):
threading.Thread.__init__(self)
self.funname = funname
self.filelist = filelist[:]
self.delay = delay
self.max_threads = max_threads
def startrun(self):
def runs():
time.sleep(self.delay)
while True:
try:
file = self.filelist.pop()
except IndexError as e:
break
else:
self.funname(file)
threads = []
while threads or self.filelist:
for thread in threads:
if not thread.is_alive():
threads.remove(thread)
while len(threads) < self.max_threads and self.filelist:
thread = threading.Thread(target = runs)
thread.setDaemon(True)
thread.start()
threads.append(thread)
class Mymultiprocessing (MyTMultithread):
'''''
多进程运行函数，多进程多线程运行函数
args:
filelist 函数的参数为列表格式，
funname 函数的名字为字符串，函数仅有一个参数为file
delay 每个线程\进程之间的延迟，
max_threads 最大的线程数
max_multiprocess 最大的进程数
'''
def __init__(self,filelist,delay,funname,max_multiprocess = 1,max_threads = 1):
self.funname = funname
self.filelist = filelist[:]
self.delay = delay
self.max_threads = max_threads
self.max_multiprocess = max_multiprocess
self.num_cpus = multiprocessing.cpu_count()
def multiprocessingOnly(self):
'''''
只使用多进程
'''
num_process = min(self.num_cpus,self.max_multiprocess)
processes = []
while processes or self.filelist:
for p in processes:
if not p.is_alive():
# print(p.pid,p.name,len(self.filelist))
processes.remove(p)
while len(processes) < num_process and self.filelist:
try:
file = self.filelist.pop()
except IndexError as e:
break
else:
p = multiprocessing.Process(target=self.funname,args=(file,))
p.start()
processes.append(p)
def multiprocessingThreads(self):
num_process = min(self.num_cpus,self.max_multiprocess)
p = Pool(num_process)
DATALISTS = []
tempmod = len(self.filelist) % (num_process)
CD = int((len(self.filelist) + 1 + tempmod)/ (num_process))
for i in range(num_process):
if i == num_process:
DATALISTS.append(self.filelist[i*CD:-1])
DATALISTS.append(self.filelist[(i*CD):((i+1)*CD)])
try:
processes = []
for i in range(num_process):
#print('wait add process:',i+1,time.clock())
#print(eval(self.funname),DATALISTS[i])
MultThread = MyTMultithread(DATALISTS[i],self.delay,self.funname,self.max_threads)
p = multiprocessing.Process(target=MultThread.startrun())
#print('pid & name:',p.pid,p.name)
processes.append(p)
for p in processes:
print('wait join ')
p.start()
print('waite over')
except Exception as e:
print('error :',e)
print ('end process')
def func1(file):
print(file)
if __name__ == '__main__':
a = list(range(0,97))
'''''
测试使用5线程
'''
st = time.clock()
asc = MyTMultithread(a,0,'func1',5)
asc.startrun()
end = time.clock()
print('*'*50)
print('多线程使用时间:',end-st)
#测试使用5个进程
st = time.clock()
asd = Mymultiprocessing(a,0,'func1',5)
asd.multiprocessingOnly()
end = time.clock()
print('*'*50)
print('多进程使用时间:',end-st)
#测试使用5进程10线程
st = time.clock()
multiPT = Mymultiprocessing(a,0,'func1',5,10)
multiPT.multiprocessingThreads()
end = time.clock()
print('*'*50)
print('多进程多线程使用时间:',end-st)
更多关于Python相关内容感兴趣的读者可查看本站专题：《Python进程与线程操作技巧总结》、《Python Socket编程技巧总结》、《Python数据结构与算法教程》、《Python函数使用技巧总结》、《Python字符串操作技巧汇总》、《Python入门与进阶经典教程》及《Python文件与目录操作技巧汇总》
希望本文所述内容对大家的Python程序设计有所帮助。
免责声明：本文内容仅供参考，不代表本站观点。转载或引用请注明出处。
