python多線程與線程

進程與線程的概念

進程

考慮一個場景：瀏覽器，網易雲音樂以及notepad++ 三個軟件只能順序執行是怎樣一種場景呢？另外，假若有兩個程序A和B，程序A在執行到一半的過程當中，須要讀取大量的數據輸入（I/O操做），而此時CPU只能靜靜地等待任務A讀取完數據才能繼續執行，這樣就白白浪費了CPU資源。你是否是已經想到在程序A讀取數據的過程當中，讓程序B去執行，當程序A讀取完數據以後，讓程序B暫停。聰明，這固然沒問題，但這裏有一個關鍵詞：切換。

既然是切換，那麼這就涉及到了狀態的保存，狀態的恢復，加上程序A與程序B所須要的系統資源（內存，硬盤，鍵盤等等）是不同的。天然而然的就須要有一個東西去記錄程序A和程序B分別須要什麼資源，怎樣去識別程序A和程序B等等(好比讀書)。

進程定義：

進程就是一個程序在一個數據集上的一次動態執行過程。進程通常由程序、數據集、進程控制塊三部分組成。咱們編寫的程序用來描述進程要完成哪些功能以及如何完成；數據集則是程序在執行過程當中所須要使用的資源；進程控制塊用來記錄進程的外部特徵，描述進程的執行變化過程，系統能夠利用它來控制和管理進程，它是系統感知進程存在的惟一標誌。

舉一例說明進程：
想象一位有一手好廚藝的計算機科學家正在爲他的女兒烘製生日蛋糕。他有作生日蛋糕的食譜，廚房裏有所需的原料:麪粉、雞蛋、糖、香草汁等。在這個比喻中，作蛋糕的食譜就是程序(即用適當形式描述的算法)計算機科學家就是處理器(cpu)，而作蛋糕的各類原料就是輸入數據。進程就是廚師閱讀食譜、取來各類原料以及烘製蛋糕等一系列動做的總和。如今假設計算機科學家的兒子哭着跑了進來，說他的頭被一隻蜜蜂蟄了。計算機科學家就記錄下他照着食譜作到哪兒了(保存進程的當前狀態)，而後拿出一本急救手冊，按照其中的指示處理蟄傷。這裏，咱們看處處理機從一個進程(作蛋糕)切換到另外一個高優先級的進程(實施醫療救治)，每一個進程擁有各自的程序(食譜和急救手冊)。當蜜蜂蟄傷處理完以後，這位計算機科學家又回來作蛋糕，從他
離開時的那一步繼續作下去。

線程

線程的出現是爲了下降上下文切換的消耗，提升系統的併發性，並突破一個進程只能幹同樣事的缺陷，使到進程內併發成爲可能。

假設，一個文本程序，須要接受鍵盤輸入，將內容顯示在屏幕上，還須要保存信息到硬盤中。若只有一個進程，勢必形成同一時間只能幹同樣事的尷尬（當保存時，就不能經過鍵盤輸入內容）。如有多個進程，每一個進程負責一個任務，進程A負責接收鍵盤輸入的任務，進程B負責將內容顯示在屏幕上的任務，進程C負責保存內容到硬盤中的任務。這裏進程A，B，C間的協做涉及到了進程通訊問題，並且有共同都須要擁有的東西——-文本內容，不停的切換形成性能上的損失。如有一種機制，可使任務A，B，C共享資源，這樣上下文切換所須要保存和恢復的內容就少了，同時又能夠減小通訊所帶來的性能損耗，那就行了。是的，這種機制就是線程。
線程也叫輕量級進程，它是一個基本的CPU執行單元，也是程序執行過程當中的最小單元，由線程ID、程序計數器、寄存器集合和堆棧共同組成。線程的引入減少了程序併發執行時的開銷，提升了操做系統的併發性能。線程沒有本身的系統資源。

進程與線程的關係

進程是計算機中的程序關於某數據集合上的一次運行活動，是系統進行資源分配和調度的基本單位，是操做系統結構的基礎。或者說進程是具備必定獨立功能的程序關於某個數據集合上的一次運行活動,進程是系統進行資源分配和調度的一個獨立單位。
線程則是進程的一個實體,是CPU調度和分派的基本單位,它是比進程更小的能獨立運行的基本單位。

 

進程和線程的關係：

(1)一個線程只能屬於一個進程，而一個進程能夠有多個線程，但至少有一個線程。
(2)資源分配給進程，同一進程的全部線程共享該進程的全部資源。
(3)CPU分給線程，即真正在CPU上運行的是線程。

 

並行和併發

並行處理（Parallel Processing）是計算機系統中能同時執行兩個或更多個處理的一種計算方法。並行處理可同時工做於同一程序的不一樣方面。並行處理的主要目的是節省大型和複雜問題的解決時間。併發處理(concurrency Processing)：指一個時間段中有幾個程序都處於已啓動運行到運行完畢之間，且這幾個程序都是在同一個處理機(CPU)上運行，但任一個時刻點上只有一個程序在處理機(CPU)上運行

併發的關鍵是你有處理多個任務的能力，不必定要同時。並行的關鍵是你有同時處理多個任務的能力。因此說，並行是併發的子集

同步與異步

在計算機領域，同步就是指一個進程在執行某個請求的時候，若該請求須要一段時間才能返回信息，那麼這個進程將會一直等待下去，直到收到返回信息才繼續執行下去；異步是指進程不須要一直等下去，而是繼續執行下面的操做，無論其餘進程的狀態。當有消息返回時系統會通知進程進行處理，這樣能夠提升執行的效率。舉個例子，打電話時就是同步通訊，發短息時就是異步通訊。

 

threading模塊

線程對象的建立

Thread類直接建立

import threading
import time

def countNum(n): # 定義某個線程要運行的函數

    print("running on number:%s" %n)

    time.sleep(3)

if __name__ == '__main__':

    t1 = threading.Thread(target=countNum,args=(23,)) #生成一個線程實例
    t2 = threading.Thread(target=countNum,args=(34,))

    t1.start() #啓動線程
    t2.start()

    print("ending!")
複製代碼

Thread類繼承式建立

 

 

#繼承Thread式建立

import threading
import time

class MyThread(threading.Thread):

    def __init__(self,num):
        threading.Thread.__init__(self)
        self.num=num

    def run(self):
        print("running on number:%s" %self.num)
        time.sleep(3)

t1=MyThread(56)
t2=MyThread(78)

t1.start()
t2.start()
print("ending")

Thread類的實例方法

join()和setDaemon()

 join()：在子線程完成運行以前，這個子線程的父線程將一直被阻塞。

# setDaemon(True)：
        '''
         將線程聲明爲守護線程，必須在start() 方法調用以前設置，若是不設置爲守護線程程序會被無限掛起。

         當咱們在程序運行中，執行一個主線程，若是主線程又建立一個子線程，主線程和子線程 就分兵兩路，分別運行，那麼當主線程完成

         想退出時，會檢驗子線程是否完成。若是子線程未完成，則主線程會等待子線程完成後再退出。可是有時候咱們須要的是隻要主線程

         完成了，無論子線程是否完成，都要和主線程一塊兒退出，這時就能夠 用setDaemon方法啦'''


import threading
from time import ctime,sleep
import time

def Music(name):

        print ("Begin listening to {name}. {time}".format(name=name,time=ctime()))
        sleep(3)
        print("end listening {time}".format(time=ctime()))

def Blog(title):

        print ("Begin recording the {title}. {time}".format(title=title,time=ctime()))
        sleep(5)
        print('end recording {time}'.format(time=ctime()))


threads = []


t1 = threading.Thread(target=Music,args=('FILL ME',))
t2 = threading.Thread(target=Blog,args=('',))

threads.append(t1)
threads.append(t2)

if __name__ == '__main__':

    #t2.setDaemon(True)

    for t in threads:

        #t.setDaemon(True) #注意:必定在start以前設置
        t.start()

        #t.join()

    #t1.join()
    #t2.join()    #  考慮這三種join位置下的結果？

    print ("all over %s" %ctime())

其它方法

 

Thread實例對象的方法
  # isAlive(): 返回線程是否活動的。
  # getName(): 返回線程名。
  # setName(): 設置線程名。

threading模塊提供的一些方法：
  # threading.currentThread(): 返回當前的線程變量。
  # threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啓動後、結束前，不包括啓動前和終止後的線程。
  # threading.activeCount(): 返回正在運行的線程數量，與len(threading.enumerate())有相同的結果。

GIL(全局解釋器鎖)

 

'''

定義：
In CPython, the global interpreter lock, or GIL, is a mutex that prevents multiple 
native threads from executing Python bytecodes at once. This lock is necessary mainly 
because CPython’s memory management is not thread-safe. (However, since the GIL 
exists, other features have grown to depend on the guarantees that it enforces.)

'''

Python中的線程是操做系統的原生線程，Python虛擬機使用一個全局解釋器鎖（Global Interpreter Lock）來互斥線程對Python虛擬機的使用。爲了支持多線程機制，一個基本的要求就是須要實現不一樣線程對共享資源訪問的互斥，因此引入了GIL。
GIL：在一個線程擁有了解釋器的訪問權以後，其餘的全部線程都必須等待它釋放解釋器的訪問權，即便這些線程的下一條指令並不會互相影響。
在調用任何Python C API以前，要先得到GIL
GIL缺點：多處理器退化爲單處理器；優勢：避免大量的加鎖解鎖操做

GIL的早期設計

Python支持多線程，而解決多線程之間數據完整性和狀態同步的最簡單方法天然就是加鎖。 因而有了GIL這把超級大鎖，而當愈來愈多的代碼庫開發者接受了這種設定後，他們開始大量依賴這種特性（即默認python內部對象是thread-safe的，無需在實現時考慮額外的內存鎖和同步操做）。慢慢的這種實現方式被發現是蛋疼且低效的。但當你們試圖去拆分和去除GIL的時候，發現大量庫代碼開發者已經重度依賴GIL而很是難以去除了。有多難？作個類比，像MySQL這樣的「小項目」爲了把Buffer Pool Mutex這把大鎖拆分紅各個小鎖也花了從5.5到5.6再到5.7多個大版爲期近5年的時間，而且仍在繼續。MySQL這個背後有公司支持且有固定開發團隊的產品走的如此艱難，那又更況且Python這樣核心開發和代碼貢獻者高度社區化的團隊呢？

GIL的影響

不管你啓多少個線程，你有多少個cpu, Python在執行一個進程的時候會淡定的在同一時刻只容許一個線程運行。
因此，python是沒法利用多核CPU實現多線程的。
這樣，python對於計算密集型的任務開多線程的效率甚至不如串行(沒有大量切換)，可是，對於IO密集型的任務效率仍是有顯著提高的。

 

計算密集型：

 

#coding:utf8
from threading import Thread
import time

def counter():
    i = 0
    for _ in range(50000000):
        i = i + 1

    return True


def main():

    l=[]
    start_time = time.time()

    for i in range(2):

        t = Thread(target=counter)
        t.start()
        l.append(t)
        t.join()

    # for t in l:
    #     t.join()

    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))

if __name__ == '__main__':
    main()


'''
py2.7:
     串行:25.4523348808s
     併發:31.4084379673s
py3.5:
     串行:8.62115597724914s
     併發:8.99609899520874s

'''

解決方案

用multiprocessing替代Thread multiprocessing庫的出現很大程度上是爲了彌補thread庫由於GIL而低效的缺陷。它完整的複製了一套thread所提供的接口方便遷移。惟一的不一樣就是它使用了多進程而不是多線程。每一個進程有本身的獨立的GIL，所以也不會出現進程之間的GIL爭搶。

#coding:utf8
from multiprocessing import Process
import time

def counter():
    i = 0
    for _ in range(40000000):
        i = i + 1

    return True

def main():

    l=[]
    start_time = time.time()

    for _ in range(2):
        t=Process(target=counter)
        t.start()
        l.append(t)
        #t.join()

    for t in l:
       t.join()

    end_time = time.time()
    print("Total time: {}".format(end_time - start_time))

if __name__ == '__main__':
    main()


'''

py2.7:
     串行:6.1565990448 s
     並行:3.1639978885 s

py3.5:
     串行:6.556925058364868 s
     併發:3.5378448963165283 s

'''

固然multiprocessing也不是萬能良藥。它的引入會增長程序實現時線程間數據通信和同步的困難。就拿計數器來舉例子，若是咱們要多個線程累加同一個變量，對於thread來講，申明一個global變量，用thread.Lock的context包裹住三行就搞定了。而multiprocessing因爲進程之間沒法看到對方的數據，只能經過在主線程申明一個Queue，put再get或者用share memory的方法。這個額外的實現成本使得原本就很是痛苦的多線程程序編碼，變得更加痛苦了。

總結：由於GIL的存在，只有IO Bound場景下得多線程會獲得較好的性能 - 若是對並行計算性能較高的程序能夠考慮把核心部分也成C模塊，或者索性用其餘語言實現 - GIL在較長一段時間內將會繼續存在，可是會不斷對其進行改進。

因此對於GIL，既然不能反抗，那就學會去享受它吧！

同步鎖 (Lock)

 

import time
import threading

def addNum():
    global num #在每一個線程中都獲取這個全局變量
    #num-=1

    temp=num
    time.sleep(0.1)
    num =temp-1  # 對此公共變量進行-1操做

num = 100  #設定一個共享變量

thread_list = []

for i in range(100):
    t = threading.Thread(target=addNum)
    t.start()
    thread_list.append(t)

for t in thread_list: #等待全部線程執行完畢
    t.join()

print('Result: ', num)
複製代碼

鎖一般被用來實現對共享資源的同步訪問。爲每個共享資源建立一個Lock對象，當你須要訪問該資源時，調用acquire方法來獲取鎖對象（若是其它線程已經得到了該鎖，則當前線程需等待其被釋放），待資源訪問完後，再調用release方法釋放鎖：

import threading

R=threading.Lock()

R.acquire()
'''
對公共數據的操做
'''
R.release()
複製代碼

死鎖與遞歸鎖

所謂死鎖： 是指兩個或兩個以上的進程或線程在執行過程當中，因爭奪資源而形成的一種互相等待的現象，若無外力做用，它們都將沒法推動下去。此時稱系統處於死鎖狀態或系統產生了死鎖，這些永遠在互相等待的進程稱爲死鎖進程。

import threading
import time

mutexA = threading.Lock()
mutexB = threading.Lock()

class MyThread(threading.Thread):

    def __init__(self):
        threading.Thread.__init__(self)

    def run(self):
        self.fun1()
        self.fun2()

    def fun1(self):

        mutexA.acquire()  # 若是鎖被佔用,則阻塞在這裏,等待鎖的釋放

        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))

        mutexB.acquire()
        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))
        mutexB.release()

        mutexA.release()


    def fun2(self):

        mutexB.acquire()
        print ("I am %s , get res: %s---%s" %(self.name, "ResB",time.time()))
        time.sleep(0.2)

        mutexA.acquire()
        print ("I am %s , get res: %s---%s" %(self.name, "ResA",time.time()))
        mutexA.release()

        mutexB.release()

if __name__ == "__main__":

    print("start---------------------------%s"%time.time())

    for i in range(0, 10):
        my_thread = MyThread()
        my_thread.start()

在Python中爲了支持在同一線程中屢次請求同一資源，python提供了可重入鎖RLock。這個RLock內部維護着一個Lock和一個counter變量，counter記錄了acquire的次數，從而使得資源能夠被屢次require。直到一個線程全部的acquire都被release，其餘的線程才能得到資源。上面的例子若是使用RLock代替Lock，則不會發生死鎖：

1	mutex  =  threading.RLock()

Event對象

線程的一個關鍵特性是每一個線程都是獨立運行且狀態不可預測。若是程序中的其 他線程須要經過判斷某個線程的狀態來肯定本身下一步的操做,這時線程同步問題就 會變得很是棘手。爲了解決這些問題,咱們須要使用threading庫中的Event對象。 對象包含一個可由線程設置的信號標誌,它容許線程等待某些事件的發生。在 初始狀況下,Event對象中的信號標誌被設置爲假。若是有線程等待一個Event對象, 而這個Event對象的標誌爲假,那麼這個線程將會被一直阻塞直至該標誌爲真。一個線程若是將一個Event對象的信號標誌設置爲真,它將喚醒全部等待這個Event對象的線程。若是一個線程等待一個已經被設置爲真的Event對象,那麼它將忽略這個事件, 繼續執行

event.isSet()：返回event的狀態值；

event.wait()：若是 event.isSet()==False將阻塞線程；

event.set()： 設置event的狀態值爲True，全部阻塞池的線程激活進入就緒狀態， 等待操做系統調度；

event.clear()：恢復event的狀態值爲False。

       

能夠考慮一種應用場景（僅僅做爲說明），例如，咱們有多個線程從Redis隊列中讀取數據來處理，這些線程都要嘗試去鏈接Redis的服務，通常狀況下，若是Redis鏈接不成功，在各個線程的代碼中，都會去嘗試從新鏈接。若是咱們想要在啓動時確保Redis服務正常，才讓那些工做線程去鏈接Redis服務器，那麼咱們就能夠採用threading.Event機制來協調各個工做線程的鏈接操做：主線程中會去嘗試鏈接Redis服務，若是正常的話，觸發事件，各工做線程會嘗試鏈接Redis服務。

import threading
import time
import logging

logging.basicConfig(level=logging.DEBUG, format='(%(threadName)-10s) %(message)s',)

def worker(event):
    logging.debug('Waiting for redis ready...')
    event.wait()
    logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
    time.sleep(1)

def main():
    readis_ready = threading.Event()
    t1 = threading.Thread(target=worker, args=(readis_ready,), name='t1')
    t1.start()

    t2 = threading.Thread(target=worker, args=(readis_ready,), name='t2')
    t2.start()

    logging.debug('first of all, check redis server, make sure it is OK, and then trigger the redis ready event')
    time.sleep(3) # simulate the check progress
    readis_ready.set()

if __name__=="__main__":
    main()

threading.Event的wait方法還接受一個超時參數，默認狀況下若是事件一致沒有發生，wait方法會一直阻塞下去，而加入這個超時參數以後，若是阻塞時間超過這個參數設定的值以後，wait方法會返回。對應於上面的應用場景，若是Redis服務器一致沒有啓動，咱們但願子線程可以打印一些日誌來不斷地提醒咱們當前沒有一個能夠鏈接的Redis服務，咱們就能夠經過設置這個超時參數來達成這樣的目的：

def worker(event):
    while not event.is_set():
        logging.debug('Waiting for redis ready...')
        event.wait(2)
    logging.debug('redis ready, and connect to redis server and do some work [%s]', time.ctime())
    time.sleep(1)

這樣，咱們就能夠在等待Redis服務啓動的同時，看到工做線程里正在等待的狀況。

Semaphore（信號量）

Semaphore管理一個內置的計數器，
每當調用acquire()時內置計數器-1；
調用release() 時內置計數器+1；
計數器不能小於0；當計數器爲0時，acquire()將阻塞線程直到其餘線程調用release()。

實例：(同時只有5個線程能夠得到semaphore,便可以限制最大鏈接數爲5)：

import threading
import time

semaphore = threading.Semaphore(5)

def func():
    if semaphore.acquire():
        print (threading.currentThread().getName() + ' get semaphore')
        time.sleep(2)
        semaphore.release()

for i in range(20):
  t1 = threading.Thread(target=func)
  t1.start()

應用：鏈接池

思考：與Rlock的區別？

multiprocessing模塊

Multiprocessing is a package that supports spawning processes using an API similar to the threading module. The multiprocessing package offers both local and remote concurrency,effectively side-stepping the Global Interpreter Lock by using subprocesses instead of threads. Due to this, the multiprocessing module allows the programmer to fully leverage multiple processors on a given machine. It runs on both Unix and Windows.

因爲GIL的存在，python中的多線程其實並非真正的多線程，若是想要充分地使用多核CPU的資源，在python中大部分狀況須要使用多進程。

multiprocessing包是Python中的多進程管理包。與threading.Thread相似，它能夠利用multiprocessing.Process對象來建立一個進程。該進程能夠運行在Python程序內部編寫的函數。該Process對象與Thread對象的用法相同，也有start(), run(), join()的方法。此外multiprocessing包中也有Lock/Event/Semaphore/Condition類 (這些對象能夠像多線程那樣，經過參數傳遞給各個進程)，用以同步進程，其用法與threading包中的同名類一致。因此，multiprocessing的很大一部份與threading使用同一套API，只不過換到了多進程的情境。

python的進程調用

# Process類調用

from multiprocessing import Process
import time
def f(name):

    print('hello', name,time.ctime())
    time.sleep(1)

if __name__ == '__main__':
    p_list=[]
    for i in range(3):
        p = Process(target=f, args=('alvin:%s'%i,))
        p_list.append(p)
        p.start()
    for i in p_list:
        p.join()
    print('end')

# 繼承Process類調用
from multiprocessing import Process
import time

class MyProcess(Process):
    def __init__(self):
        super(MyProcess, self).__init__()
        # self.name = name

    def run(self):

        print ('hello', self.name,time.ctime())
        time.sleep(1)


if __name__ == '__main__':
    p_list=[]
    for i in range(3):
        p = MyProcess()
        p.start()
        p_list.append(p)

    for p in p_list:
        p.join()

    print('end')

process類

構造方法：

Process([group [, target [, name [, args [, kwargs]]]]])

　　group: 線程組，目前尚未實現，庫引用中提示必須是None； 
　　target: 要執行的方法； 
　　name: 進程名； 
　　args/kwargs: 要傳入方法的參數。

實例方法：

　　is_alive()：返回進程是否在運行。

　　join([timeout])：阻塞當前上下文環境的進程程，直到調用此方法的進程終止或到達指定的timeout（可選參數）。

　　start()：進程準備就緒，等待CPU調度

　　run()：strat()調用run方法，若是實例進程時未制定傳入target，這star執行t默認run()方法。

　　terminate()：無論任務是否完成，當即中止工做進程

屬性：

　　daemon：和線程的setDeamon功能同樣

　　name：進程名字。

　　pid：進程號。

from multiprocessing import Process
import os
import time
def info(name):


    print("name:",name)
    print('parent process:', os.getppid())
    print('process id:', os.getpid())
    print("------------------")
    time.sleep(1)

def foo(name):

    info(name)

if __name__ == '__main__':

    info('main process line')


    p1 = Process(target=info, args=('alvin',))
    p2 = Process(target=foo, args=('egon',))
    p1.start()
    p2.start()

    p1.join()
    p2.join()

    print("ending")

經過tasklist(Win)或者ps -elf |grep(linux)命令檢測每個進程號(PID)對應的進程名

協程

yield與協程

import time

"""
傳統的生產者-消費者模型是一個線程寫消息，一個線程取消息，經過鎖機制控制隊列和等待，但一不當心就可能死鎖。
若是改用協程，生產者生產消息後，直接經過yield跳轉到消費者開始執行，待消費者執行完畢後，切換回生產者繼續生產，效率極高。
"""
# 注意到consumer函數是一個generator（生成器）:
# 任何包含yield關鍵字的函數都會自動成爲生成器(generator)對象

def consumer():
    r = ''
    while True:
        # 三、consumer經過yield拿到消息，處理，又經過yield把結果傳回；
        #    yield指令具備return關鍵字的做用。而後函數的堆棧會自動凍結(freeze)在這一行。
        #    當函數調用者的下一次利用next()或generator.send()或for-in來再次調用該函數時，
        #    就會從yield代碼的下一行開始，繼續執行，再返回下一次迭代結果。經過這種方式，迭代器能夠實現無限序列和惰性求值。
        n = yield r
        if not n:
            return
        print('[CONSUMER] ←← Consuming %s...' % n)
        time.sleep(1)
        r = '200 OK'
def produce(c):
    # 一、首先調用c.next()啓動生成器
    next(c)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] →→ Producing %s...' % n)
        # 二、而後，一旦生產了東西，經過c.send(n)切換到consumer執行；
        cr = c.send(n)
        # 四、produce拿到consumer處理的結果，繼續生產下一條消息；
        print('[PRODUCER] Consumer return: %s' % cr)
    # 五、produce決定不生產了，經過c.close()關閉consumer，整個過程結束。
    c.close()
if __name__=='__main__':
    # 六、整個流程無鎖，由一個線程執行，produce和consumer協做完成任務，因此稱爲「協程」，而非線程的搶佔式多任務。
    c = consumer()
    produce(c)
    
    
'''
result:

[PRODUCER] →→ Producing 1...
[CONSUMER] ←← Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 2...
[CONSUMER] ←← Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 3...
[CONSUMER] ←← Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 4...
[CONSUMER] ←← Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] →→ Producing 5...
[CONSUMER] ←← Consuming 5...
[PRODUCER] Consumer return: 200 OK
'''

greentlet

greelet機制的主要思想是：生成器函數或者協程函數中的yield語句掛起函數的執行，直到稍後使用next()或send()操做進行恢復爲止。可使用一個調度器循環在一組生成器函數之間協做多個任務。greentlet是python中實現咱們所謂的"Coroutine(協程)"的一個基礎庫.

from greenlet import greenlet
 
def test1():
    print (12)
    gr2.switch()
    print (34)
    gr2.switch()
 
def test2():
    print (56)
    gr1.switch()
    print (78)
 
gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

基於greenlet的框架

gevent模塊實現協程

Python經過yield提供了對協程的基本支持，可是不徹底。而第三方的gevent爲Python提供了比較完善的協程支持。

gevent是第三方庫，經過greenlet實現協程，其基本思想是：

當一個greenlet遇到IO操做時，好比訪問網絡，就自動切換到其餘的greenlet，等到IO操做完成，再在適當的時候切換回來繼續執行。因爲IO操做很是耗時，常常使程序處於等待狀態，有了gevent爲咱們自動切換協程，就保證總有greenlet在運行，而不是等待IO。

因爲切換是在IO操做時自動完成，因此gevent須要修改Python自帶的一些標準庫，這一過程在啓動時經過monkey patch完成：

import gevent
import time

def foo():
    print("running in foo")
    gevent.sleep(2)
    print("switch to foo again")

def bar():
    print("switch to bar")
    gevent.sleep(5)
    print("switch to bar again")

start=time.time()

gevent.joinall(
    [gevent.spawn(foo),
    gevent.spawn(bar)]
)

print(time.time()-start)

固然，實際代碼裏，咱們不會用gevent.sleep()去切換協程，而是在執行到IO操做時，gevent自動切換，代碼以下：

from gevent import monkey
monkey.patch_all()
import gevent
from urllib import request
import time

def f(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

start=time.time()

gevent.joinall([
        gevent.spawn(f, 'https://itk.org/'),
        gevent.spawn(f, 'https://www.github.com/'),
        gevent.spawn(f, 'https://zhihu.com/'),
])

# f('https://itk.org/')
# f('https://www.github.com/')
# f('https://zhihu.com/')

print(time.time()-start)

擴展：

gevent是一個基於協程（coroutine）的Python網絡函數庫，經過使用greenlet提供了一個在libev事件循環頂部的高級別併發API。

主要特性有如下幾點：

<1> 基於libev的快速事件循環，Linux上面的是epoll機制

<2> 基於greenlet的輕量級執行單元

<3> API複用了Python標準庫裏的內容

<4> 支持SSL的協做式sockets

<5> 可經過線程池或c-ares實現DNS查詢

<6> 經過monkey patching功能來使得第三方模塊變成協做式

gevent.spawn()方法spawn一些jobs，而後經過gevent.joinall將jobs加入到微線程執行隊列中等待其完成，設置超時爲2秒。執行後的結果經過檢查gevent.Greenlet.value值來收集。

  ps

4.2.2 官方文檔中的示例：

import gevent

from gevent import socket

urls = [‘www.google.com.hk’,’www.example.com’, ‘www.python.org’ ]

jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls]

gevent.joinall(jobs, timeout=2)

[job.value for job in jobs]

[‘74.125.128.199’, ‘208.77.188.166’, ‘82.94.164.162’]

註解：gevent.spawn()方法spawn一些jobs，而後經過gevent.joinall將jobs加入到微線程執行隊列中等待其完成，設置超時爲2秒。執行後的結果經過檢查gevent.Greenlet.value值來收集。gevent.socket.gethostbyname()函數與標準的socket.gethotbyname()有相同的接口，但它不會阻塞整個解釋器，所以會使得其餘的greenlets跟隨着無阻的請求而執行。

4.2.3 Monket patching

Python的運行環境容許咱們在運行時修改大部分的對象，包括模塊、類甚至函數。雖然這樣作會產生「隱式的反作用」，並且出現問題很難調試，但在須要修改Python自己的基礎行爲時，Monkey patching就派上用場了。Monkey patching可以使得gevent修改標準庫裏面大部分的阻塞式系統調用，包括socket,ssl,threading和select等模塊，而變成協做式運行。

from gevent import monkey ;

monkey . patch_socket ()

import urllib2

經過monkey.patch_socket()方法，urllib2模塊可使用在多微線程環境，達到與gevent共同工做的目的。

4.2.4 事件循環

不像其餘網絡庫，gevent和eventlet相似， 在一個greenlet中隱式開始事件循環。沒有必須調用run()或dispatch()的反應器(reactor)，在twisted中是有 reactor的。當gevent的API函數想阻塞時，它得到Hub實例(執行時間循環的greenlet),並切換過去。若是沒有集線器實例則會動態 建立。

libev提供的事件循環默認使用系統最快輪詢機制，設置LIBEV_FLAGS環境變量可指定輪詢機制。LIBEV_FLAGS=1爲select， LIBEV_FLAGS = 2爲poll， LIBEV_FLAGS = 4爲epoll,LIBEV_FLAGS = 8爲kqueue。

Libev的API位於gevent.core下。注意libev API的回調在Hub的greenlet運行，所以使用同步greenlet的API。可使用spawn()和Event.set()等異步API。

eventlet實現協程(瞭解)

eventlet 是基於 greenlet 實現的面向網絡應用的併發處理框架，提供「線程」池、隊列等與其餘 Python 線程、進程模型很是類似的 api，而且提供了對 Python 發行版自帶庫及其餘模塊的超輕量併發適應性調整方法，比直接使用 greenlet 要方便得多。

其基本原理是調整 Python 的 socket 調用，當發生阻塞時則切換到其餘 greenlet 執行，這樣來保證資源的有效利用。須要注意的是：eventlet 提供的函數只能對 Python 代碼中的 socket 調用進行處理，而不能對模塊的 C 語言部分的 socket 調用進行修改。對後者這類模塊，仍然須要把調用模塊的代碼封裝在 Python 標準線程調用中，以後利用 eventlet 提供的適配器實現 eventlet 與標準線程之間的協做。雖然 eventlet 把 api 封裝成了很是相似標準線程庫的形式，但二者的實際併發執行流程仍然有明顯區別。在沒有出現 I/O 阻塞時，除非顯式聲明，不然當前正在執行的 eventlet 永遠不會把 cpu 交給其餘的 eventlet，而標準線程則是不管是否出現阻塞，老是由全部線程一塊兒爭奪運行資源。全部 eventlet 對 I/O 阻塞無關的大運算量耗時操做基本沒有什麼幫助。