python多線程與threading模塊

python多線程與_thread模塊　中介紹了線程的基本概念以及_thread模塊的簡單示例。然而，_thread模塊過於簡單，使得我們無法用它來准確地控制線程，本文介紹threading模塊，它提供了更強大的多線程管理方案。

 

threading模塊的對象

Thread　　表示一個執行線程的對象　

Lock　　鎖原語

RLock　　可重入鎖對象，使單一線程可以再次獲得已持有的鎖（遞歸鎖）

Condition　　條件變量對象，使得一個線程等待另一個線程滿足特定條件

Event　　條件變量的通用版本，任意數量的線程等待某個事件的發生，該事件發生后所有線程將被激活

Semaphore　　為線程間的共享資源提供了一個計數器，如果沒有可用資源時會被阻塞

BoundedSemaphone　　與Semaphore相似，不過它不允許超過初始值

Timer　　與Thread相似，不過運行前要等待一段時間

Barrier　　創建一個”障礙“,必須要達到指定數量的線程才能繼續

Thread對象

Thread類表示在單獨的控制線程中運行的活動。

主要方法：

threading.Thread(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

//target是將被run()方法調用的可調用對象。默認為None，表示不調用任何東西。

//name是線程的名字。默認情況下，以“Thread-N”的形式構造一個唯一的名字，N是一個小的十進制整數。

//args是給調用目標的參數元組。默認為()。

//kwargs是給調用目標的關鍵字參數的一個字典。默認為{}

//daemon表示是否為守護線程

 

start()　　開始執行線程　　

run()　　//定義線程功能

join(timeout=None)　　//直至啟動的線程之前一直掛起，除非給出timeout時間，否則一直阻塞

gerName()　　//返回線程名

屬性：

name　　//線程名

ident　　//線程標識符

daemon　　//是否為守護線程

 

Threading模塊主要函數

threading.active_count()　　// 返回當前處於alive狀態的Thread對象的個數。返回的數目等於enumerate()返回的列表的長度。

threading.current_thread()　　// 返回當前的Thread對象，對應於調用者控制的線程。

threading.get_ident()　　// 返回當前線程的'線程標識符'。它是一個非零的整數。它的價值沒有直接的意義。

threading.enumerate()　　// 返回當前活着的Thread對象的列表。該列表包括守護線程、由current_thread()創建的虛假線程對象和主線程。它不包括已終止的線程和尚未開始的線程。

threading.main_thread()　　// 返回主 Thread 對象。在正常情況下，主線程是從 Python 解釋器中啟動的線程。

守護線程

如果把一個線程設置為守護線程，就表示這個線程是不重要的，進程退出時不需要等待這些線程執行完成。

執行如下賦值語句可以將一個線程設置為守護線程。thread.daemon=True。

一個新線程會繼承父線程的守護標記。 

使用鎖

鎖有兩種狀態：鎖定和未鎖定。同時也只支持兩種函數，獲得鎖和釋放鎖。

lock.acquire()　　//獲取鎖

lock.release()　　//釋放鎖

當多線程競爭時，允許第一個獲得鎖的線程進入臨界區，執行相應代碼。所有之后到達的線程將被阻塞，直到第一個線程執行結束，退出臨界區，釋放鎖。此時，等待的線程可以獲得鎖並進入臨界區，哪個正在等待的線程獲取鎖是隨機的。

#!/usr/bin/env python3
# coding:utf-8
from atexit import register
from random import randrange
from threading import Thread, current_thread, Lock
from time import sleep, ctime


class CleanOutputSet(set):
    def __str__(self):    //改變輸出格式
        return ', '.join(x for x in self)


#lock = Lock()
loops = (randrange(2, 5) for x in range(randrange(3, 7)))
remaining = CleanOutputSet()


def loop(nsec):
    myname = current_thread().name    //獲取當前線程名
    remaining.add(myname)    //添加到集合
    print('[{0}] Start {1}'.format(ctime(), myname))
    sleep(nsec)
    remaining.remove(myname)
    print('[{0}] Completed {1} ({2} secs)'.format(ctime(), myname, nsec))
    print('  (remaining: {0})'.format(remaining or 'None'))


def main():
    for pause in loops:
        Thread(target=loop, args=(pause,)).start()


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


if __name__ == '__main__':
    main()

[Wed Jan 24 20:41:48 2018] Start Thread-1
[Wed Jan 24 20:41:48 2018] Start Thread-2
[Wed Jan 24 20:41:48 2018] Start Thread-3
[Wed Jan 24 20:41:48 2018] Start Thread-4
[Wed Jan 24 20:41:48 2018] Start Thread-5
[Wed Jan 24 20:41:48 2018] Start Thread-6
[Wed Jan 24 20:41:50 2018] Completed Thread-3 (2 secs)
  (remaining: Thread-6, Thread-1, Thread-2, Thread-5, Thread-4)
[Wed Jan 24 20:41:50 2018] Completed Thread-4 (2 secs)
  (remaining: Thread-6, Thread-1, Thread-2, Thread-5)
[Wed Jan 24 20:41:51 2018] Completed Thread-5 (3 secs)
  (remaining: Thread-6, Thread-1, Thread-2)
[Wed Jan 24 20:41:52 2018] Completed Thread-1 (4 secs)
[Wed Jan 24 20:41:52 2018] Completed Thread-6 (4 secs)
  (remaining: Thread-2)
  (remaining: Thread-2)
[Wed Jan 24 20:41:52 2018] Completed Thread-2 (4 secs)
  (remaining: None)
all DONE at: Wed Jan 24 20:41:52 2018

輸出結果中可以看到，多個線程並行I/O導致結果混亂。I/O和訪問相同的數據結構都屬於臨界區，因此需要用鎖來防止多個線程同時進入臨界區。

#!/usr/bin/env python3
# coding:utf-8
from atexit import register
from random import randrange
from threading import Thread, current_thread, Lock
from time import sleep, ctime


class CleanOutputSet(set):
    def __str__(self):  # 改變輸出格式
        return ', '.join(x for x in self)


lock = Lock()
loops = (randrange(2, 5) for x in range(randrange(3, 7)))
remaining = CleanOutputSet()


def loop(nsec):
    myname = current_thread().name   #　獲取當前線程名
    lock.acquire()
    remaining.add(myname)   #　加入集合
    print('[{0}] Start {1}'.format(ctime(), myname))
    lock.release()
    sleep(nsec)
    lock.acquire()
    remaining.remove(myname)
    print('[{0}] Completed {1} ({2} secs)'.format(ctime(), myname, nsec))
    print('  (remaining: {0})'.format(remaining or 'None'))
    lock.release()


def main():
    for pause in loops:
        Thread(target=loop, args=(pause,)).start()


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


if __name__ == '__main__':
    main()

[Wed Jan 24 20:51:11 2018] Start Thread-1
[Wed Jan 24 20:51:11 2018] Start Thread-2
[Wed Jan 24 20:51:11 2018] Start Thread-3
[Wed Jan 24 20:51:11 2018] Start Thread-4
[Wed Jan 24 20:51:11 2018] Start Thread-5
[Wed Jan 24 20:51:13 2018] Completed Thread-2 (2 secs)
  (remaining: Thread-3, Thread-1, Thread-4, Thread-5)
[Wed Jan 24 20:51:14 2018] Completed Thread-3 (3 secs)
  (remaining: Thread-1, Thread-4, Thread-5)
[Wed Jan 24 20:51:15 2018] Completed Thread-4 (4 secs)
  (remaining: Thread-1, Thread-5)
[Wed Jan 24 20:51:15 2018] Completed Thread-1 (4 secs)
  (remaining: Thread-5)
[Wed Jan 24 20:51:15 2018] Completed Thread-5 (4 secs)
  (remaining: None)
all DONE at: Wed Jan 24 20:51:15 2018

使用信號量

鎖只能控制臨界區的訪問，面對一些更復雜的情況就無能為力了。比如，讀者與寫者共享一段8單位長的緩沖區，讀者每次取走1單位，寫者每次產生1單位。使用鎖顯然無法解決，這就用到信號量。

信號量時最古老的原語之一。它是一個計數器，當資源消耗時遞減，資源釋放時遞增。

threading.Semaphore(value=1)　　//構造函數（也可使用threading.BoundedSemaphore(value=1)，value可自行設置，BoundedSemaphore創建的信號量不可超過初值，超過將拋出ValueError）

semaphore.acquire(block=true,timeout=None)　　//資源消耗，block為true時，若無法獲得資源，將阻塞timeout時間等待其他線程釋放資源。block為false時，無法獲得資源將拋出ValueError

semaphore.release()　　//資源增加

#!/usr/bin/env python3
# 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 range(loops):
        refill()
        sleep(randrange(3))


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


def main():
    print('starting at:', ctime())
    nloops = randrange(2, 6)
    print('THE CANDY MACHINE (full with {0} bars)'.format(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()

以上為糖果機信號量示例。糖果機5個槽,refill()函數添加糖果，buy()函數買走糖果。這兩個函數對型號量的作用是相反的。

starting at: Wed Jan 24 21:16:00 2018
THE CANDY MACHINE (full with 5 bars)
Buying candy...
OK
Refilling candy...
OK
Refilling candy...
full, skipping
Buying candy...
OK
Refilling candy...
OK
Buying candy...
OK
Buying candy...
OK
Buying candy...
OK
Buying candy...
OK
Buying candy...
OK
Buying candy...
empty, skipping
Buying candy...
empty, skipping
all DONE at: Wed Jan 24 21:16:05 2018

使用Condition

可以把Condiftion理解為一把高級的瑣，它提供了比Lock, RLock更高級的功能，允許我們能夠控制復雜的線程同步問題。threadiong.Condition在內部維護一個瑣對象（默認是RLock），可以在創建Condigtion對象的時候把瑣對象作為參數傳入。Condition也提供了acquire, release方法，其含義與瑣的acquire, release方法一致，其實它只是簡單的調用內部瑣對象的對應的方法而已。Condition還提供了如下方法(特別要注意：這些方法只有在占用瑣(acquire)之后才能調用，否則將會報RuntimeError異常。)：

Condition.wait([timeout])　　//wait方法釋放內部所占用的瑣，同時線程被掛起，直至接收到通知被喚醒或超時（如果提供了timeout參數的話）。當線程被喚醒並重新占有瑣的時候，程序才會繼續執行下去。

Condition.notify()　　//喚醒一個掛起的線程（如果存在掛起的線程）。注意：notify()方法不會釋放所占用的瑣。

Condition.notify_all()

Condition.notifyAll()　　//喚醒所有掛起的線程（如果存在掛起的線程）。注意：這些方法不會釋放所占用的瑣。

 

from threading import Condition, Thread
from time import ctime


con = Condition()


def get_odd():
    count = 3
    while(True):
        con.acquire(timeout=1)
        print(1)
        count -= 1
        if count is 0:
            break
        con.notify()
        con.wait()


def get_even():
    count = 3
    while(True):
        con.acquire(timeout=1)
        print(2)
        count -= 1
        if count is 0:
            break
        con.notify()
        con.wait()



def main():
    print('start at:', ctime())
    Thread(target=get_odd, name='').start()
    Thread(target=get_even, name='').start()


if __name__ == '__main__':
    main()

 

創建線程的三種方法

以生產者消費者模型為例，緩沖區大小為５，開始為空。生產者產生一個數置入緩沖區，消費者取走一個數。生產者運行10次，消費者運行5次 

方法一：創建Thread實例，並傳給它一個函數(包括函數參數)

#!/usr/bin/env python3
# coding:utf-8'

from atexit import register
from time import ctime
from random import randint
from threading import BoundedSemaphore, Semaphore, Lock, Thread

lock = Lock()
buffer = [0 for x in range(5)]
MAX = len(buffer)
sem_full = Semaphore(0)
sem_empty = BoundedSemaphore(MAX)


def producer():
    sem_empty.acquire()
    lock.acquire()
    print("produce a number")
    for i in range(len(buffer)):
        if buffer[i] is 0:
            buffer[i] = randint(1, 10)
            break
    print(buffer)
    sem_full.release()
    lock.release()


def consumer():
    sem_full.acquire()
    lock.acquire()
    print("consume a number")
    for i in range(len(buffer)):
        if buffer[i] is not 0:
            buffer[i] = 0
            break
    print(buffer)
    sem_empty.release()
    lock.release()


def main():
    print('starting at:', ctime())
    for i in range(5):
        Thread(target=consumer, args='').start()
        Thread(target=producer, args='').start()
        Thread(target=producer, args='').start()


@register
def _atexit():
    print("all DOne at:", ctime())


if __name__ == '__main__':
    main()

方法二：創建Thread實例，傳給它一個可調用的類實例（將調用__call__方法）

#!/usr/bin/env python3
# coding:utf-8

from atexit import register
from time import ctime
from random import randint
from threading import BoundedSemaphore, Semaphore, Lock, Thread

lock = Lock()
buffer = [0 for x in range(5)]
MAX = len(buffer)
sem_full = Semaphore(0)
sem_empty = BoundedSemaphore(MAX)


class ThreadFunc(object):

    def __init__(self, func, args, name=""):
        self.name = name
        self.func = func
        self.args = args

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


def producer():
    sem_empty.acquire()
    lock.acquire()
    print("produce a number")
    for i in range(len(buffer)):
        if buffer[i] is 0:
            buffer[i] = randint(1, 10)
            break
    print(buffer)
    sem_full.release()
    lock.release()


def consumer():
    sem_full.acquire()
    lock.acquire()
    print("consume a number")
    for i in range(len(buffer)):
        if buffer[i] is not 0:
            buffer[i] = 0
            break
    print(buffer)
    sem_empty.release()
    lock.release()


threads = []


def main():
    print('starting at:', ctime())
    for i in range(5):
        c = Thread(target=ThreadFunc(consumer, '', consumer.__name__))
        threads.append(c)
        p1 = Thread(target=ThreadFunc(producer, '', producer.__name__))
        threads.append(p1)
        p2 = Thread(target=ThreadFunc(producer, '', producer.__name__))
        threads.append(p2)

    for i in range(len(threads)):
        threads[i].start()

    for i in range(len(threads)):
        threads[i].join()


@register
def _atexit():
    print("all DOne at:", ctime())


if __name__ == '__main__':
    main()

方法三：派生Thread子類，創建子類的實例（重寫run方法）

#!/usr/bin/env python3
# coding:utf-8

from atexit import register
from time import ctime
from random import randint
from threading import BoundedSemaphore, Semaphore, Lock, Thread

lock = Lock()
buffer = [0 for x in range(5)]
MAX = len(buffer)
sem_full = Semaphore(0)
sem_empty = BoundedSemaphore(MAX)


class MyThread(Thread):

    def __init__(self, func, args, name=""):
        Thread.__init__(self)
        self.name = name
        self.func = func
        self.args = args

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


def producer():
    sem_empty.acquire()
    lock.acquire()
    print("produce a number")
    for i in range(len(buffer)):
        if buffer[i] is 0:
            buffer[i] = randint(1, 10)
            break
    print(buffer)
    sem_full.release()
    lock.release()


def consumer():
    sem_full.acquire()
    lock.acquire()
    print("consume a number")
    for i in range(len(buffer)):
        if buffer[i] is not 0:
            buffer[i] = 0
            break
    print(buffer)
    sem_empty.release()
    lock.release()


threads = []


def main():
    print('starting at:', ctime())
    for i in range(5):
        c = MyThread(consumer, '', consumer.__name__)
        threads.append(c)
        p1 = MyThread(producer, '', producer.__name__)
        threads.append(p1)
        p2 = MyThread(producer, '', producer.__name__)
        threads.append(p2)

    for i in range(len(threads)):
        threads[i].start()

    for i in range(len(threads)):
        threads[i].join()


@register
def _atexit():
    print("all DOne at:", ctime())


if __name__ == '__main__':
    main()

輸出結果

starting at: Wed Jan 24 22:25:05 2018
produce a number
[7, 0, 0, 0, 0]
produce a number
[7, 9, 0, 0, 0]
consume a number
[0, 9, 0, 0, 0]
produce a number
[7, 9, 0, 0, 0]
consume a number
[0, 9, 0, 0, 0]
consume a number
[0, 0, 0, 0, 0]
produce a number
[9, 0, 0, 0, 0]
consume a number
[0, 0, 0, 0, 0]
produce a number
[9, 0, 0, 0, 0]
produce a number
[9, 5, 0, 0, 0]
produce a number
[9, 5, 6, 0, 0]
produce a number
[9, 5, 6, 10, 0]
consume a number
[0, 5, 6, 10, 0]
produce a number
[7, 5, 6, 10, 0]
produce a number
[7, 5, 6, 10, 2]
all DOne at: Wed Jan 24 22:25:05 2018

其他未提及的對象及方法可參考python標准庫