1七、第七 - 網絡編程基礎 - 協程概念介紹、協程gevent模塊併發爬網頁

時間 2019-12-01

原文原文鏈接

協程，又稱微線程，纖程。什麼是線程：協程是一種用戶態的輕量級線程。html

　　協程擁有本身的寄存器上下文和棧。協程調度切換時，將寄存器上下文和棧保存到其餘地方，在切回來的時候，恢復先前保存的寄存器上下文和棧。所以：協程能保留上一次調用時的狀態（即全部局部狀態的一個特定組合），每次過程重入時，就至關於進入上一次調用的狀態，換種說法：進入上一次離開時所處邏輯流的位置。python

協程的好處：編程

無需線程上下文切換的開銷
無需原子操做鎖定及同步的開銷（註解："原子操做(atomic operation)是不須要synchronized"，所謂原子操做是指不會被線程調度機制打斷的操做；這種操做一旦開始，就一直運行到結束，中間不會有任何（切換到另外一個線程。原子操做能夠是一個步驟，也能夠是多個操做步驟，可是其順序是不能夠被打亂，或者切割掉只執行部分。視做總體是原子性的核心。）
方便切換控制流，簡化編程模型
高併發+高擴展性+低成本：一個CPU支持上萬的協程都不是問題。因此很適合用於高併發處理。

協程的缺點：數組

沒法利用多核資源：協程的本質是個單線程,它不能同時將單個CPU 的多個核用上,協程須要和進程配合才能運行在多CPU上.固然咱們平常所編寫的絕大部分應用都沒有這個必要，除非是cpu密集型應用。
進行阻塞（Blocking）操做（如IO時）會阻塞掉整個程序

符合協程的標準網絡

必須在只有一個單線程裏實現併發
修改共享數據不需加鎖
用戶程序裏本身保存多個控制流的上下文棧
一個協程遇到IO操做自動切換到其它協程

舉例：併發

一、使用yield實現協程操做：

import time
import queue


def consumer(name):
    print("--->starting eating baozi...")
    while True:
        new_baozi = yield
        print("[%s] is eating baozi %s" % (name, new_baozi))
        # time.sleep(1)


def producer():
    r = con.__next__()
    r = con2.__next__()
    n = 0
    while n < 5:
        n += 1
        con.send(n)
        con2.send(n)
        print("\033[32;1m[producer]\033[0m is making baozi %s" % n)


if __name__ == '__main__':
    con = consumer("c1")
    con2 = consumer("c2")
    p = producer()
輸出：
--->starting eating baozi...
--->starting eating baozi...
[c1] is eating baozi 1
[c2] is eating baozi 1
[producer] is making baozi 1
[c1] is eating baozi 2
[c2] is eating baozi 2
[producer] is making baozi 2
[c1] is eating baozi 3
[c2] is eating baozi 3
[producer] is making baozi 3
[c1] is eating baozi 4
[c2] is eating baozi 4
[producer] is making baozi 4
[c1] is eating baozi 5
[c2] is eating baozi 5
[producer] is making baozi 5

二、使用第三方模塊：greenlet （手動指定執行切換協程）

greenlet是一個用C實現的協程模塊，相比與python自帶的yield，它可使你在任意函數之間隨意切換，而不需把這個函數先聲明爲generator。less

　　greenlet是python的並行處理的一個庫。 python 有一個很是有名的庫叫作 stackless ，用來作併發處理，主要是弄了個叫作tasklet的微線程的東西，而greenlet 跟stackless的最大區別是greenlet須要你本身來處理線程切換，就是說，你須要本身指定如今執行哪一個greenlet再執行哪一個greenlet。至關於手動切換協程。異步

　　一個「greenlet」是一個小型的獨立僞線程。能夠把它想像成一些棧幀，棧底是初始調用的函數，而棧頂是當前greenlet的暫停位置。你使用greenlet建立一堆這樣的堆棧，而後在他們之間跳轉執行。跳轉必須顯式聲明的：一個greenlet必須選擇要跳轉到的另外一個greenlet，這會讓前一個掛起，然後一個在此前掛起處恢復執行。不一樣greenlets之間的跳轉稱爲切換(switching) 。socket

　　當你建立一個greenlet時，它獲得一個開始時爲空的棧；當你第一次切換到它時，它會執行指定的函數，這個函數可能會調用其餘函數、切換跳出greenlet等等。當最終棧底的函數執行結束出棧時，這個greenlet的棧又變成空的，這個greenlet也就死掉了。greenlet也會由於一個未捕捉的異常死掉。async

舉例：

from greenlet import greenlet

def test1():
    print(12) #二、打印
    gr2.switch()#三、切換協程到 test2
    print(34)#六、打印
    gr2.switch()#七、切換到test2

def test2():
    print(56)#四、打印
    gr1.switch() #五、切換協程到 test1
    print(78)#八、打印，執行完成

gr1 = greenlet(test1) #啓動一個協程
gr2 = greenlet(test2) #啓動一個協程
gr1.switch()  #一、開始調用切換協程

輸出：
12
56
34
78

注意：執行的步驟順序，從1-8。

以上例子還不能實如今協程中自動切換，greenlet 只能手動指定執行，但對於生成器來講簡單不少。要實現自動監控，而且自動切換協程，如何實現？引入gevent模塊。

三、使用第三方模塊：gevent （自動監控並自動切換協程）

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

　　Gevent是第三方庫，經過greenlet實現協程，其基本思想是：當一個greenlet遇到IO操做時，好比訪問網絡，就自動切換到其餘的greenlet，等到IO操做完成，再在適當的時候切換回來繼續執行。因爲IO操做很是耗時，常常使程序處於等待狀態，有了gevent爲咱們自動切換協程，就保證總有greenlet在運行，而不是等待IO。因爲切換是在IO操做時自動完成，因此gevent須要修改Python自帶的一些標準庫，這一過程在啓動時經過monkey patch完成。

包含的特性：
１．基於libev的快速事件循環
２．基於greenlet的輕量級執行單元
３．重用Python標準庫且概念類似的API
４．支持SSL的協做socket
５．經過c-ares或者線程池進行DNS查詢
６．使用標準庫和第三方庫中使用了阻塞socket的代碼的能力

第三庫須要另外，開源包進行安裝。

舉例：

A、驗證gevent經過自動判斷，選擇最優的線路進行判斷執行。注意：可gevent.sleep() 調整時間，進行驗證測試。結論：每一個函數裏面最後一次打印,看等待的時間越短越先執行。

import gevent

def foo():
    print('running foo')
    gevent.sleep(3)
    print('Explicit context switch to foo again')
def bar():
    print('running bar')
    gevent.sleep(6)
    print('Implicit context switch back to bar ')
def func3():
    print("running func3")
    gevent.sleep(1)
    print ("switch back to func3")

gevent.joinall([
    gevent.spawn(foo),
    gevent.spawn(bar),
    gevent.spawn(func3)
])
輸出：
running foo
running bar
running func3
switch back to func3
Explicit context switch to foo again
Implicit context switch back to bar

B、同步與異步性能對區別，以下：

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

import gevent

def task(pid):
    gevent.sleep(1)
    print('Task %s done' % pid)

def synchronous():
    for i in range(1, 6): #range從1到5打印
        task(i)

def asynchronous():
    threads = [gevent.spawn(task, i) for i in range(5)]
    gevent.joinall(threads) #


print('Synchronous:')
synchronous() #正常函數，串行的調用會每一個1s打印一次

print(" ")

print('Asynchronous:')
asynchronous() #並行打印，等待一次性打印出來
輸出：
每相隔一秒打印
Synchronous:
Task 1 done
Task 2 done
Task 3 done
Task 4 done
Task 5 done
 
等待同一時間打印
Asynchronous:
Task 0 done
Task 1 done
Task 2 done
Task 3 done
Task 4 done

C、gevent協程爬取網頁，遇到IO阻塞時會自動切換業務。舉例以下：

from gevent import monkey;
monkey.patch_all()

import gevent,time
from  urllib.request import urlopen

def f(url):
    print('GET: %s' % url)
    resp = urlopen(url)
    data = resp.read()
    #print(data) #打印爬取到的網頁內容
    print('%d bytes received from %s.' % (len(data), url))

time_start = time.time()

urls = ['http://www.cnblogs.com/alex3714/articles/5248247.html',
	    'http://www.cnblogs.com/chen170615/p/8797609.html',
	    'http://www.cnblogs.com/chen170615/p/8761768.html',
        ]
for i in urls:
    f(i)

print("同步執行時間：",time.time() - time_start)
print (" ")
async_time_start = time.time()
gevent.joinall([
    gevent.spawn(f, 'http://www.cnblogs.com/alex3714/articles/5248247.html'),
    gevent.spawn(f, 'http://www.cnblogs.com/chen170615/p/8797609.html'),
    gevent.spawn(f,'http://www.cnblogs.com/chen170615/p/8761768.html')
])
print("異步執行時間：",time.time() - async_time_start)
輸出：
GET: http://www.cnblogs.com/alex3714/articles/5248247.html
92147 bytes received from http://www.cnblogs.com/alex3714/articles/5248247.html.
GET: http://www.cnblogs.com/chen170615/p/8797609.html
10930 bytes received from http://www.cnblogs.com/chen170615/p/8797609.html.
GET: http://www.cnblogs.com/chen170615/p/8761768.html
11853 bytes received from http://www.cnblogs.com/chen170615/p/8761768.html.
同步執行時間 20.319132089614868
 
GET: http://www.cnblogs.com/alex3714/articles/5248247.html
GET: http://www.cnblogs.com/chen170615/p/8797609.html
GET: http://www.cnblogs.com/chen170615/p/8761768.html
11853 bytes received from http://www.cnblogs.com/chen170615/p/8761768.html.
10930 bytes received from http://www.cnblogs.com/chen170615/p/8797609.html.
92147 bytes received from http://www.cnblogs.com/alex3714/articles/5248247.html.
異步執行時間： 0.28768205642700195

以上例子能夠看出，gevent協程異步併發執行的性能高於同步串行的執行，遇到會等待的IO同時，異步的性能就表現的優異起來。（多執行幾回，就能看出對比。）

D、通過gevent實現單線程下的多socket併發

舉例：

服務端：

協程gevent_socket_server.py

import sys,socket,time,gevent

from gevent import socket,monkey

monkey.patch_all()

def server(port):
    gevent_server = socket.socket()
    gevent_server.bind(('0.0.0.0',port))
    gevent_server.listen()
    while True:
        cli,addr = gevent_server.accept()
        gevent.spawn(handle_request,cli)

def handle_request(conn):
    try:
        while True:
            data = conn.recv(1024)
            print("recv:",data)
            conn.send(data)
            if not data:
                conn.shutdown(socket.SHUT_WR)

    except Exception as ex:
        print(ex)

    finally:
        conn.close()

if __name__ == "__main__":
    server(8001)

客戶端兩種類型，以下：

客戶端1：協程gevent_socket_client.py（普通的手工輸入模式）

import socket

HOST = "localhost"
PORT = 8001

s = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
s.connect((HOST,PORT))

while True:
    msg = bytes(input(">>:").strip(),encoding="utf-8")
    s.sendall(msg)
    data = s.recv(1024)

    print('Received',repr(data))

s.close()

客戶端2：協程gevent_socket_cli.py（經過起進程的方式，併發執行）

import socket,threading

HOST = "localhost"
PORT = 8001

def sock_conn():

   s = socket.socket()
   s.connect((HOST,PORT))
   count = 0

   while True:
       s.sendall(("hello %s" % count).encode("utf-8"))
       data = s.recv(1024)

       print("[%s]recv from server:" % threading.get_ident(),data.decode())
       count += 1
   s.close()

for i in range(10): #測試注意數值，不要設置太大。要否則，機器回被卡死
    t = threading.Thread(target=sock_conn)
    t.start()