Python實(shí)現(xiàn)的自定義多線程多進(jìn)程類示例

更新時(shí)間：2018年03月23日 08:50:55 作者：Magicapprentice

這篇文章主要介紹了Python實(shí)現(xiàn)的自定義多線程多進(jìn)程類,結(jié)合實(shí)例形式分析了Python多線程多進(jìn)程的相關(guān)調(diào)用與使用操作技巧,需要的朋友可以參考下

本文實(shí)例講述了Python實(shí)現(xiàn)的自定義多線程多進(jìn)程類。分享給大家供大家參考，具體如下：

最近經(jīng)常使用到對大量文件進(jìn)行操作的程序以前每次寫的時(shí)候都要在函數(shù)中再寫一個(gè)多線程多進(jìn)程的函數(shù)，做了些重復(fù)的工作遇到新的任務(wù)時(shí)還要重寫，因此將多線程與多進(jìn)程的一些簡單功能寫成一個(gè)類，方便使用。功能簡單只為以后方便使用。

使用中發(fā)現(xiàn)bug會再進(jìn)行更新

#!/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):
    '''''
    自定義的線程函數(shù),
    功能:使用多線程運(yùn)行函數(shù),函數(shù)的參數(shù)只有一個(gè)file,并且未實(shí)現(xiàn)結(jié)果值的返回
    args:
      filelist  函數(shù)的參數(shù)為列表格式，
      funname  函數(shù)的名字為字符串，函數(shù)僅有一個(gè)參數(shù)為file
      delay   每個(gè)線程之間的延遲，
      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):
  '''''
  多進(jìn)程運(yùn)行函數(shù)，多進(jìn)程多線程運(yùn)行函數(shù)
  args:
    filelist  函數(shù)的參數(shù)為列表格式，
    funname  函數(shù)的名字為字符串，函數(shù)僅有一個(gè)參數(shù)為file
    delay   每個(gè)線程\進(jìn)程之間的延遲，
    max_threads 最大的線程數(shù)
    max_multiprocess 最大的進(jìn)程數(shù)
  '''
    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):
      '''''
    只使用多進(jìn)程
      '''
      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('多線程使用時(shí)間:',end-st)
    #測試使用5個(gè)進(jìn)程
    st = time.clock()
    asd = Mymultiprocessing(a,0,'func1',5)
    asd.multiprocessingOnly()
    end = time.clock()
    print('*'*50)
    print('多進(jìn)程使用時(shí)間:',end-st)
    #測試使用5進(jìn)程10線程
    st = time.clock()
    multiPT = Mymultiprocessing(a,0,'func1',5,10)
    multiPT.multiprocessingThreads()
    end = time.clock()
    print('*'*50)
    print('多進(jìn)程多線程使用時(shí)間:',end-st)

更多關(guān)于Python相關(guān)內(nèi)容感興趣的讀者可查看本站專題：《Python進(jìn)程與線程操作技巧總結(jié)》、《Python Socket編程技巧總結(jié)》、《Python數(shù)據(jù)結(jié)構(gòu)與算法教程》、《Python函數(shù)使用技巧總結(jié)》、《Python字符串操作技巧匯總》、《Python入門與進(jìn)階經(jīng)典教程》及《Python文件與目錄操作技巧匯總》

希望本文所述對大家Python程序設(shè)計(jì)有所幫助。

您可能感興趣的文章: