python 執行緒與程序簡介

程序與執行緒的歷史

我們都知道計算機是由硬體和軟體組成的。硬體中的CPU是計算機的核心，它承擔計算機的所有任務。 作業系統是執行在硬體之上的軟體，是計算機的管理者，它負責資源的管理和分配、任務的排程。 程式是執行在系統上的具有某種功能的軟體，比如說瀏覽器，音樂播放器等。 每次執行程式的時候，都會完成一定的功能，比如說瀏覽器幫我們開啟網頁，為了保證其獨立性，就需要一個專門的管理和控制執行程式的資料結構——程序控制塊。程序就是一個程式在一個數據集上的一次動態執行過程。程序一般由程式、資料集、程序控制塊三部分組成。我們編寫的程式用來描述程序要完成哪些功能以及如何完成；資料集則是程式在執行過程中所需要使用的資源；程序控制塊用來記錄程序的外部特徵，描述程序的執行變化過程，系統可以利用它來控制和管理程序，它是系統感知程序存在的唯一標誌。

在早期的作業系統裡，計算機只有一個核心，程序執行程式的最小單位，任務排程採用時間片輪轉的搶佔式方式進行程序排程。每個程序都有各自的一塊獨立的記憶體，保證程序彼此間的記憶體地址空間的隔離。 隨著計算機技術的發展，程序出現了很多弊端，一是程序的建立、撤銷和切換的開銷比較大，二是由於對稱多處理機（對稱多處理機（SymmetricalMulti-Processing）又叫SMP，是指在一個計算機上彙集了一組處理器(多CPU)，各CPU之間共享記憶體子系統以及匯流排結構）的出現，可以滿足多個執行單位，而多程序並行開銷過大。 這個時候就引入了執行緒的概念。 執行緒也叫輕量級程序，它是一個基本的CPU執行單元，也是程式執行過程中的最小單元，由執行緒ID、程式計數器、暫存器集合 和堆疊共同組成。執行緒的引入減小了程式併發執行時的開銷，提高了作業系統的併發效能。 執行緒沒有自己的系統資源，只擁有在執行時必不可少的資源。但執行緒可以與同屬與同一程序的其他執行緒共享程序所擁有的其他資源。

程序與執行緒之間的關係

執行緒是屬於程序的，執行緒執行在程序空間內，同一程序所產生的執行緒共享同一記憶體空間，當程序退出時該程序所產生的執行緒都會被強制退出並清除。執行緒可與屬於同一程序的其它執行緒共享程序所擁有的全部資源，但是其本身基本上不擁有系統資源，只擁有一點在執行中必不可少的資訊(如程式計數器、一組暫存器和棧)。

python 執行緒

Threading用於提供執行緒相關的操作，執行緒是應用程式中工作的最小單元。

1、threading模組

threading 模組建立在 _thread 模組之上。thread 模組以低階、原始的方式來處理和控制執行緒，而 threading 模組通過對 thread 進行二次封裝，提供了更方便的 api 來處理執行緒。

import threading
import time
 
def worker(num):
    """
    thread worker function
    :return:
    """
    time.sleep(1)
    print("The num is  %d" % num)
    return
 
for i in range(20):
    t = threading.Thread(target=worker,args=(i,)，name=“t.%d” % i)
    t.start()

上述程式碼建立了20個“前臺”執行緒，然後控制器就交給了CPU，CPU根據指定演算法進行排程，分片執行指令。

Thread方法說明

t.start() : 啟用執行緒，

t.getName() : 獲取執行緒的名稱

t.setName() ： 設定執行緒的名稱

t.name: 獲取或設定執行緒的名稱

t.is_alive() ： 判斷執行緒是否為啟用狀態

t.isAlive() ：判斷執行緒是否為啟用狀態

t.setDaemon() 設定為後臺執行緒或前臺執行緒（預設：False）;通過一個布林值設定執行緒是否為守護執行緒，必須在執行start()方法之後才可以使用。如果是後臺執行緒，主執行緒執行過程中，後臺執行緒也在進行，主執行緒執行完畢後，後臺執行緒不論成功與否，均停止；如果是前臺執行緒，主執行緒執行過程中，前臺執行緒也在進行，主執行緒執行完畢後，等待前臺執行緒也執行完成後，程式停止

t.isDaemon() ： 判斷是否為守護執行緒

t.ident ：獲取執行緒的識別符號。執行緒識別符號是一個非零整數，只有在呼叫了start()方法之後該屬性才有效，否則它只返回None。

t.join() ：逐個執行每個執行緒，執行完畢後繼續往下執行，該方法使得多執行緒變得無意義

t.run() ：執行緒被cpu排程後自動執行執行緒物件的run方法

2、執行緒鎖threading.RLock和threading.Lock

由於執行緒之間是進行隨機排程，並且每個執行緒可能只執行n條執行之後，CPU接著執行其他執行緒。為了保證資料的準確性，引入了鎖的概念。所以，可能出現如下問題：

例：假設列表A的所有元素就為0，當一個執行緒從前向後列印列表的所有元素，另外一個執行緒則從後向前修改列表的元素為1,那麼輸出的時候，列表的元素就會一部分為0，一部分為1,這就導致了資料的不一致。鎖的出現解決了這個問題。

import threading
import time
 
globals_num = 0
 
lock = threading.RLock()
 
def Func():
    lock.acquire()  # 獲得鎖 
    global globals_num
    globals_num += 1
    time.sleep(1)
    print(globals_num)
    lock.release()  # 釋放鎖 
 
for i in range(10):
    t = threading.Thread(target=Func)
    t.start()

3、threading.RLock和threading.Lock 的區別

RLock允許在同一執行緒中被多次acquire。而Lock卻不允許這種情況。 如果使用RLock，那麼acquire和release必須成對出現，即呼叫了n次acquire，必須呼叫n次的release才能真正釋放所佔用的瑣。

import threading
lock = threading.Lock()    #Lock物件
lock.acquire()
lock.acquire()  #產生了死瑣。
lock.release()
lock.release()　

import threading
rLock = threading.RLock()  #RLock物件
rLock.acquire()
rLock.acquire()    #在同一執行緒內，程式不會堵塞。
rLock.release()
rLock.release()

4、threading.Event

python執行緒的事件用於主執行緒控制其他執行緒的執行，事件主要提供了三個方法set、wait、clear。

事件處理的機制：全域性定義了一個“Flag”，如果“Flag”值為 False，那麼當程式執行 event.wait 方法時就會阻塞，如果“Flag”值為True，那麼event.wait 方法時便不再阻塞。

• clear：將“Flag”設定為False
• set：將“Flag”設定為True
• Event.isSet() ：判斷標識位是否為Ture。

import threading
 
def do(event):
    print('start')
    event.wait()
    print('execute')
 
event_obj = threading.Event()
for i in range(10):
    t = threading.Thread(target=do, args=(event_obj,))
    t.start()
 
event_obj.clear()
inp = input('input:')
if inp == 'true':
    event_obj.set()

當執行緒執行的時候，如果flag為False，則執行緒會阻塞，當flag為True的時候，執行緒不會阻塞。它提供了本地和遠端的併發性。

5、threading.Condition

一個condition變數總是與某些型別的鎖相聯絡，這個可以使用預設的情況或建立一個，當幾個condition變數必須共享和同一個鎖的時候，是很有用的。鎖是conditon物件的一部分：沒有必要分別跟蹤。

condition變數服從上下文管理協議：with語句塊封閉之前可以獲取與鎖的聯絡。 acquire() 和 release() 會呼叫與鎖相關聯的相應的方法。

其他和鎖關聯的方法必須被呼叫，wait()方法會釋放鎖，當另外一個執行緒使用 notify() or notify_all()喚醒它之前會一直阻塞。一旦被喚醒，wait()會重新獲得鎖並返回，

Condition類實現了一個conditon變數。 這個conditiaon變數允許一個或多個執行緒等待，直到他們被另一個執行緒通知。 如果lock引數，被給定一個非空的值，，那麼他必須是一個lock或者Rlock物件，它用來做底層鎖。否則，會建立一個新的Rlock物件，用來做底層鎖。

• wait(timeout=None) ： 等待通知，或者等到設定的超時時間。當呼叫這wait()方法時，如果呼叫它的執行緒沒有得到鎖，那麼會丟擲一個RuntimeError 異常。 wati()釋放鎖以後，在被呼叫相同條件的另一個程序用notify() or notify_all() 叫醒之前 會一直阻塞。wait() 還可以指定一個超時時間。

如果有等待的執行緒，notify()方法會喚醒一個在等待conditon變數的執行緒。notify_all() 則會喚醒所有在等待conditon變數的執行緒。

注意： notify()和notify_all()不會釋放鎖，也就是說，執行緒被喚醒後不會立刻返回他們的wait() 呼叫。除非執行緒呼叫notify()和notify_all()之後放棄了鎖的所有權。

在典型的設計風格里，利用condition變數用鎖去通許訪問一些共享狀態，執行緒在獲取到它想得到的狀態前，會反覆呼叫wait()。修改狀態的執行緒在他們狀態改變時呼叫 notify() or notify_all()，用這種方式，執行緒會盡可能的獲取到想要的一個等待者狀態。 例子： 生產者-消費者模型，

import threading
import time
def consumer(cond):
    with cond:
        print("consumer before wait")
        cond.wait()
        print("consumer after wait")
 
def producer(cond):
    with cond:
        print("producer before notifyAll")
        cond.notifyAll()
        print("producer after notifyAll")
 
condition = threading.Condition()
c1 = threading.Thread(name="c1", target=consumer, args=(condition,))
c2 = threading.Thread(name="c2", target=consumer, args=(condition,))
 
p = threading.Thread(name="p", target=producer, args=(condition,))
 
c1.start()
time.sleep(2)
c2.start()
time.sleep(2)
p.start()

6、queue模組

Queue 就是對佇列，它是執行緒安全的

舉例來說，我們去麥當勞吃飯。飯店裡面有廚師職位，前臺負責把廚房做好的飯賣給顧客，顧客則去前臺領取做好的飯。這裡的前臺就相當於我們的佇列。形成管道樣，廚師做好飯通過前臺傳送給顧客，所謂單向佇列

這個模型也叫生產者-消費者模型。

import queue

q = queue.Queue(maxsize=0)  # 構造一個先進顯出佇列，maxsize指定佇列長度，為0 時，表示佇列長度無限制。

q.join()    # 等到佇列為kong的時候，在執行別的操作
q.qsize()   # 返回佇列的大小 （不可靠）
q.empty()   # 當佇列為空的時候，返回True 否則返回False （不可靠）
q.full()    # 當佇列滿的時候，返回True，否則返回False （不可靠）
q.put(item, block=True, timeout=None) #  將item放入Queue尾部，item必須存在，可以引數block預設為True,表示當佇列滿時，會等待佇列給出可用位置，
　　　　　　　　　　　　　　　　　　　　　　　　 為False時為非阻塞，此時如果佇列已滿，會引發queue.Full 異常。 可選引數timeout，表示 會阻塞設定的時間，過後，
　　　　　　　　　　　　　　　　　　　　　　　　  如果佇列無法給出放入item的位置，則引發 queue.Full 異常
q.get(block=True, timeout=None) #   移除並返回佇列頭部的一個值，可選引數block預設為True，表示獲取值的時候，如果佇列為空，則阻塞，為False時，不阻塞，
　　　　　　　　　　　　　　　　　　　　　　若此時佇列為空，則引發 queue.Empty異常。 可選引數timeout，表示會阻塞設定的時候，過後，如果佇列為空，則引發Empty異常。
q.put_nowait(item) #   等效於 put(item,block=False)
q.get_nowait() #    等效於 get(item,block=False)

程式碼如下：

#!/usr/bin/env python
import Queue
import threading


message = Queue.Queue(10)


def producer(i):
    while True:
        message.put(i)


def consumer(i):
    while True:
        msg = message.get()


for i in range(12):
    t = threading.Thread(target=producer, args=(i,))
    t.start()

for i in range(10):
    t = threading.Thread(target=consumer, args=(i,))
    t.start()

那就自己做個執行緒池吧：

python 程序

multiprocessing是python的多程序管理包，和threading.Thread類似。

1、multiprocessing模組

直接從側面用subprocesses替換執行緒使用GIL的方式，由於這一點，multiprocessing模組可以讓程式設計師在給定的機器上充分的利用CPU。在multiprocessing中，通過建立Process物件生成程序，然後呼叫它的start()方法，

from multiprocessing import Process

def func(name):
    print('hello', name)


if __name__ == "__main__":
    p = Process(target=func,args=('zhangyanlin',))
    p.start()
    p.join()  # 等待程序執行完畢

在使用併發設計的時候最好儘可能的避免共享資料，尤其是在使用多程序的時候。 如果你真有需要 要共享資料， multiprocessing提供了兩種方式。

（1）multiprocessing，Array,Value

資料可以用Value或Array儲存在一個共享記憶體地圖裡，如下：

from multiprocessing import Array,Value,Process

def func(a,b):
    a.value = 3.333333333333333
    for i in range(len(b)):
        b[i] = -b[i]


if __name__ == "__main__":
    num = Value('d',0.0)
    arr = Array('i',range(11))


    c = Process(target=func,args=(num,arr))
    d= Process(target=func,args=(num,arr))
    c.start()
    d.start()
    c.join()
    d.join()

    print(num.value)
    for i in arr:
        print(i)
輸出：
　　3.1415927
　　[0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

建立num和arr時，“d”和“i”引數由Array模組使用的typecodes建立：“d”表示一個雙精度的浮點數，“i”表示一個有符號的整數，這些共享物件將被執行緒安全的處理。

Array(‘i’, range(10))中的‘i’引數：

‘c’: ctypes.c_char　　　　 ‘u’: ctypes.c_wchar　　　　‘b’: ctypes.c_byte　　　　 ‘B’: ctypes.c_ubyte
‘h’: ctypes.c_short　　　  ‘H’: ctypes.c_ushort　　  ‘i’: ctypes.c_int　　　　　 ‘I’: ctypes.c_uint
‘l’: ctypes.c_long,　　　　‘L’: ctypes.c_ulong　　　　‘f’: ctypes.c_float　　　　‘d’: ctypes.c_double

（2）multiprocessing，Manager

由Manager()返回的manager提供list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array型別的支援。

from multiprocessing import Process,Manager
def f(d,l):
    d["name"] = "zhangyanlin"
    d["age"] = 18
    d["Job"] = "pythoner"
    l.reverse()

if __name__ == "__main__":
    with Manager() as man:
        d = man.dict()
        l = man.list(range(10))

        p = Process(target=f,args=(d,l))
        p.start()
        p.join()

        print(d)
        print(l)

輸出：
　　{0.25: None, 1: '1', '2': 2}
　　[9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

Server process manager比 shared memory 更靈活，因為它可以支援任意的物件型別。另外，一個單獨的manager可以通過程序在網路上不同的計算機之間共享，不過他比shared memory要慢。

2、程序池（Using a pool of workers）

Pool類描述了一個工作程序池，他有幾種不同的方法讓任務解除安裝工作程序。

程序池內部維護一個程序序列，當使用時，則去程序池中獲取一個程序，如果程序池序列中沒有可供使用的進程序，那麼程式就會等待，直到程序池中有可用程序為止。

我們可以用Pool類建立一個程序池， 展開提交的任務給程序池。 例：

#apply
from  multiprocessing import Pool
import time

def f1(i):
    time.sleep(0.5)
    print(i)
    return i + 100

if __name__ == "__main__":
    pool = Pool(5)
    for i in range(1,31):
        pool.apply(func=f1,args=(i,))

#apply_async
def f1(i):
    time.sleep(0.5)
    print(i)
    return i + 100
def f2(arg):
    print(arg)

if __name__ == "__main__":
    pool = Pool(5)
    for i in range(1,31):
        pool.apply_async(func=f1,args=(i,),callback=f2)
    pool.close()
    pool.join()

一個程序池物件可以控制工作程序池的哪些工作可以被提交，它支援超時和回撥的非同步結果，有一個類似map的實現。

• processes ：使用的工作程序的數量，如果processes是None那麼使用 os.cpu_count()返回的數量。
• initializer： 如果initializer是None，那麼每一個工作程序在開始的時候會呼叫initializer(*initargs)。
• maxtasksperchild：工作程序退出之前可以完成的任務數，完成後用一個心的工作程序來替代原程序，來讓閒置的資源被釋放。maxtasksperchild預設是None，意味著只要Pool存在工作程序就會一直存活。
• context: 用在制定工作程序啟動時的上下文，一般使用 multiprocessing.Pool() 或者一個context物件的Pool()方法來建立一個池，兩種方法都適當的設定了context

注意：Pool物件的方法只可以被建立pool的程序所呼叫。

New in version 3.2: maxtasksperchild

New in version 3.4: context

程序池的方法

• apply(func[, args[, kwds]]) ：使用arg和kwds引數呼叫func函式，結果返回前會一直阻塞，由於這個原因，apply_async()更適合併發執行，另外，func函式僅被pool中的一個程序執行。

• apply_async(func[, args[, kwds[, callback[, error_callback]]]]) ： apply()方法的一個變體，會返回一個結果物件。如果callback被指定，那麼callback可以接收一個引數然後被呼叫，當結果準備好回撥時會呼叫callback，呼叫失敗時，則用error_callback替換callback。 Callbacks應被立即完成，否則處理結果的執行緒會被阻塞。

• close() ： 阻止更多的任務提交到pool，待任務完成後，工作程序會退出。

• terminate() ： 不管任務是否完成，立即停止工作程序。在對pool物件程序垃圾回收的時候，會立即呼叫terminate()。

• join() : wait工作執行緒的退出，在呼叫join()前，必須呼叫close() or terminate()。這樣是因為被終止的程序需要被父程序呼叫wait（join等價與wait），否則程序會成為殭屍程序。

• map(func, iterable[, chunksize])¶

• map_async(func, iterable[, chunksize[, callback[, error_callback]]])¶

• imap(func, iterable[, chunksize])¶

• imap_unordered(func, iterable[, chunksize])

• starmap(func, iterable[, chunksize])¶

• starmap_async(func, iterable[, chunksize[, callback[, error_back]]])

python 協程

執行緒和程序的操作是由程式觸發系統介面，最後的執行者是系統；協程的操作則是程式設計師。

協程存在的意義：對於多執行緒應用，CPU通過切片的方式來切換執行緒間的執行，執行緒切換時需要耗時（儲存狀態，下次繼續）。協程，則只使用一個執行緒，在一個執行緒中規定某個程式碼塊執行順序。

協程的適用場景：當程式中存在大量不需要CPU的操作時（IO），適用於協程；

event loop是協程執行的控制點， 如果你希望執行協程， 就需要用到它們。

event loop提供瞭如下的特性：

• 註冊、執行、取消延時呼叫(非同步函式)
• 建立用於通訊的client和server協議(工具)
• 建立和別的程式通訊的子程序和協議(工具)
• 把函式呼叫送入執行緒池中

協程示例：

import asyncio
 
async def cor1():
    print("COR1 start")
    await cor2()
    print("COR1 end")
 
async def cor2():
    print("COR2")
 
loop = asyncio.get_event_loop()
loop.run_until_complete(cor1())
loop.close()

最後三行是重點。

• asyncio.get_event_loop() :asyncio啟動預設的event loop
• run_until_complete() : 這個函式是阻塞執行的，知道所有的非同步函式執行完成，
• close() : 關閉event loop。

1、greenlet

import greenlet


def fun1():
    print("12")
    gr2.switch()
    print("56")
    gr2.switch()

def fun2():
    print("34")
    gr1.switch()
    print("78")


gr1 = greenlet.greenlet(fun1)
gr2 = greenlet.greenlet(fun2)
gr1.switch()

2、gevent

gevent屬於第三方模組需要下載安裝包

• pip3 install --upgrade pip3

• pip3installgevent

import gevent

def fun1():
    print("www.baidu.com")   # 第一步
    gevent.sleep(0)
    print("end the baidu.com")  # 第三步

def fun2():
    print("www.zhihu.com")   # 第二步
    gevent.sleep(0)
    print("end th zhihu.com")  # 第四步

gevent.joinall([
    gevent.spawn(fun1),
    gevent.spawn(fun2),
])

遇到IO操作自動切換：