多線程編程---python

參考 《Python 核心編程》（第三版）

一、Thread模板

缺點:
  • 不支持守護線程：當主線程退出時，所有子線程都將終止，不管它們是否在工作。
  • 同步原語少
#-*- coding: UTF - 8 - *-
import thread
from time import sleep, ctime


def loop0():
    print 'start loop0 at:', ctime()
    sleep(4)
    print 'end loop0 at:', ctime()


def loop1():
    print 'start loop1 at:', ctime()
    sleep(2)
    print 'end loop1 at;', ctime()


def main():
    print 'all start at:', ctime()
    thread.start_new_thread(loop0, ())#派生一個新線程
    thread.start_new_thread(loop1, ())
    sleep(6)
    print 'all end at:', ctime()

if __name__ == '__main__':
    main()



#-*- coding: UTF - 8 - *-
import thread
from time import sleep, ctime

loops = [4, 2]


def loop(nloop, nsec, lock):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'end loop', nloop, 'at:', ctime()
    lock.release()  # release()：釋放鎖


def main():
    print 'start at:', ctime()
    locks = []
    nloops = range(len(loops))

    for i in nloops:
        lock = thread.allocate_lock()  # allocate_lock()：分配鎖對象
        lock.acquire()  # acquire()：獲得鎖對象
        locks.append(lock)

    for i in nloops:
        thread.start_new_thread(loop, (i, loops[i], locks[i]))
        for i in nloops:
            while locks[i].locked():
                pass
        print 'all end at:', ctime()

if __name__ == '__main__':
    main()



二、threading模板

優(yōu)點:
  • 支持守護線程：如果把一個線程設置為守護線程，就表示這個線程是不重要的，進程退出時不需要等待這個線程執(zhí)行完成
thread.daemon = True

Thread類

方案一：創(chuàng)建Thread實例，傳給它一個函數(shù) 
#-*- coding: UTF - 8 - *-
import threading
from time import sleep, ctime
loops = [4, 2]


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'end loop', nloop, 'at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = threading.Thread(target=loop, args=(i, loops[i]))
        threads.append(t)

    for i in nloops:
        threads[i].start()#start()：開始執(zhí)行該線程

    for i in nloops:
        threads[i].join()#join(timeout=None)：直至啟動的線程終止之前一直掛起。除非給出了timeout（秒），否則一直堵塞

    print 'ending at:', ctime()

if __name__ == '__main__':
    main()
方案二：創(chuàng)建Thread實例，傳給它一個可調用的類實例 
#-*- coding: UTF - 8 - *-
import threading
from time import sleep, ctime
loops = [4, 2]


class ThreadFunc(object):
    def __init__(self, func, args, name=''):
        self.name = name
        self.func = func
        self.args = args

    def __call__(self):
        self.func(*self.args)


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'end loop', nloop, 'at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = threading.Thread(target=ThreadFunc(
            loop, (i, loops[i]), loop.__name__))
        threads.append(t)

    for i in nloops:
        threads[i].start()  # start()：開始執(zhí)行該線程

    for i in nloops:
        # join(timeout=None)：直至啟動的線程終止之前一直掛起。除非給出了timeout（秒），否則一直堵塞
        threads[i].join()

    print 'ending at:', ctime()


if __name__ == '__main__':
    main()
方案三：派生Thread的子類，并創(chuàng)建子類的實例 
#-*- coding: UTF - 8 - *-
import threading
from time import sleep, ctime
loops = [4, 2]


class MyThread(threading.Thread):
    def __init__(self, func, args, name=''):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

    def run(self):  # run():定義線程功能的方法（通常在子類中被應用開發(fā)者重寫）
        self.func(*self.args)


def loop(nloop, nsec):
    print 'start loop', nloop, 'at:', ctime()
    sleep(nsec)
    print 'end loop', nloop, 'at:', ctime()


def main():
    print 'starting at:', ctime()
    threads = []
    nloops = range(len(loops))

    for i in nloops:
        t = MyThread(loop, (i, loops[i]), loop.__name__)
        threads.append(t)

    for i in nloops:
        threads[i].start()  # start()：開始執(zhí)行該線程

    for i in nloops:
        # join(timeout=None)：直至啟動的線程終止之前一直掛起。除非給出了timeout（秒），否則一直堵塞
        threads[i].join()

    print 'ending at:', ctime()


if __name__ == '__main__':
    main()
修改上面的MyThread類（把結果保存在實例屬性self.res中，并創(chuàng)建新方法getResult()來獲取其值）>>>方便被導入
class MyThread(threading.Thread):
    def __init__(self, func, args, name=''):
        threading.Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

    def getResult(self):
        return self.res

    def run(self):#run():定義線程功能的方法（通常在子類中被應用開發(fā)者重寫）
        self.res=self.func(*self.args)


同步源語：鎖/互斥、信號量

一、鎖：

原理：

當多線程爭奪鎖時，允許第一個獲得鎖的線程進入臨界區(qū)，并執(zhí)行代碼。所有之后到達的線程將被堵塞，直到第一個線程執(zhí)行結束，退出臨界區(qū)，并釋放鎖。此時其他等待的線程可以獲得鎖并進入臨界區(qū)。（被堵塞的線程是沒有順序的）


應用場景：

特殊的函數(shù)、代碼塊不希望（或不應該）被多個線程同時執(zhí)行
  • 修改數(shù)據(jù)庫
  • 更新文件
  • ......

代碼：

from threading import Lock
lock = Lock()
lock.acquire()#獲取鎖
lock.release()#釋放鎖

#上下文管理器
from __future__ import with_statement
with lock:
    ......#鎖的線程塊
二、信號量：
信號量是最古老的同步原語之一
threading模塊包括兩種信號量類：Semaphore和BoundedSemaphore（BoundedSemaphore額外功能：計數(shù)器永遠不會超過初始值，可以防范其中信號量釋放次數(shù)多于獲得次數(shù)的異常用例）


原理：

它是一個計數(shù)器，當資源消耗(acquire)時，計數(shù)器值減1；當資源釋放(release)時，計數(shù)器值加1


應用場景：

  • 線程擁有有限資源
  • ......

代碼（糖果機）：

#-*- coding: UTF - 8 - *-
from atexit import register
from random import randrange
from threading import BoundedSemaphore, Lock, Thread
from time import sleep, ctime

lock = Lock()
Max = 5
candytray = BoundedSemaphore(Max)


def refill():
    lock.acquire()
    print 'Refilling candy...'
    try:
        candytray.release()
    except ValueError:
        print 'full,skipping'
    else:
        print 'OK'
    lock.release()


def buy():
    lock.acquire()
    print 'Buying candy...'
    if candytray.acquire(False):
        print 'OK'
    else:
        print 'empty,skipping'
    lock.release()


def producer(loops):
    for i in xrange(loops):
        refill()
        sleep(randrange(3))


def consumer(loops):
    for i in xrange(loops):
        buy()
        sleep(randrange(3))


def _main():
    print 'starting at:', ctime()
    nloops = randrange(2, 6)
    print 'THE CANDY MACHINE (full with %d bars)!' % Max
    Thread(target=consumer, args=(randrange(nloops, nloops + Max + 2),)).start()
    Thread(target=producer, args=(nloops,)).start()


@register
def _atexit():
    print 'all DONE at:', ctime()


if __name__ == '__main__':
    _main()




生產者-消費者問題（Queue/queue模塊）

原理：

創(chuàng)建一個隊列，生產者（線程）生產時放入商品，消費者（線程）消費時取出商品


應用場景：

生產者-消費者及類似情景【生產時間不確定，消費時間不確定】


代碼：

#-*- coding: UTF - 8 - *-
from random import randint
from time import sleep
from Queue import Queue
from threading3 import MyThread


def writeQ(queue):
    print 'producing object for Q...',
    queue.put('xxx', 1)
    print "size now", queue.qsize()  # qsize()：返回隊列大小


def readQ(queue):
    print 'consumed object from Q... size now', queue.qsize()


def writer(queue, loops):
    for i in range(loops):
        writeQ(queue)
        sleep(randint(1, 3))


def reader(queue, loops):
    for i in range(loops):
        readQ(queue)
        sleep(randint(2, 5))


funcs = [writer, reader]
nfuncs = range(len(funcs))


def main():
    nloops = randint(2, 5)
    # Queue(maxsize=0)：創(chuàng)建一個先入先出的隊列，如果給出最大值，則在隊列沒有空間時堵塞；否則（沒有指定最大值），為無限隊列。
    q = Queue(32)
    threads = []

    for i in nfuncs:
        t = MyThread(funcs[i], (q, nloops), funcs[i].__name__)
        threads.append(t)

    for i in nfuncs:
        threads[i].start()

    for i in nfuncs:
        threads[i].join()

    print 'all Done'


if __name__ == '__main__':
    main()




concurrent.futures模塊

優(yōu)點：

  • "任務"級別進行操作
  • 不需要過分關注同步和線程/進程的管理

原理：

指定一個給定數(shù)量的線程池/進程池------提交任務------整理結果

代碼：

#-*- coding: UTF - 8 - *-
from concurrent.futures import ThreadPoolExecutor#ThreadPoolExecutor-多線程,ProcessPoolExecutor-多進程
from re import compile
from time import ctime
from urllib.request import urlopen as uopen

REGEX = compile('#([\d,]+) in Books ')
AMZN = 'http://amazon.com/dp/'
ISBNS = {
    '0132269937': 'Core Python Programming',
    '0132356139': 'Python Web Development with Django',
    '0137143419': 'Python Fundamentals',
}


def getRanking(isbn):
    with uopen('{0}{1}'.format(AMZN, isbn)) as page:
        return str(REGEX.findall(page.read())[0],'utf-8')


def _main():
    print ('Start at', ctime(), 'on Amazon...')
    with ThreadPoolExecutor(3) as executor:#ThreadPoolExecutor(n)：n代表線程池個數(shù)
        for isbn, ranking in zip(ISBNS, executor.map(getRanking, ISBNS)):
            print ('- %r ranked - %s' % (ISBNS[isbn], ranking))
    print('all Done at:', ctime())


if __name__ == '__main__':
    _main()



實踐

1、Amazon圖書排行排名

#-*- coding: UTF - 8 - *-
from atexit import register#atexit.register()函數(shù)：告知腳本結束時間
from re import compile
from threading import Thread
from time import ctime
from urllib2 import urlopen as uopen

REGEX = compile('#([\d,]+) in Books ')
AMZN = 'http://amazon.com/dp/'
ISBNS = {
    '0132269937':'Core Python Programming',
    '0132356139':'Python Web Development with Django',
    '0137143419':'Python Fundamentals',
}

def getRanking(isbn):
    page = uopen('%s%s' % (AMZN,isbn))
    data = page.read()
    page.close()
    return REGEX.findall(data)[0]

def _showRanking(isbn):#函數(shù)名前面的單劃線--->特殊函數(shù)--->只能被本模塊的代碼使用，不能被其他使用本文件作為庫或者工具模塊的應用導入
    print '- %r ranked %s' %(ISBNS[isbn],getRanking(isbn))

def _main():
    print 'At',ctime(),'on Amazon......'
    for isbn in ISBNS:
        #單線程
        # _showRanking(isbn)
        #多線程
        Thread(target=_showRanking,args=(isbn,)).start()

@register
def _atexit():
    print 'all DONE at:',ctime()

if __name__=='__main__':
    _main()
最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯(lián)系作者
【社區(qū)內容提示】社區(qū)部分內容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內容（如有圖片或視頻亦包括在內）由作者上傳并發(fā)布，文章內容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內容

  • 1 引言 首先區(qū)分幾個概念： 程序——能被計算機執(zhí)行的文件；進程——處于執(zhí)行狀態(tài)的程序，包含計算機為其分配的空間、...
    rebirth_2017閱讀 300評論 0贊 0
  • Java-Review-Note——4.多線程 標簽： JavaStudy PS：本來是分開三篇的，后來想想還是整...
    coder_pig閱讀 1,761評論 2贊 17
  • 1.解決信號量丟失和假喚醒 public class MyWaitNotify3{ MonitorObject m...
    Q羅閱讀 995評論 0贊 1
  • 下面是我自己收集整理的Java線程相關的面試題，可以用它來好好準備面試。 參考文檔：-《Java核心技術 卷一》-...
    阿呆變Geek閱讀 15,120評論 14贊 507
  • 如果我還年輕 我要像一把鋒利的劍 插在土地上 也要發(fā)出萬丈光芒 如果我還年輕 我會帶著理想 一個永遠不可能實現(xiàn)的理...
    更向遠行閱讀 400評論 0贊 0

友情鏈接更多精彩內容