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程序设计有所帮助。

您可能感兴趣的文章:

  • python使用线程封装的一个简单定时器类实例
  • Python多线程编程(三):threading.Thread类的重要函数和方法
  • Python自定义线程池实现方法分析
  • Python探索之自定义实现线程池
  • 详解Python中的多线程编程
  • python多线程编程中的join函数使用心得
  • python杀死一个线程的方法
  • python通过线程实现定时器timer的方法
  • 浅析Python中的多进程与多线程的使用
  • Python threading多线程编程实例
  • python线程锁(thread)学习示例
  • Python自定义线程类简单示例
(0)

相关推荐

  • python线程锁(thread)学习示例

    复制代码 代码如下: # encoding: UTF-8import threadimport time # 一个用于在线程中执行的函数def func():    for i in range(5):        print 'func'        time.sleep(1) # 结束当前线程    # 这个方法与thread.exit_thread()等价    thread.exit() # 当func返回时,线程同样会结束 # 启动一个线程,线程立即开始运行# 这个方法与threa

  • Python自定义线程类简单示例

    本文实例讲述了Python自定义线程类.分享给大家供大家参考,具体如下: 一. 代码 # -*- coding:utf-8 -*- #! python2 import threading class mythread(threading.Thread): def __init__(self, num): threading.Thread.__init__(self) self.num = num def run(self): print('I am {0}'.format(self.num))

  • Python threading多线程编程实例

    Python 的多线程有两种实现方法: 函数,线程类 1.函数 调用 thread 模块中的 start_new_thread() 函数来创建线程,以线程函数的形式告诉线程该做什么 复制代码 代码如下: # -*- coding: utf-8 -*- import thread def f(name):   #定义线程函数   print "this is " + name   if __name__ == '__main__':   thread.start_new_thread(f

  • python杀死一个线程的方法

    最近在项目中遇到这一需求: 我需要一个函数工作,比如远程连接一个端口,远程读取文件等,但是我给的时间有限,比如,4秒钟如果你还没有读取完成或者连接成功,我就不等了,很可能对方已经宕机或者拒绝了.这样可以批量做一些事情而不需要一直等,浪费时间. 结合我的需求,我想到这种办法: 1.在主进程执行,调用一个进程执行函数,然后主进程sleep,等时间到了,就kill 执行函数的进程. 测试一个例子: import time import threading def p(i): print i class

  • python使用线程封装的一个简单定时器类实例

    本文实例讲述了python使用线程封装的一个简单定时器类.分享给大家供大家参考.具体实现方法如下: from threading import Timer class MyTimer: def __init__(self): self._timer= None self._tm = None self._fn = None def _do_func(self): if self._fn: self._fn() self._do_start() def _do_start(self): self.

  • 详解Python中的多线程编程

    一.简介 多线程编程技术可以实现代码并行性,优化处理能力,同时功能的更小划分可以使代码的可重用性更好.Python中threading和Queue模块可以用来实现多线程编程. 二.详解 1.线程和进程        进程(有时被称为重量级进程)是程序的一次执行.每个进程都有自己的地址空间.内存.数据栈以及其它记录其运行轨迹的辅助数据.操作系统管理在其上运行的所有进程,并为这些进程公平地分配时间.进程也可以通过fork和spawn操作来完成其它的任务,不过各个进程有自己的内存空间.数据栈等,所以只

  • Python多线程编程(三):threading.Thread类的重要函数和方法

    这篇文章主要介绍threading模块中的主类Thread的一些主要方法,实例代码如下: 复制代码 代码如下: '''  Created on 2012-9-7    @author:  walfred @module: thread.ThreadTest3  @description: '''    import threading    class MyThread(threading.Thread):      def __init__(self):          threading.

  • python多线程编程中的join函数使用心得

    今天去辛集买箱包,下午挺晚才回来,又是恶心又是头痛.恶心是因为早上吃坏东西+晕车+回来时看到车祸现场,头痛大概是烈日和空调混合刺激而成.没有时间没有精神没有力气学习了,这篇博客就说说python中一个小小函数. 由于坑爹的学校坑爷的专业,多线程编程老师从来没教过,多线程的概念也是教的稀里糊涂,本人python也是菜鸟级别,所以遇到多线程的编程就傻眼了,别人用的顺手的join函数我却偏偏理解不来.早上在去辛集的路上想这个问题想到恶心,回来后继续写代码测试,终于有些理解了(python官方的英文解释

  • 浅析Python中的多进程与多线程的使用

    在批评Python的讨论中,常常说起Python多线程是多么的难用.还有人对 global interpreter lock(也被亲切的称为"GIL")指指点点,说它阻碍了Python的多线程程序同时运行.因此,如果你是从其他语言(比如C++或Java)转过来的话,Python线程模块并不会像你想象的那样去运行.必须要说明的是,我们还是可以用Python写出能并发或并行的代码,并且能带来性能的显著提升,只要你能顾及到一些事情.如果你还没看过的话,我建议你看看Eqbal Quran的文章

  • python通过线程实现定时器timer的方法

    本文实例讲述了python通过线程实现定时器timer的方法.分享给大家供大家参考.具体分析如下: 这个python类实现了一个定时器效果,调用非常简单,可以让系统定时执行指定的函数 下面介绍以threading模块来实现定时器的方法. 使用前先做一个简单试验: import threading def sayhello(): print "hello world" global t #Notice: use global variable! t = threading.Timer(5

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

    本文实例讲述了Python自定义线程池实现方法.分享给大家供大家参考,具体如下: 关于python的多线程,由与GIL的存在被广大群主所诟病,说python的多线程不是真正的多线程.但多线程处理IO密集的任务效率还是可以杠杠的. 我实现的这个线程池其实是根据银角的思路来实现的. 主要思路: 任务获取和执行: 1.任务加入队列,等待线程来获取并执行. 2.按需生成线程,每个线程循环取任务. 线程销毁: 1.获取任务是终止符时,线程停止. 2.线程池close()时,向任务队列加入和已生成线程等量的

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

    为什么需要线程池呢? 设想一下,如果我们使用有任务就开启一个子线程处理,处理完成后,销毁子线程或等得子线程自然死亡,那么如果我们的任务所需时间比较短,但是任务数量比较多,那么更多的时间是花在线程的创建和结束上面,效率肯定就低了.     线程池的原理: 既然是线程池(Thread pool),其实名字很形象,就是把指定数量的可用子线程放进一个"池里",有任务时取出一个线程执行,任务执行完后,并不立即销毁线程,而是放进线程池中,等待接收下一个任务.这样内存和cpu的开销也比较小,并且我们

随机推荐