python 全棧開發，Day43(引子,協程介紹,Greenlet模塊,Gevent模塊,Gevent之同步與異步)

昨日內容回顧                                                                   

I/O模型，面試會問到
I/O操做，不佔用CPU。它內部有一個專門的處理I/O模塊。
print和寫log 屬於I/O操做，它不佔用CPU

線程
GIL保證一個進程中的多個線程在同一時刻只有一個能夠被CPU執行

後續的項目，特別是處理網絡請求，很是多。

 

實例化一個Lock()，它就是一個互斥鎖

LCOK 和RLOCK
互斥鎖LCOK
死鎖
rlock 遞歸鎖
遞歸鎖不會發生死鎖現象

 

2個進程中的線程，不會受到GIL影響。
GIL是針對一個進程中的多個線程，同一時間，只能有一個線程訪問CPU
針對GIL的CPU利用率問題
起多個進程，就能夠解決CPU利用率問題。

 

昨天的科學家吃麪的例子，它不能用一把鎖，必須2個鎖。

def eat1(noodle_lock,fork_lock,name):
    noodle_lock.acquire()
    print('%s搶到了面'%name)
    fork_lock.acquire()
    print('%s搶到了叉子'%name)
    print('%s正在吃麪'%name)
    fork_lock.release()
    print('%s歸還了叉子' % name)
    noodle_lock.release()
    print('%s歸還了面' % name)

　　

看下圖

假設有三我的，

A要面和叉子

B只要面

C只要叉子

若是隻有一個鎖，那麼就沒法處理這3我的的需求，會發生數據不安全的狀況。

 

semaphore 在一開始固定一個線程的流量
condition 經過一個信號控制線程的流量
event 經過一個信號控制全部線程
timer 定時器
隊列 線程數據安全

線程池
可以在多線程的基礎上進一步節省內存和時間開銷

 

 1、引子                                                                         

以前咱們學習了線程、進程的概念，瞭解了在操做系統中進程是資源分配的最小單位,線程是CPU調度的最小單位。按道理來講咱們已經算是把cpu的利用率提升不少了。可是咱們知道不管是建立多進程仍是建立多線程來解決問題，都要消耗必定的時間來建立進程、建立線程、以及管理他們之間的切換。

　　隨着咱們對於效率的追求不斷提升，基於單線程來實現併發又成爲一個新的課題，即只用一個主線程（很明顯可利用的cpu只有一個）狀況下實現併發。這樣就能夠節省建立線進程所消耗的時間。

　　爲此咱們須要先回顧下併發的本質：切換+保存狀態

 　　cpu正在運行一個任務，會在兩種狀況下切走去執行其餘的任務（切換由操做系統強制控制），一種狀況是該任務發生了阻塞，另一種狀況是該任務計算的時間過長

　  　

　　ps：在介紹進程理論時，說起進程的三種執行狀態，而線程纔是執行單位，因此也能夠將上圖理解爲線程的三種狀態 

　　 一：其中第二種狀況並不能提高效率，只是爲了讓cpu可以雨露均沾，實現看起來全部任務都被「同時」執行的效果，若是多個任務都是純計算的，這種切換反而會下降效率。

　　爲此咱們能夠基於yield來驗證。yield自己就是一種在單線程下能夠保存任務運行狀態的方法，咱們來簡單複習一下：

#1 yiled能夠保存狀態，yield的狀態保存與操做系統的保存線程狀態很像，可是yield是代碼級別控制的，更輕量級
#2 send能夠把一個函數的結果傳給另一個函數，以此實現單線程內程序之間的切換

#串行執行
import time
def consumer(res):
    '''任務1:接收數據,處理數據'''
    pass

def producer():
    '''任務2:生產數據'''
    res=[]
    for i in range(10000000):
        res.append(i)
    return res

start=time.time()
#串行執行
res=producer()
consumer(res) #寫成consumer(producer())會下降執行效率
stop=time.time()
print(stop-start) #1.5536692142486572



#基於yield併發執行
import time
def consumer():
    '''任務1:接收數據,處理數據'''
    while True:
        x=yield

def producer():
    '''任務2:生產數據'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)

start=time.time()
#基於yield保存狀態,實現兩個任務直接來回切換,即併發的效果
#PS:若是每一個任務中都加上打印,那麼明顯地看到兩個任務的打印是你一次我一次,即併發執行的.
producer()

stop=time.time()
print(stop-start) #2.0272178649902344

單純地切換反而會下降運行效率

二：第一種狀況的切換。在任務一遇到io狀況下，切到任務二去執行，這樣就能夠利用任務一阻塞的時間完成任務二的計算，效率的提高就在於此。

import time
def consumer():
    '''任務1:接收數據,處理數據'''
    while True:
        x=yield

def producer():
    '''任務2:生產數據'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
        time.sleep(2)

start=time.time()
producer() #併發執行,可是任務producer遇到io就會阻塞住,並不會切到該線程內的其餘任務去執行

stop=time.time()
print(stop-start)

yield沒法作到遇到io阻塞

對於單線程下，咱們不可避免程序中出現io操做，但若是咱們能在本身的程序中（即用戶程序級別，而非操做系統級別）控制單線程下的多個任務能在一個任務遇到io阻塞時就切換到另一個任務去計算，這樣就保證了該線程可以最大限度地處於就緒態，即隨時均可以被cpu執行的狀態，至關於咱們在用戶程序級別將本身的io操做最大限度地隱藏起來，從而能夠迷惑操做系統，讓其看到：該線程好像是一直在計算，io比較少，從而更多的將cpu的執行權限分配給咱們的線程。

    協程的本質就是在單線程下，由用戶本身控制一個任務遇到io阻塞了就切換另一個任務去執行，以此來提高效率。爲了實現它，咱們須要找尋一種能夠同時知足如下條件的解決方案：

#1. 能夠控制多個任務之間的切換，切換以前將任務的狀態保存下來，以便從新運行時，能夠基於暫停的位置繼續執行。
#2. 做爲1的補充：能夠檢測io操做，在遇到io操做的狀況下才發生切換

 

紅色表示忙(IO)，綠色表示正常運行

左邊是正常的線程，右邊是協程

在執行程序的過程當中，遇到IO操做就切換其餘線程執行，好比b。

網絡的recv，訪問網頁，都存在IO

協程
協程並非實際存在的實體
它的本質 就是一個線程的多個部分

比線程的單位更小 —— 協程、纖程
它的本質就是一個線程的多個部分
在一個線程中能夠開啓不少協程
在執行程序的過程當中，遇到IO操做就凍結當前位置的狀態
去執行其餘任務，在執行其餘任務過程當中，
會不斷的檢測上一個凍結的任務是否IO結束
若是IO結束了，就繼續從凍結的位置開始執行

一個線程不會遇到阻塞 —— 一直在使用CPU
多個線程 —— 只能有一個線程使用CPU
協程比線程之間的切換和線程的建立銷燬
所花費的時間、空間開銷要小的多

 

協程的特色
凍結當前程序/任務的執行狀態 —— 技能解鎖
能夠規避IO操做的時間

它的特色，是線程沒有的。

 

凍結函數狀態-->生成器

def func():
    print(1)
    yield 'aaa'
    print(2)
    yield 'bbb'
    print(3)
    yield 'ccc'

g = func()
next(g)

執行輸出：1

 

def func():
    x = yield 1
    print(x)
    yield 2

g = func()
print(next(g))
print(g.send('aaa'))

執行輸出：

1

aaa

2

 

上面代碼的運行過程以下。
1.當調用next(g)方法時，python首先會執行func方法的yield 1語句。因爲是一個yield語句，所以方法的執行過程被掛起，而next方法返回值爲yield關鍵字後面表達式的值，即爲1。

2.當調用g.send('aaa')方法時，python首先恢復func方法的運行環境。同時，將表達式yield 1的返回值定義爲send方法參數的值，即爲aaa
這樣，接下來x = yield 1 這一賦值語句會將x的值置爲aaa。繼續運行會遇到yield 2語句。
所以，func方法再次被掛起。同時，send方法的返回值爲yield關鍵字後面表達式的值，爲2。
最終輸出:
1
aaa
2

 

單純的切換狀態，會不會影響程序執行時間？

看上面的例子：單純地切換反而會下降運行效率

結論：

單純的切換 仍是要耗費一些時間的 記住當前執行的狀態

上面的列表雖然執行快，可是它佔用了大量內存。它是用時間換了空間。

 

2、協程介紹                                                                   

協程：是單線程下的併發，又稱微線程，纖程。英文名Coroutine。一句話說明什麼是線程：協程是一種用戶態的輕量級線程，即協程是由用戶程序本身控制調度的。、

須要強調的是：

#1. python的線程屬於內核級別的，即由操做系統控制調度（如單線程遇到io或執行時間過長就會被迫交出cpu執行權限，切換其餘線程運行）
#2. 單線程內開啓協程，一旦遇到io，就會從應用程序級別（而非操做系統）控制切換，以此來提高效率（！！！非io操做的切換與效率無關）

對比操做系統控制線程的切換，用戶在單線程內控制協程的切換

優勢以下：

#1. 協程的切換開銷更小，屬於程序級別的切換，操做系統徹底感知不到，於是更加輕量級
#2. 單線程內就能夠實現併發的效果，最大限度地利用cpu

缺點以下：

#1. 協程的本質是單線程下，沒法利用多核，能夠是一個程序開啓多個進程，每一個進程內開啓多個線程，每一個線程內開啓協程
#2. 協程指的是單個線程，於是一旦協程出現阻塞，將會阻塞整個線程

總結協程特色：

1. 必須在只有一個單線程裏實現併發
2. 修改共享數據不需加鎖
3. 用戶程序裏本身保存多個控制流的上下文棧
4. 附加：一個協程遇到IO操做自動切換到其它協程（如何實現檢測IO，yield、greenlet都沒法實現，就用到了gevent模塊（select機制））

 

3、Greenlet模塊                                                            

簡介：

Greenlet是python的一個C擴展，來源於Stackless python，旨在提供可自行調度的‘微線程’， 即協程。generator實現的協程在yield value時只能將value返回給調用者(caller)。 而在greenlet中，target.switch（value）能夠切換到指定的協程（target）， 而後yield value。greenlet用switch來表示協程的切換，從一個協程切換到另外一個協程須要顯式指定。

 

安裝 ：pip3 install greenlet

from greenlet import greenlet

def eat(name):
    print('%s eat 1' %name)
    g2.switch('egon')
    print('%s eat 2' %name)
    g2.switch()
def play(name):
    print('%s play 1' %name)
    g1.switch()
    print('%s play 2' %name)

g1=greenlet(eat)
g2=greenlet(play)

g1.switch('egon')#能夠在第一次switch時傳入參數，之後都不須要

greenlet實現狀態切換

單純的切換（在沒有io的狀況下或者沒有重複開闢內存空間的操做），反而會下降程序的執行速度

#順序執行
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i

def f2():
    res=1
    for i in range(100000000):
        res*=i

start=time.time()
f1()
f2()
stop=time.time()
print('run time is %s' %(stop-start)) #10.985628366470337

#切換
from greenlet import greenlet
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i
        g2.switch()

def f2():
    res=1
    for i in range(100000000):
        res*=i
        g1.switch()

start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 52.763017892837524

效率對比

greenlet只是提供了一種比generator更加便捷的切換方式，當切到一個任務執行時若是遇到io，那就原地阻塞，仍然是沒有解決遇到IO自動切換來提高效率的問題。

單線程裏的這20個任務的代碼一般會既有計算操做又有阻塞操做，咱們徹底能夠在執行任務1時遇到阻塞，就利用阻塞的時間去執行任務2。。。。如此，才能提升效率，這就用到了Gevent模塊。

 

greenlet不是創造協程的模塊
它是用來作多個協程任務切換的
它究竟是怎麼實現切換的呢？

from greenlet import greenlet

def func():
    print(123)

def func2():
    print(456)

g1 = greenlet(func)  # 實例化
g2 = greenlet(func2)
g1.switch()  # 開始運行,它會運行到下一個switch結束。不然一直運行

執行輸出：123

 

from greenlet import greenlet
def test1():
    print 12
    gr2.switch()
    print 34

def test2():
    print 56
    gr1.switch()
    print 78

gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

　　

執行輸出：

12
56
34

執行過程：

當建立一個greenlet時，首先初始化一個空的棧， switch到這個棧的時候，會運行在greenlet構造時傳入的函數（首先在test1中打印 12）， 若是在這個函數（test1）中switch到其餘協程（到了test2 打印34），那麼該協程會被掛起，等到切換回來（在test2中切換回來 打印34）。當這個協程對應函數執行完畢，那麼這個協程就變成dead狀態。
　　注意 上面沒有打印test2的最後一行輸出 78，由於在test2中切換到gr1以後掛起，可是沒有地方再切換回來。這個可能形成泄漏，後面細說。

 

上面的例子，有幾個缺點
1.手動切換
2.不能規避I/O操做(睡眠)

 

4、Gevent模塊                                                              

安裝：pip3 install gevent

Gevent 是一個第三方庫，能夠輕鬆經過gevent實現併發同步或異步編程，在gevent中用到的主要模式是Greenlet, 它是以C擴展模塊形式接入Python的輕量級協程。 Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度。

g1=gevent.spawn(func,1,,2,3,x=4,y=5)建立一個協程對象g1，spawn括號內第一個參數是函數名，如eat，後面能夠有多個參數，能夠是位置實參或關鍵字實參，都是傳給函數eat的

g2=gevent.spawn(func2)

g1.join() #等待g1結束

g2.join() #等待g2結束

#或者上述兩步合做一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

用法介紹

from gevent import monkey;monkey.patch_all()

import gevent
import time
def eat():
    print('eat food 1')
    time.sleep(2)
    print('eat food 2')

def play():
    print('play 1')
    time.sleep(1)
    print('play 2')

g1=gevent.spawn(eat)
g2=gevent.spawn(play)
gevent.joinall([g1,g2])
print('主')

例：遇到io主動切換

咱們能夠用threading.current_thread().getName()來查看每一個g1和g2，查看的結果爲DummyThread-n，即假線程

from gevent import monkey;monkey.patch_all()
import threading
import gevent
import time
def eat():
    print(threading.current_thread().getName())
    print('eat food 1')
    time.sleep(2)
    print('eat food 2')

def play():
    print(threading.current_thread().getName())
    print('play 1')
    time.sleep(1)
    print('play 2')

g1=gevent.spawn(eat)
g2=gevent.spawn(play)
gevent.joinall([g1,g2])
print('主')

查看threading.current_thread().getName()

 

真正能實現協程的模塊gevent

import gevent
def eat():
    print('eating1')
    print('eating2')
g1 = gevent.spawn(eat)  #建立一個協程對象g1

執行輸出爲空，表示它還沒執行。

import gevent
def eat():
    print('eating1')
    print('eating2')
g1 = gevent.spawn(eat)  #建立一個協程對象g1
g1.join()  #等待g1結束

執行輸出：

eating1
eating2

 

import time
import gevent
def eat():
    print('eating1')
    time.sleep(1)
    print('eating2')

def play():
    print('playing1')
    time.sleep(1)
    print('playing2')

g1 = gevent.spawn(eat)  #建立一個協程對象g1
g2 = gevent.spawn(play)
g1.join()  #等待g1結束
g2.join()

執行輸出：

eating1
eating2
playing1
playing2

 

若是想順序執行呢？須要用到gevent.sleep

import time
import gevent
def eat():
    print('eating1')
    gevent.sleep(1)  #延時調用 
    print('eating2')

def play():
    print('playing1')
    gevent.sleep(1)  #延時調用 
    print('playing2')

g1 = gevent.spawn(eat)  #建立一個協程對象g1
g2 = gevent.spawn(play)
g1.join()  #等待g1結束
g2.join()

執行輸出：

eating1
playing1
eating2
playing2

 

若是想讓協程執行time.sleep()呢？因爲默認，協程沒法識別time.sleep()方法，須要導入一個模塊monkey

monkey patch (猴子補丁)
用來在運行時動態修改已有的代碼，而不須要修改原始代碼。

from gevent import monkey;monkey.patch_all()
# 它會把下面導入的全部的模塊中的IO操做都打成一個包，gevent就可以認識這些IO了
import time
import gevent
def eat():
    print('eating1')
    time.sleep(1)  #延時調用
    print('eating2')

def play():
    print('playing1')
    time.sleep(1)  #延時調用
    print('playing2')

g1 = gevent.spawn(eat)  #建立一個協程對象g1
g2 = gevent.spawn(play)
g1.join()  #等待g1結束
g2.join()

執行輸出：

eating1
playing1
eating2
playing2

 

結論：

使用gevent模塊來執行多個函數，表示在這些函數遇到IO操做的時候能夠在同一個線程中進行切換
利用其餘任務的IO阻塞時間來切換到其餘的任務繼續執行

前提是：

spawn來發布協程任務
join負責開啓並等待任務執行結束
gevent自己不認識其餘模塊中的IO操做，可是若是咱們在導入其餘模塊以前執行from gevent import monkey;monkey.patch_all()  這行代碼，必須在文件最開頭
gevent就可以認識在這句話以後導入的模塊中的全部IO操做了

 

5、Gevent之同步與異步                                                

from gevent import spawn,joinall,monkey;monkey.patch_all()

import time
def task(pid):
    """
    Some non-deterministic task
    """
    time.sleep(0.5)
    print('Task %s done' % pid)


def synchronous():  # 同步
    for i in range(10):
        task(i)

def asynchronous(): # 異步
    g_l=[spawn(task,i) for i in range(10)]
    joinall(g_l)
    print('DONE')
    
if __name__ == '__main__':
    print('Synchronous:')
    synchronous()
    print('Asynchronous:')
    asynchronous()
#  上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。
#  初始化的greenlet列表存放在數組threads中，此數組被傳給gevent.joinall 函數，
#  後者阻塞當前流程，並執行全部給定的greenlet任務。執行流程只會在 全部greenlet執行完後纔會繼續向下走。

當一個任務執行時，依賴另一個任務的結果時，這種狀況不適合異步，只能用同步

 

Gevent之應用舉例一                                                                                                                      

手動安裝模塊requests

pip3 install requests

from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time

def get_page(url):
    print('GET: %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        print('%d bytes received from %s' %(len(response.text),url))


start_time=time.time()
gevent.joinall([
    gevent.spawn(get_page,'https://www.python.org/'),
    gevent.spawn(get_page,'https://www.yahoo.com/'),
    gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))

協程應用：爬蟲

等待網頁請求結果是，去執行其餘任務

紅色表示等待，綠色表示執行任務

 另一個爬蟲例子：

對比使用普通函數和使用協程，誰更快一點

因爲操做系統，訪問一次網頁後，會有緩存。

因此測試時，先訪問一遍網頁。再分別測試協程和普通函數。

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

def get_page(url):
    res = urlopen(url)
    #print(len(res.read()))

url_lst = [
    'http://www.baidu.com',
    'http://www.sogou.com',
    'http://www.sohu.com',
    'http://www.qq.com',
    'http://www.cnblogs.com',
]
start = time.time()
gevent.joinall([gevent.spawn(get_page,url) for url in url_lst])
print('先執行一次',time.time() - start)

start = time.time()
gevent.joinall([gevent.spawn(get_page,url) for url in url_lst])
print('協程',time.time() - start)

start = time.time()
for url in url_lst:get_page(url)
print('普通',time.time() - start)

執行輸出：

先執行一次 0.6465449333190918
協程 0.34525322914123535
普通 0.570899486541748

 結論
之後用爬蟲，可使用協程，它的速度更快。

 

Gevent之應用舉例二                                                                                                                      

經過gevent實現單線程下的socket併發

注意 ：from gevent import monkey;monkey.patch_all()必定要放到導入socket模塊以前，不然gevent沒法識別socket的阻塞

from gevent import monkey;monkey.patch_all()
from socket import *
import gevent

#若是不想用money.patch_all()打補丁,能夠用gevent自帶的socket
# from gevent import socket
# s=socket.socket()

def server(server_ip,port):
    s=socket(AF_INET,SOCK_STREAM)
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
    s.bind((server_ip,port))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    try:
        while True:
            res=conn.recv(1024)
            print('client %s:%s msg: %s' %(addr[0],addr[1],res))
            conn.send(res.upper())
    except Exception as e:
        print(e)
    finally:
        conn.close()

if __name__ == '__main__':
    server('127.0.0.1',8080)

server

from socket import *

client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))


while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

client

from threading import Thread
from socket import *
import threading

def client(server_ip,port):
    c=socket(AF_INET,SOCK_STREAM) #套接字對象必定要加到函數內，即局部名稱空間內，放在函數外則被全部線程共享，則你們公用一個套接字對象，那麼客戶端端口永遠同樣了
    c.connect((server_ip,port))

    count=0
    while True:
        c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
        msg=c.recv(1024)
        print(msg.decode('utf-8'))
        count+=1
if __name__ == '__main__':
    for i in range(500):
        t=Thread(target=client,args=('127.0.0.1',8080))
        t.start()

多線程併發多個客戶端

 

進程5個，線程20個，協程500個 —— 通用的組合  —— 50000qps

0.3s之內，用戶是感受不到的

只有進程能處理並行

重點掌握進程，線程，協程

這3者之間的區別，優缺點，理論知識。面試會問道。

task，翻譯是任務

1 多進程/多線程網絡編程都是一個進程或者線程處理一個task，當task過多時，就會致使巨量的進程/線程。 
2 巨量的進程/線程會致使 上下文切換極其頻繁！  你們知道：上下文切換是要消耗cpu資源的 因此當進程/線程數量過多時，cpu資源就得不到有效利用
3 而協程實際上就是：在用戶空間實現task的上下文切換！ 這種上下文切換消耗的代價相較而言微乎其微。這就是協程的優點！
4 固然協程也有劣勢：就是沒法利用多核cpu，可是咱們有解決辦法：多進程 + 協程 

 

看下圖

 playing2沒有輸出，是由於阻塞結束，再也不切換。

 

明天默寫：

socket_server

from gevent import monkey;monkey.patch_all()
import socket
import gevent
def async_talk(conn):
    try:
        while True:
            conn.send(b'hello')
            ret = conn.recv(1024)
            print(ret)
    finally:
        conn.close()
sk = socket.socket()
sk.bind(('127.0.0.1',9000))
sk.listen()
while True:
    conn,addr = sk.accept()
    gevent.spawn(async_talk,conn)
sk.close()

socket_client

import socket
from threading import Thread
def socket_client():
    sk = socket.socket()
    sk.connect(('127.0.0.1',9000))
    while True:
        print(sk.recv(1024))
        sk.send(b'bye')
    sk.close()
for i in range(500):
    Thread(target=socket_client).start()