快捷導(dǎo)航

Python控制多進程與多線程并發(fā)數(shù)總結(jié)

更新時間：2016年10月26日 10:39:53 作者：張豪飛

本篇文章主要介紹了Python控制多進程與多線程并發(fā)數(shù)，詳細講訴了進程和線程的區(qū)別，并介紹了處理方法，有需要的朋友可以了解一下。

一、前言

本來寫了腳本用于暴力破解密碼，可是1秒鐘嘗試一個密碼2220000個密碼我的天，想用多線程可是只會一個for全開，難道開2220000個線程嗎？只好學(xué)習(xí)控制線程數(shù)了，官方文檔不好看，覺得結(jié)構(gòu)不夠清晰，網(wǎng)上找很多文章也都不很清晰，只有for全開線程，沒有控制線程數(shù)的具體說明，最終終于根據(jù)多篇文章和官方文檔算是搞明白基礎(chǔ)的多線程怎么實現(xiàn)法了，怕長時間不用又忘記，找著麻煩就貼這了，跟我一樣新手也可以參照參照。

先說進程和線程的區(qū)別：

地址空間:進程內(nèi)的一個執(zhí)行單元;進程至少有一個線程;它們共享進程的地址空間;而進程有自己獨立的地址空間;
資源擁有:進程是資源分配和擁有的單位,同一個進程內(nèi)的線程共享進程的資源
線程是處理器調(diào)度的基本單位,但進程不是.
二者均可并發(fā)執(zhí)行.

不能理解的話簡單打比方就是一個進程就像一個程序一樣，并發(fā)互不干擾。一個進程靠一個或多個線程執(zhí)行處理，并發(fā)的線程是cpu在不停的來回切換執(zhí)行，當然是快到你感覺不出的。

拿上面我遇到的困難來說吧，大量的數(shù)據(jù)需要執(zhí)行相同的處理，一個操作中間可能會有一些等待時間，一個一個執(zhí)行浪費大量時間，那么就同時執(zhí)行吧，我們可以用兩種并行辦法：

進程并行或者線程并行

各有優(yōu)缺點，要看情況，不是絕對的，在此不討論這個，這引出下面兩種Python并行處理方法（注釋感覺很清晰詳細了，不再多說）

二、進程處理方法

#coding:utf-8
import random
from time import sleep
import sys
import multiprocessing
import os
#
#需求分析：有大批量數(shù)據(jù)需要執(zhí)行，而且是重復(fù)一個函數(shù)操作（例如爆破密碼），如果全部開始線程數(shù)N多，這里控制住線程數(shù)m個并行執(zhí)行，其他等待
#
lock=multiprocessing.Lock()#一個鎖
def a(x):#模擬需要重復(fù)執(zhí)行的函數(shù)
  lock.acquire()#輸出時候上鎖，否則進程同時輸出時候會混亂，不可讀
  print '開始進程：',os.getpid(),'模擬進程時間:',x
  lock.release()
  
  sleep(x)#模擬執(zhí)行操作
  
  lock.acquire()
  print '結(jié)束進程：',os.getpid(),'預(yù)測下一個進程啟動會使用該進程號'
  lock.release()
list=[]
for i in range(10):#產(chǎn)生一個隨機數(shù)數(shù)組，模擬每次調(diào)用函數(shù)需要的輸入，這里模擬總共有10組需要處理
  list.append(random.randint(1,10))
  
pool=multiprocessing.Pool(processes=3)#限制并行進程數(shù)為3
pool.map(a,list)#創(chuàng)建進程池，調(diào)用函數(shù)a，傳入?yún)?shù)為list,此參數(shù)必須是一個可迭代對象,因為map是在迭代創(chuàng)建每個進程

輸出：

三、線程處理方法：

#coding:utf-8
import threading
import random
import Queue
from time import sleep
import sys
#
#需求分析：有大批量數(shù)據(jù)需要執(zhí)行，而且是重復(fù)一個函數(shù)操作（例如爆破密碼），如果全部開始線程數(shù)N多，這里控制住線程數(shù)m個并行執(zhí)行，其他等待
#
#繼承一個Thread類，在run方法中進行需要重復(fù)的單個函數(shù)操作
class Test(threading.Thread):
  def __init__(self,queue,lock,num):
    #傳遞一個隊列queue和線程鎖，并行數(shù)
    threading.Thread.__init__(self)
    self.queue=queue
    self.lock=lock
    self.num=num
  def run(self):
    #while True:#不使用threading.Semaphore，直接開始所有線程，程序執(zhí)行完畢線程都還不死，最后的print threading.enumerate()可以看出
    with self.num:#同時并行指定的線程數(shù)量，執(zhí)行完畢一個則死掉一個線程
      #以下為需要重復(fù)的單次函數(shù)操作
      n=self.queue.get()#等待隊列進入
      lock.acquire()#鎖住線程，防止同時輸出造成混亂
      print '開始一個線程：',self.name,'模擬的執(zhí)行時間：',n
      print '隊列剩余：',queue.qsize()
      print threading.enumerate()
      lock.release()
      sleep(n)#執(zhí)行單次操作，這里sleep模擬執(zhí)行過程
      self.queue.task_done()#發(fā)出此隊列完成信號
threads=[]
queue=Queue.Queue()
lock=threading.Lock()
num=threading.Semaphore(3)#設(shè)置同時執(zhí)行的線程數(shù)為3，其他等待執(zhí)行
#啟動所有線程
for i in range(10):#總共需要執(zhí)行的次數(shù)
  t=Test(queue,lock,num)
  t.start()
  threads.append(t)
  #吧隊列傳入線程，是run結(jié)束等待開始執(zhí)行，放下面單獨一個for也行，這里少個循環(huán)吧
  n=random.randint(1,10)
  queue.put(n)#模擬執(zhí)行函數(shù)的逐個不同輸入
#吧隊列傳入線程，是run結(jié)束等待開始執(zhí)行
#for t in threads:
#  n=random.randint(1,10)
#  queue.put(n)
#等待線程執(zhí)行完畢
for t in threads:
  t.join()
queue.join()#等待隊列執(zhí)行完畢才繼續(xù)執(zhí)行，否則下面語句會在線程未接受就開始執(zhí)行
print '所有執(zhí)行完畢'
print threading.active_count()
print threading.enumerate()

輸出：