併發編程之協程

1、引子

　　本節的主題是基於單線程來實現併發，即只用一個主線程（很明顯可利用的cpu只有一個）狀況下實現併發，爲此咱們須要先回顧下併發的本質：切換+保存狀態

一、切換：

　　cpu正在運行一個任務，會在兩種狀況下切走去執行其餘的任務（切換由操做系統強制控制）：

　　一種狀況是該任務發生了阻塞；

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

　　另一種狀況是該任務計算的時間過長或有一個優先級更高的程序替代了它。

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

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

二、保存狀態：

　　yield自己就是一種在單線程下能夠保存任務運行狀態的方法，yield能夠保存狀態，yield的狀態保存和操做系統的保存線程狀態很是像，可是yield是代碼級別控制的，更輕量級。

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

　　驗證示例下：

# 併發
import time
def producer():
    g = consumer()
    next(g)
    for i in range(1000000):
        g.send(i)   # send一次就會切到consumer


def consumer():
    while True:
        res = yield   # yield收到producer的i的值，處理完切回producer的for循環

start_time = time.time()
producer()
stop_time = time.time()
print(stop_time-start_time)
"""0.26706719398498535"""

併發示例

# 併發切換分兩種，但只有遇到阻塞了切換會提高效率，串行改寫以下：
import time
def producer():
    res = []
    for i in range(1000000):
        res.append(i)
    return res

def consumer(res):
    pass

start_time = time.time()
res = producer()
consumer(res)
stop_time = time.time()
print(stop_time-start_time)
"""0.18116021156311035"""

串行改寫示例

　　能夠看到這種形式的併發反而下降了效率。

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

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

　　一、能夠控制多個任務之間的切換，切換以前將任務的狀態保存下來，以便從新運行時，能夠基於暫停的位置繼續執行。

　　二、做爲第一條的補充，能夠檢測IO操做，在遇到Io操做的狀況下才發生切換。

2、協程介紹

　　協程的概念：單線程下的併發，又稱微線程，纖程。英文名Coroutine。

　　一句話說明什麼是協程：協程是一種用戶態的輕量級線程，即協程是由用戶程序本身控制調度的。

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

優勢：
　　一、協程的切換開銷更小，屬於程序級別的切換，操做系統徹底感知不到，於是更加輕量級
　　二、單線程內就能夠實現併發的效果，最大限度地利用cpu （佔用CPU的時間多了，就叫作效率高，對操做系統的一種欺騙）

缺點：
　　一、協程的本質是單線程下，沒法利用多核，能夠是一個程序開啓多個進程，每一個進程內開啓多個線程，每一個線程內開啓協程
　　二、協程指的是單個線程，於是一旦協程出現阻塞，將會阻塞整個線程（只能實現併發，不能實現並行）

總結協程特色：
　　一、必須在只有一個單線程裏實現併發
　　二、修改共享數據不需加鎖（本質是一個一個執行，所以不須要考慮加鎖）
　　三、用戶程序裏本身保存多個控制流的上下文棧（本身控制切換）
　　四、附加：一個協程遇到IO操做自動切換到其它協程(如何實現檢測IO，yield、greenlet都沒法實現，就用到了gevent模塊(select機制))

3、Greenlet模塊

　　若是咱們在單個線程內有20個任務，要想實如今多個任務之間切換，使用yield生成器的方式過於麻煩（須要先獲得初始化一次的生成器，而後再調用send。。。很是麻煩）。

　　greenlet模塊：比generator更加便捷的切換方式，多個任務間很方便的切換。但不能監測I/O，所以比yield強一點，但不是研究的重點。

　　安裝greenlet：pip3 install greenlet

from greenlet import greenlet

def eat(name):
    print("%s eat 1" % name)
    g2.switch("hqs")
    print("%s eat 2" % name)
    g2.switch()

def play(name):
    print("%s play 1" % name)
    g1.switch()
    print("%s paly 2" % name)

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

g1.switch('egon')   #能夠在第一次switch時傳入參數，之後都不須要
"""
egon eat 1
hqs play 1
egon eat 2
hqs paly 2
"""

　　檢查greenlet面對i/o的狀況是否能夠切換：不能切換，沒法提高效率

from greenlet import greenlet
import time

def eat(name):
    print("%s eat 1" % name)
    time.sleep(10)
    g2.switch("hqs")
    print("%s eat 2" % name)
    g2.switch()

def play(name):
    print("%s play 1" % name)
    g1.switch()
    print("%s paly 2" % name)

g1 = greenlet(eat)
g2 = greenlet(play)
g1.switch('egon')
"""
egon eat 1   # 這一步卡了足足10秒
hqs play 1
egon eat 2
hqs paly 2
"""

檢驗發現greenlet不支持i/o切換

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

　　單純的切換（在沒有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對比示例

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

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的返回值

　　驗證gevent遇到IO阻塞時是否會自動切換任務：

import gevent
import time

def eat(name):
    print("%s eat 1" % name)
    # time.sleep(1)  # 不切，不能識別其餘程序的I/O操做
    gevent.sleep(3)  # gevent只能識別本身模擬的I/O操做
    print("%s eat 2" % name)

def play(name):
    print("%s play 1" % name)
    gevent.sleep(4)
    print("%s paly 2" % name)


start_time = time.time()
g1 = gevent.spawn(eat, 'egon')
g2 = gevent.spawn(play, "alex")
# 異步提交，提交後不會等待完成結果

g1.join()
g2.join()
stop_time = time.time()
print(stop_time-start_time)
"""
egon eat 1
alex play 1
egon eat 2  ----->等三秒
alex paly 2 —————>再等一秒
4.006500005722046
"""

gevent使用及驗證代碼

　　上例gevent.sleep(2)模擬的是gevent能夠識別的io阻塞,而time.sleep(2)或其餘的阻塞,gevent是不能直接識別的須要用下面一行代碼,打補丁,就能夠識別了。

　　from gevent import monkey;monkey.patch_all()必須放到被打補丁者的前面，如time，socket模塊以前

或者咱們乾脆記憶成：要用gevent，須要將from gevent import monkey;monkey.patch_all()放到文件的開頭。

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('主')
"""
eat food 1
play 1
play 2
eat food 2
主
"""

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

5、Gevent之同步與異步

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

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

def eat(name):
    print('%s eat food 1' % name)
    time.sleep(2)
    print('%s eat food 2' % name)

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

g1=gevent.spawn(eat, 'egon')   # 建立一個協程對象g1，spawn括號內第一個參數是函數名
g2=gevent.spawn(play, 'alex')
"""
直接執行上述代碼，沒有任何輸出。線程把代碼運行完直接結束，任務還沒來的及開啓就結束了。
"""
# time.sleep(1)  # 睡一秒是不夠的，到時間，線程依然是結束了
"""
egon eat food 1
alex play 1
"""
# time.sleep(5)
"""
egon eat food 1
alex play 1
alex play 2
egon eat food 2  ————》這一行輸出完後，還要等sleep的5秒所有睡晚，程序結束
"""
# 解決方案一：上述兩步合做一步
# g1.join()  # 等待g1結束
# g2.join()  # 等待g2結束

# 解決方案二：
gevent.joinall([g1, g2])

　　同步、異步對比：

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)

if __name__ == '__main__':
    print('Synchronous:')
    synchronous()

    print('Asynchronous:')
    asynchronous()
#上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。 初始化的greenlet列表存放在數組threads中，此數組被傳給gevent.joinall 函數，後者阻塞當前流程，並執行全部給定的greenlet。執行流程只會在 全部greenlet執行完後纔會繼續向下走。

gevent示例

6、Gevent之應用舉例

 　　協程應用：爬蟲

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))

gevent實現爬蟲

 7、Gevent之應用舉例二

　　經過gevent實現單線程下的socket併發（from gevent import monkey;monkey.patch_all()必定要放到導入socket模塊以前，不然gevent沒法識別socket的阻塞）

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

# 基於gevent實現
from gevent import monkey;monkey.patch_all()  # 標準寫法，打補丁識別全部阻塞
#若是不想用money.patch_all()打補丁,能夠用gevent自帶的socket
# from gevent import socket
# s=socket.socket()
from gevent import spawn
from socket import *


def communicate(conn):
    while True:
        try:
            data = conn.recv(1024)
            if not data: break
            conn.send(data.upper())

        except ConnectionResetError:
            break
    conn.close()


def server(ip, port):
    server = socket(AF_INET, SOCK_STREAM)
    server.bind((ip, port))
    server.listen(5)

    while True:
        conn, addr = server.accept()  # 建連接
        spawn(communicate, conn)

    server.close()


if __name__ == '__main__':
    g = spawn(server, '127.0.0.1', 8091)
    g.join()   # 若是不寫這個，程序直接結束

服務端

# -*- coding:utf-8 -*-
__author__ = 'Qiushi Huang'

from socket import *
from threading import Thread,currentThread

# 併發多個線程同時發消息
def client():
    client = socket(AF_INET, SOCK_STREAM)
    client.connect(("127.0.0.1", 8091))

    while True:
        client.send(("%s hello" % currentThread().getName()).encode('utf-8'))  # 發送的消息必須轉爲bytes
        data = client.recv(1024)
        print(data.decode("utf-8"))

    client.close()


if __name__ == '__main__':
    for i in range(500):
        t = Thread(target=client)
        t.start()

客戶端

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()

多線程併發多個客戶端