Python實現協程

什麼是進程和線程

 

 

有必定基礎的小夥伴們確定都知道進程和線程。

 

進程是什麼呢？

 

直白地講，進程就是應用程序的啓動實例。好比咱們運行一個遊戲，打開一個軟件，就是開啓了一個進程。

 

進程擁有代碼和打開的文件資源、數據資源、獨立的內存空間。

 

線程又是什麼呢？

 

線程從屬於進程，是程序的實際執行者。一個進程至少包含一個主線程，也能夠有更多的子線程。

 

線程擁有本身的棧空間。

 

 

 

 

 

有人給出了很好的概括：

 

對操做系統來講，線程是最小的執行單元，進程是最小的資源管理單元。

 

 

不管進程仍是線程，都是由操做系統所管理的。

 

Java中線程具備五種狀態：

 

初始化

可運行

運行中

阻塞

銷燬

 

這五種狀態的轉化關係以下：

 

 

 

 

可是，線程不一樣狀態之間的轉化是誰來實現的呢？是JVM嗎？

 

並非。JVM須要經過操做系統內核中的TCB（Thread Control Block）模塊來改變線程的狀態，這一過程須要耗費必定的CPU資源。

 

 

 

 

 

 

 

進程和線程的痛點

 

線程之間是如何進行協做的呢？

 

最經典的例子就是生產者/消費者模式：

 

若干個生產者線程向隊列中寫入數據，若干個消費者線程從隊列中消費數據。

 

 

 

 

 

如何用java語言實現生產者/消費者模式呢？

 

讓咱們來看一看代碼：

 

 

 

public class ProducerConsumerTest {

}

class Producer extends Thread {

}

class Consumer extends Thread { private final QueuesharedQueue;

}

這段代碼作了下面幾件事：

1.定義了一個生產者類，一個消費者類。

2.生產者類循環100次，向同步隊列當中插入數據。

3.消費者循環監聽同步隊列，當隊列有數據時拉取數據。

4.若是隊列滿了（達到5個元素），生產者阻塞。

5.若是隊列空了，消費者阻塞。

 

上面的代碼正確地實現了生產者/消費者模式，可是卻並非一個高性能的實現。爲何性能不高呢？緣由以下：

 

1.涉及到同步鎖。

2.涉及到線程阻塞狀態和可運行狀態之間的切換。

3.涉及到線程上下文的切換。

 

以上涉及到的任何一點，都是很是耗費性能的操做。

 

 

 

 

 

 

什麼是協程

 

協程，英文Coroutines，是一種比線程更加輕量級的存在。正如一個進程能夠擁有多個線程同樣，一個線程也能夠擁有多個協程。

 

 

 

 

 

最重要的是，協程不是被操做系統內核所管理，而徹底是由程序所控制（也就是在用戶態執行）。

 

這樣帶來的好處就是性能獲得了很大的提高，不會像線程切換那樣消耗資源。

 

既然協程這麼好，它究竟是怎麼來使用的呢？

 

因爲Java的原生語法中並無實現協程（某些開源框架實現了協程，可是不多被使用），因此咱們來看一看python當中對協程的實現案例，一樣以生產者消費者模式爲例：

 

 

 

 

這段代碼十分簡單，即便沒用過python的小夥伴應該也能基本看懂。

 

代碼中建立了一個叫作consumer的協程，而且在主線程中生產數據，協程中消費數據。

 

其中 yield 是python當中的語法。當協程執行到yield關鍵字時，會暫停在那一行，等到主線程調用send方法發送了數據，協程纔會接到數據繼續執行。

 

可是，yield讓協程暫停，和線程的阻塞是有本質區別的。協程的暫停徹底由程序控制，線程的阻塞狀態是由操做系統內核來進行切換。

 

所以，協程的開銷遠遠小於線程的開銷。

 

 

 

 

 

 

 

協程的應用

 

有哪些編程語言應用到了協程呢？咱們舉幾個栗子：

 

Lua語言

Lua從5.0版本開始使用協程，經過擴展庫coroutine來實現。

 

Python語言

正如剛纔所寫的代碼示例，python能夠經過 yield/send 的方式實現協程。在python 3.5之後， async/await 成爲了更好的替代方案。

 

Go語言

Go語言對協程的實現很是強大而簡潔，能夠輕鬆建立成百上千個協程併發執行。

 

Java語言

如上文所說，Java語言並無對協程的原生支持，可是某些開源框架模擬出了協程的功能，有興趣的小夥伴能夠看一看Kilim框架的源碼：

https://github.com/kilim/kilim

 

 

 

 

 

幾點補充：

 

1.關於協程的概念，小灰也僅僅是知道一些皮毛，但願小夥伴們多多指正。

yield與send實現協程操做

以前咱們說過，在函數內部含有yield語句即稱爲生成器。

下面，咱們來看看在函數內部含有yield語句達到的效果。首先，咱們來看看如下代碼：

def foo():
    while True:
        x = yield
        print("value:",x)

g = foo() # g是一個生成器
next(g) # 程序運行到yield就停住了,等待下一個next
g.send(1)  # 咱們給yield發送值1,而後這個值被賦值給了x，而且打印出來,而後繼續下一次循環停在yield處
g.send(2)  # 同上
next(g)  # 沒有給x賦值，執行print語句，打印出None,繼續循環停在yield處

咱們都知道，程序一旦執行到yield就會停在該處,而且將其返回值進行返回。上面的例子中，咱們並無設置返回值，全部默認程序返回的是None。咱們經過打印語句來查看一下第一次next的返回值：

print(next(g))

####輸出結果#####
None

　　正如咱們所說的，程序返回None。接着程序往下執行，可是並無看到next()方法。爲何還會繼續執行yield語句後面的代碼呢？這是由於，send()方法具備兩種功能：第一，傳值，send()方法，將其攜帶的值傳遞給yield，注意，是傳遞給yield，而不是x,而後再將其賦值給x；第二，send()方法具備和next()方法同樣的功能，也就是說，傳值完畢後，會接着上次執行的結果繼續執行，知道遇到yield中止。這也就爲何在調用g.send()方法後，還會打印出x的數值。有了上面的分析，咱們能夠很快知道，執行了send(1)後，函數被中止在了yield處，等待下一個next()的到來。程序往下執行，有遇到了send(2)，其執行流程與send(1)徹底同樣。

　　有了上述的分析，咱們能夠總結出send()的兩個功能：1.傳值；2.next()。

　　既然send()方法有和next同樣的做用，那麼咱們可不能夠這樣作：

 

def foo():
    while True:
        x = yield
        print("value:",x)

g = foo()
g.send(1) #執行給yield傳值,這樣行不行呢?

 

　　很明顯，是不行的。

 

 

TypeError: can't send non-None value to a just-started generator

 

　　錯誤提示:不能傳遞一個非空值給一個未啓動的生成器。

　　也就是說，在一個生成器函數未啓動以前，是不能傳遞數值進去。必須先傳遞一個None進去或者調用一次next(g)方法，才能進行傳值操做。至於爲何要先傳遞一個None進去，能夠看一下官方說法。

Because generator-iterators begin execution at the top of the
    generator's function body, there is no yield expression to receive
    a value when the generator has just been created.  Therefore,
    calling send() with a non-None argument is prohibited when the
    generator iterator has just started, and a TypeError is raised if
    this occurs (presumably due to a logic error of some kind).  Thus,
    before you can communicate with a coroutine you must first call
    next() or send(None) to advance its execution to the first yield
    expression.

　　問題就來，既然在給yield傳值過程當中，會調用next()方法，那麼是否是在調用一次函數的時候，是否是每次都要給它傳遞一個空值？有沒有什麼簡便方法來解決這個問題呢？答案，裝飾器！！看下面代碼:

def deco(func):  # 裝飾器:用來開啓協程
    def wrapper():
        res = func()
        next(res)
        return res  # 返回一個已經執行了next()方法的函數對象
    return wrapper
@deco
def foo():
    pass

　　上面我yield是沒有返回值的，下面咱們看看有返回值的生成器函數。

def deco(func):
    def wrapper():
        res = func()
        next(res)
        return res
    return wrapper
@deco
def foo():
    food_list = []
    while True:
        food = yield food_list  #返回添加food的列表
        food_list.append(food)
        print("elements in foodlist are:",food)
g = foo()
print(g.send('蘋果'))
print(g.send('香蕉'))
print(g.send('菠蘿'))
###########輸出結果爲######
elements in foodlist are: 蘋果
['蘋果']
elements in foodlist are: 香蕉
['蘋果', '香蕉']
elements in foodlist are: 菠蘿
['蘋果', '香蕉', '菠蘿']

　　分析：首先，咱們要清楚，在函數執行以前，已經執行了一次next()(裝飾器的功能)，程序中止yield。接着程序往下執行，遇到g.send()，而後將其值傳遞給food，而後再將得到的food添加到列表food_list中。打印出food，再次循環程序停在yield。程序繼續執行，又遇到g.send()，其過程與上面是如出一轍的。看看如下的程序執行流程，你可能會更清楚。

　　

　　這裏咱們要明確一點，yield的返回值和傳給yield的值是兩碼事！！

　　yiedl的返回值就至關於return的返回值，這個值是要被傳遞出去的,而send()傳遞的值，是要被yield接受，供函數內部使用的的，明確這一點很重要的。那麼上面的打印，就應該打印出yield的返回值，而傳遞進去的值則本保存在一個列表中。

3、實例

"""模擬:grep -rl 'root' /etc"""
import os
def deco(func):  # 用來開啓協程
    def wrapper(*args,**kwargs):
        res = func(*args,**kwargs)
        next(res)  # res.seng(None)
        return res
    return wrapper
@deco
def search(target):
    while True:
        PATH = yield
        g = os.walk(PATH)  # 獲取PATH目錄下的文件，文件夾
        for par_dir, _, files in g:  #迭代解包,取出當前目錄路徑和文件名
            for file in files:
                file_path = r'%s\%s' %(par_dir,file)  # 拼接文件的絕對路徑
                target.send(file_path)  # 給下一個
@deco
def opener(target, pattern=None):
    while True:
        file_path = yield
        with open(file_path, encoding='utf-8') as f:
            target.send((file_path, f))  # 將文件路徑和文件對象一塊兒傳遞給下一個函數的yield，由於在打印路徑時候，須要打印出文件路徑，只有從這裏傳遞下去
@deco
def cat(target):
    while True:
        filepath, f = yield # 這裏接收opener傳遞進來的路徑和文件對象
        for line in f:
            tag = target.send((filepath, line))  # 一樣，也要傳遞文件路徑,而且獲取下一個函數grep的返回值，從而判斷該文件是否重複讀取了
            if tag:  # 若是爲真，說明該文件讀取過了，則執行退出循環
                break
@deco
def grep(target, pattern):
    tag = False
    while True:
        filepath, line = yield tag  # 接受兩個值，而且設置返回值，這個返回值要傳遞給發送消息的send()，也就是cat()函數send
        tag = False
        if pattern in line:  # 若是待匹配字符串在該行
            target.send(filepath)   # 把文件路徑傳遞給priter
            tag = True   # 設置tag
@deco
def printer():
    while True:
        filename = yield
        print(filename)

調用方式:

PATH1 = r'D:\CODE_FILE\python\test'
search(opener(cat(grep(printer(), 'root')))).send(PATH1)

輸出結果:

######找出了含有'root'的全部文件#######
D:\CODE_FILE\python\test\a.txt
D:\CODE_FILE\python\test\test1\c.txt
D:\CODE_FILE\python\test\test1\test2\d.txt

程序分析:

 　　有了上面的基礎，咱們來分析一下上述程序的執行。

　　每個函數以前都有一個@deco裝飾器，這個裝飾器用於開啓協程。首先咱們定義了一個search()，其內部有關鍵字yield，則search()是一個生成器。也就是說，咱們能夠經過send()給其傳遞一個值進去。search()函數的功能 是：獲取一個文件的絕對路徑，並將這個絕對路徑經過send()方法，在傳遞給下個含有yield的函數，也就是下面的opener函數。opener的yield接受了search()傳遞進來的路徑，並將其賦值給了file_path，而後咱們根據這個路徑，打開了一個文件，獲取了一個文件對象f。而後咱們在將這個文件對象send()給cat()函數，這個函數功能是讀取文件中的內容，咱們根據逐行讀取文件內容，將每次讀取到的內容，在send()給下一個函數，也就是grep()，這個函數功能是實現過濾操做。咱們從上一個函數cat()接受到的每一行內容，在grep()函數裏面進行過濾處理，若是有匹配到的過濾內容，那麼咱們將就過濾到的文件傳遞給下一個函數print()，該函數主要是打印出文件路徑。也許，上述的描述內容你沒看懂，下面看看這個程序的流程圖：

　　　　　　　　

　　根據上述流程，咱們很清楚知道，send()傳遞給下一個函數的值。可是上述代碼存在一個問題：若是待過濾的字符在一個文件中存在多個，而在讀取文件的時候，咱們是一行一行地讀取，而後再傳遞給下一個函數。咱們的目的是：過濾出包好pattern的文件，若是一個文件存在多個一樣的pattern，那麼就會輸出屢次一樣的文件名。這無疑是浪費內存，要解決這中問題，咱們能夠經過yield的返回值來控制，每次讀取時候的條件。具體實施，看上述代碼註釋。