快捷導(dǎo)航

Python 多線(xiàn)程Threading初學(xué)教程

更新時(shí)間：2017年08月22日 15:56:48 作者：閃電gogogo

多線(xiàn)程可簡(jiǎn)單理解為同時(shí)執(zhí)行多個(gè)任務(wù)。本文給大家分享Python 多線(xiàn)程Threading初學(xué)教程實(shí)例詳解，感興趣的朋友一起學(xué)習(xí)吧

1.1 什么是多線(xiàn)程 Threading

多線(xiàn)程可簡(jiǎn)單理解為同時(shí)執(zhí)行多個(gè)任務(wù)。

多進(jìn)程和多線(xiàn)程都可以執(zhí)行多個(gè)任務(wù)，線(xiàn)程是進(jìn)程的一部分。線(xiàn)程的特點(diǎn)是線(xiàn)程之間可以共享內(nèi)存和變量，資源消耗少（不過(guò)在Unix環(huán)境中，多進(jìn)程和多線(xiàn)程資源調(diào)度消耗差距不明顯，Unix調(diào)度較快），缺點(diǎn)是線(xiàn)程之間的同步和加鎖比較麻煩。

1.2 添加線(xiàn)程 Thread

導(dǎo)入模塊

import threading

獲取已激活的線(xiàn)程數(shù)

threading.active_count()

查看所有線(xiàn)程信息

threading.enumerate()

查看現(xiàn)在正在運(yùn)行的線(xiàn)程

threading.current_thread()

添加線(xiàn)程，threading.Thread()接收參數(shù)target代表這個(gè)線(xiàn)程要完成的任務(wù)，需自行定義

def thread_job():
  print('This is a thread of %s' % threading.current_thread())
def main():
  thread = threading.Thread(target=thread_job,)  # 定義線(xiàn)程 
  thread.start() # 讓線(xiàn)程開(kāi)始工作
  if __name__ == '__main__':
  main()

1.3 join 功能

因?yàn)榫€(xiàn)程是同時(shí)進(jìn)行的，使用join功能可讓線(xiàn)程完成后再進(jìn)行下一步操作，即阻塞調(diào)用線(xiàn)程，直到隊(duì)列中的所有任務(wù)被處理掉。

import threading
import time
def thread_job():
  print('T1 start\n')
  for i in range(10):
    time.sleep(0.1)
  print('T1 finish\n')
def T2_job():
  print('T2 start\n')
  print('T2 finish\n')
def main():
  added_thread=threading.Thread(target=thread_job,name='T1')
  thread2=threading.Thread(target=T2_job,name='T2')
  added_thread.start()
  #added_thread.join()
  thread2.start()
  #thread2.join()
  print('all done\n')
if __name__=='__main__':
   main()

例子如上所示，當(dāng)不使用join功能的時(shí)候，結(jié)果如下圖所示：

當(dāng)執(zhí)行了join功能之后，T1運(yùn)行完之后才運(yùn)行T2，之后再運(yùn)行print（‘a(chǎn)ll done'）

1.4 儲(chǔ)存進(jìn)程結(jié)果 queue

queue是python標(biāo)準(zhǔn)庫(kù)中的線(xiàn)程安全的隊(duì)列（FIFO）實(shí)現(xiàn),提供了一個(gè)適用于多線(xiàn)程編程的先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)，即隊(duì)列，用來(lái)在生產(chǎn)者和消費(fèi)者線(xiàn)程之間的信息傳遞

（1）基本FIFO隊(duì)列

 class queue.Queue(maxsize=0)

maxsize是整數(shù)，表明隊(duì)列中能存放的數(shù)據(jù)個(gè)數(shù)的上限，達(dá)到上限時(shí)，插入會(huì)導(dǎo)致阻塞，直至隊(duì)列中的數(shù)據(jù)被消費(fèi)掉，如果maxsize小于或者等于0，隊(duì)列大小沒(méi)有限制

（2）LIFO隊(duì)列 last in first out后進(jìn)先出

class queue.LifoQueue(maxsize=0)

（3）優(yōu)先級(jí)隊(duì)列

class queue.PriorityQueue(maxsize=0)

視頻中的代碼，看的還不是特別明白

import threading
import time
from queue import Queue
def job(l,q):
  for i in range(len(l)):
    l[i]=l[i]**2
  q.put(l)
def multithreading():
  q=Queue()
  threads=[]
  data=[[1,2,3],[3,4,5],[4,5,6],[5,6,7]]
  for i in range(4):
    t=threading.Thread(target=job,args=(data[i],q))
    t.start()
    threads.append(t)
  for thread in threads:
    thread.join()
  results=[]
  for _ in range(4):
    results.append(q.get())
  print(results)
if __name__=='__main__':
   multithreading()

運(yùn)行結(jié)果如下所示

圖片截取來(lái)源：http://www.cnblogs.com/itogo/p/5635629.html

1.5 GIL 不一定有效率

Global Interpreter Lock全局解釋器鎖，python的執(zhí)行由python虛擬機(jī)（也成解釋器主循環(huán)）控制，GIL的控制對(duì)python虛擬機(jī)的訪(fǎng)問(wèn)，保證在任意時(shí)刻，只有一個(gè)線(xiàn)程在解釋器中運(yùn)行。在多線(xiàn)程環(huán)境中能，python虛擬機(jī)按照以下方式執(zhí)行：

1.設(shè)置 GIL

2.切換到一個(gè)線(xiàn)程去運(yùn)行

3.運(yùn)行：

a.指定數(shù)量的字節(jié)碼指令，或

b.線(xiàn)程主動(dòng)讓出控制（可以調(diào)用time.sleep(0)）

4.把線(xiàn)程設(shè)置為睡眠狀態(tài)

5.解鎖GIL

6.重復(fù)1-5

在調(diào)用外部代碼（如C/C++擴(kuò)展函數(shù)）的時(shí)候，GIL將會(huì)被鎖定，直到這個(gè)函數(shù)結(jié)束為止（由于在這期間沒(méi)有python的字節(jié)碼被運(yùn)行，所以不會(huì)做線(xiàn)程切換）。

下面為視頻中所舉例的代碼，將一個(gè)數(shù)擴(kuò)大4倍，分為正常方式、以及分配給4個(gè)線(xiàn)程去做，發(fā)現(xiàn)耗時(shí)其實(shí)并沒(méi)有相差太多量級(jí)。

import threading
from queue import Queue
import copy
import time
def job(l, q):
  res = sum(l)
  q.put(res)
def multithreading(l):
  q = Queue()
  threads = []
  for i in range(4):
    t = threading.Thread(target=job, args=(copy.copy(l), q), name='T%i' % i)
    t.start()
    threads.append(t)
  [t.join() for t in threads]
  total = 0
  for _ in range(4):
    total += q.get()
  print(total)
def normal(l):
  total = sum(l)
  print(total)
if __name__ == '__main__':
  l = list(range(1000000))
  s_t = time.time()
  normal(l*4)
  print('normal: ',time.time()-s_t)
  s_t = time.time()
  multithreading(l)
  print('multithreading: ', time.time()-s_t)

運(yùn)行結(jié)果為：

1.6 線(xiàn)程鎖 Lock

如果線(xiàn)程1得到了結(jié)果，想要讓線(xiàn)程2繼續(xù)使用1的結(jié)果進(jìn)行處理，則需要對(duì)1lock，等到1執(zhí)行完，再開(kāi)始執(zhí)行線(xiàn)程2。一般來(lái)說(shuō)對(duì)share memory即對(duì)共享內(nèi)存進(jìn)行加工處理時(shí)會(huì)用到lock。

import threading
def job1():
  global A, lock #全局變量
  lock.acquire() #開(kāi)始lock
  for i in range(10):
    A += 1
    print('job1', A)
  lock.release() #釋放
def job2(): 
  global A, lock
  lock.acquire()
  for i in range(10):
    A += 10
    print('job2', A)
  lock.release()
if __name__ == '__main__':
  lock = threading.Lock()
  A = 0
  t1 = threading.Thread(target=job1)
  t2 = threading.Thread(target=job2)
  t1.start()
  t2.start()
  t1.join()
  t2.join()

運(yùn)行結(jié)果如下所示：