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:
        # 3、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):
    # 1、首先調用c.next()啟動生成器
    next(c)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] →→ Producing %s...' % n)
        # 2、然后，一旦生產了東西，通過c.send(n)切換到consumer執行；
        cr = c.send(n)
        # 4、produce拿到consumer處理的結果，繼續生產下一條消息；
        print('[PRODUCER] Consumer return: %s' % cr)
    # 5、produce決定不生產了，通過c.close()關閉consumer，整個過程結束。
    c.close()
if __name__=='__main__':
    # 6、整個流程無鎖，由一個線程執行，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值來收集。

 

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 阻塞無關的大運算量耗時操作基本沒有什么幫助。