IO多路複用，協程，

時間 2019-11-13

標籤多路简体版

原文原文鏈接

一.單線程的併發

import socket
import select

client1 = socket.socket()
client1.setblocking(False) # 百度建立鏈接: 非阻塞

try:
    client1.connect(('www.baidu.com',80))
except BlockingIOError as e:
    pass

client2 = socket.socket()
client2.setblocking(False) # 搜狗建立鏈接: 非阻塞
try:
    client2.connect(('www.sogou.com',80))
except BlockingIOError as e:
    pass

client3 = socket.socket()
client3.setblocking(False) # 建立鏈接: 非阻塞
try:
    client3.connect(('www.aiqiyi.com',80))
except BlockingIOError as e:
    pass

socket_list = [client1,client2,client3]
conn_list = [client1,client2,client3]

while True:
    rlist,wlist,elist = select.select(socket_list,conn_list,[],0.005)
    # wlist中表示已經鏈接成功的socket對象
    for sk in wlist:
        if sk == client1:
            sk.sendall(b'GET /s?wd=alex HTTP/1.0\r\nhost:www.baidu.com\r\n\r\n')
        elif sk==client2:
            sk.sendall(b'GET /web?query=fdf HTTP/1.0\r\nhost:www.sogou.com\r\n\r\n')
        else:
            sk.sendall(b'GET /s?wd=alex HTTP/1.0\r\nhost:www.aiqiyi.com\r\n\r\n')
        conn_list.remove(sk)
    for sk in rlist:
        chunk_list = []
        while True:
            try:
                chunk = sk.recv(8096)
                if not chunk:
                    break
                chunk_list.append(chunk)
            except BlockingIOError as e:
                break
        body = b''.join(chunk_list)
        # print(body.decode('utf-8'))
        print('------------>',body)
        sk.close()
        socket_list.remove(sk)
    if not socket_list:
        break

操做系統檢測socket是否發生變化，有三種模式：
select：最多1024個socket；循環去檢測。
poll：不限制監聽socket個數；循環去檢測（水平觸發）。
epoll：不限制監聽socket個數；回調方式（邊緣觸發）。
Python模塊：
select.select
select.epollpython

二.協程

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

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

協程，是由程序員創造出來的一個不是真實存在的東西；

協程：是微線程，對一個線程進程分片，使得線程在代碼塊之間進行來回切換執行，而不是在原來逐行執行。

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

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

優勢以下：github

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

缺點以下：web

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

總結協程特色：編程

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

三.greenlet模塊

安裝：pip3 install greenletwindows

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的狀況下或者沒有重複開闢內存空間的操做），反而會下降程序的執行速度數組

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

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

import gevent
def eat(name):
    print('%s eat 1' %name)
    gevent.sleep(2)
    print('%s eat 2' %name)

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


g1=gevent.spawn(eat,'egon')
g2=gevent.spawn(play,name='egon')
g1.join()
g2.join()
#或者gevent.joinall([g1,g2])
print('主')

例：遇到io主動切換

上例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('主')

#   協程IO切換
from gevent import monkey
monkey.patch_all() # 之後代碼中遇到IO都會自動執行greenlet的switch進行切換
import requests
import gevent

def get_page1(url):
    ret = requests.get(url)
    print(url,ret.content)

def get_page2(url):
    ret = requests.get(url)
    print(url,ret.content)

def get_page3(url):
    ret = requests.get(url)
    print(url,ret.content)

gevent.joinall([
    gevent.spawn(get_page1, 'https://www.python.org/'), # 協程1
    gevent.spawn(get_page2, 'https://www.yahoo.com/'),  # 協程2
    gevent.spawn(get_page3, 'https://github.com/'),     # 協程3
])

Gevent之同步與異步

1.同步就是發生調用時，必定等待結果返回，整個調用才結束；按照順序逐步執行；

2異步，通知，執行完成以後自動執行回調函數或自動執行某些操做（通知）。

好比作爬蟲中向某個地址baidu.com發送請求，當請求執行完成以後自執行回調函數。

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執行完後纔會繼續向下走。

　同步異步與阻塞，非阻塞區別

　　1.阻塞/非阻塞, 它們是程序在等待消息(無所謂同步或者異步)時的狀態；

　　2.同步/異步，是程序得到關注消息通知的機制。

import socket
import select

class Req(object):
    def __init__(self,sk,func):
        self.sock = sk
        self.func = func

    def fileno(self):
        return self.sock.fileno()


class Nb(object):

    def __init__(self):
        self.conn_list = []
        self.socket_list = []

    def add(self,url,func):
        client = socket.socket()
        client.setblocking(False)  # 非阻塞
        try:
            client.connect((url, 80))
        except BlockingIOError as e:
            pass
        obj = Req(client,func)
        self.conn_list.append(obj)
        self.socket_list.append(obj)

    def run(self):

        while True:
            rlist,wlist,elist = select.select(self.socket_list,self.conn_list,[],0.005)
            # wlist中表示已經鏈接成功的req對象
            for sk in wlist:
                # 發生變換的req對象
                sk.sock.sendall(b'GET /s?wd=alex HTTP/1.0\r\nhost:www.baidu.com\r\n\r\n')
                self.conn_list.remove(sk)
            for sk in rlist:
                chunk_list = []
                while True:
                    try:
                        chunk = sk.sock.recv(8096)
                        if not chunk:
                            break
                        chunk_list.append(chunk)
                    except BlockingIOError as e:
                        break
                body = b''.join(chunk_list)
                # print(body.decode('utf-8'))
                sk.func(body)
                sk.sock.close()
                self.socket_list.remove(sk)
            if not self.socket_list:
                break

基於事件循環實現的異步非阻塞框架

Python中開源基於事件循環實現的異步非阻塞框架 Twisted

from twisted.web.client import getPage, defer
        from twisted.internet import reactor

        def all_done(arg):
            reactor.stop()

        def callback(contents):
            print(contents)

        deferred_list = []
        url_list = ['http://www.bing.com', 'http://www.baidu.com', ]
        for url in url_list:
            deferred = getPage(bytes(url, encoding='utf8'))
            deferred.addCallback(callback)
            deferred_list.append(deferred)

        dlist = defer.DeferredList(deferred_list)
        dlist.addBoth(all_done)

        reactor.run()

小結

使用gevent，能夠得到極高的併發性能，但gevent只能在Unix/Linux下運行，在Windows下不保證正常安裝和運行。

因爲gevent是基於IO切換的協程，因此最神奇的是，咱們編寫的Web App代碼，不須要引入gevent的包，也不須要改任何代碼，僅僅在部署的時候，用一個支持gevent的WSGI服務器，馬上就得到了數倍的性能提高。

五.IO多路複用三種模式

IO複用：爲了解釋這個名詞，首先來理解下複用這個概念，複用也就是共用的意思，這樣理解仍是有些抽象，爲此，我們來理解下複用在通訊領域的使用，在通訊領域中爲了充分利用網絡鏈接的物理介質，每每在同一條網絡鏈路上採用時分複用或頻分複用的技術使其在同一鏈路上傳輸多路信號，到這裏咱們就基本上理解了複用的含義，即公用某個「介質」來儘量多的作同一類(性質)的事，那IO複用的「介質」是什麼呢？爲此咱們首先來看看服務器編程的模型，客戶端發來的請求服務端會產生一個進程來對其進行服務，每當來一個客戶請求就產生一個進程來服務，然而進程不可能無限制的產生，所以爲了解決大量客戶端訪問的問題，引入了IO複用技術，即：一個進程能夠同時對多個客戶請求進行服務。也就是說IO複用的「介質」是進程(準確的說複用的是select和poll，由於進程也是靠調用select和poll來實現的)，複用一個進程(select和poll)來對多個IO進行服務，雖然客戶端發來的IO是併發的可是IO所需的讀寫數據多數狀況下是沒有準備好的，所以就能夠利用一個函數(select和poll)來監聽IO所需的這些數據的狀態，一旦IO有數據能夠進行讀寫了，進程就來對這樣的IO進行服務。

  

理解完IO複用後，咱們在來看下實現IO複用中的三個API(select、poll和epoll)的區別和聯繫

select，poll，epoll都是IO多路複用的機制，I/O多路複用就是經過一種機制，能夠監視多個描述符，一旦某個描述符就緒（通常是讀就緒或者寫就緒），可以通知應用程序進行相應的讀寫操做。但select，poll，epoll本質上都是同步I/O，由於他們都須要在讀寫事件就緒後本身負責進行讀寫，也就是說這個讀寫過程是阻塞的，而異步I/O則無需本身負責進行讀寫，異步I/O的實現會負責把數據從內核拷貝到用戶空間。三者的原型以下所示：

int select(int nfds, fd_set *readfds, fd_set *writefds, fd_set *exceptfds, struct timeval *timeout);

int poll(struct pollfd *fds, nfds_t nfds, int timeout);

int epoll_wait(int epfd, struct epoll_event *events, int maxevents, int timeout);



 1.select的第一個參數nfds爲fdset集合中最大描述符值加1，fdset是一個位數組，其大小限制爲__FD_SETSIZE（1024），位數組的每一位表明其對應的描述符是否須要被檢查。第二三四參數表示須要關注讀、寫、錯誤事件的文件描述符位數組，這些參數既是輸入參數也是輸出參數，可能會被內核修改用於標示哪些描述符上發生了關注的事件，因此每次調用select前都須要從新初始化fdset。timeout參數爲超時時間，該結構會被內核修改，其值爲超時剩餘的時間。

 select的調用步驟以下：

（1）使用copy_from_user從用戶空間拷貝fdset到內核空間

（2）註冊回調函數__pollwait

（3）遍歷全部fd，調用其對應的poll方法（對於socket，這個poll方法是sock_poll，sock_poll根據狀況會調用到tcp_poll,udp_poll或者datagram_poll）

（4）以tcp_poll爲例，其核心實現就是__pollwait，也就是上面註冊的回調函數。

（5）__pollwait的主要工做就是把current（當前進程）掛到設備的等待隊列中，不一樣的設備有不一樣的等待隊列，對於tcp_poll 來講，其等待隊列是sk->sk_sleep（注意把進程掛到等待隊列中並不表明進程已經睡眠了）。在設備收到一條消息（網絡設備）或填寫完文件數 據（磁盤設備）後，會喚醒設備等待隊列上睡眠的進程，這時current便被喚醒了。

（6）poll方法返回時會返回一個描述讀寫操做是否就緒的mask掩碼，根據這個mask掩碼給fd_set賦值。

（7）若是遍歷完全部的fd，尚未返回一個可讀寫的mask掩碼，則會調用schedule_timeout是調用select的進程（也就是 current）進入睡眠。當設備驅動發生自身資源可讀寫後，會喚醒其等待隊列上睡眠的進程。若是超過必定的超時時間（schedule_timeout 指定），仍是沒人喚醒，則調用select的進程會從新被喚醒得到CPU，進而從新遍歷fd，判斷有沒有就緒的fd。

（8）把fd_set從內核空間拷貝到用戶空間。

總結下select的幾大缺點：

（1）每次調用select，都須要把fd集合從用戶態拷貝到內核態，這個開銷在fd不少時會很大

（2）同時每次調用select都須要在內核遍歷傳遞進來的全部fd，這個開銷在fd不少時也很大

（3）select支持的文件描述符數量過小了，默認是1024

 

2．  poll與select不一樣，經過一個pollfd數組向內核傳遞須要關注的事件，故沒有描述符個數的限制，pollfd中的events字段和revents分別用於標示關注的事件和發生的事件，故pollfd數組只須要被初始化一次。

 poll的實現機制與select相似，其對應內核中的sys_poll，只不過poll向內核傳遞pollfd數組，而後對pollfd中的每一個描述符進行poll，相比處理fdset來講，poll效率更高。poll返回後，須要對pollfd中的每一個元素檢查其revents值，來得指事件是否發生。

 

3．直到Linux2.6纔出現了由內核直接支持的實現方法，那就是epoll，被公認爲Linux2.6下性能最好的多路I/O就緒通知方法。epoll能夠同時支持水平觸發和邊緣觸發（Edge Triggered，只告訴進程哪些文件描述符剛剛變爲就緒狀態，它只說一遍，若是咱們沒有采起行動，那麼它將不會再次告知，這種方式稱爲邊緣觸發），理論上邊緣觸發的性能要更高一些，可是代碼實現至關複雜。epoll一樣只告知那些就緒的文件描述符，並且當咱們調用epoll_wait()得到就緒文件描述符時，返回的不是實際的描述符，而是一個表明就緒描述符數量的值，你只須要去epoll指定的一個數組中依次取得相應數量的文件描述符便可，這裏也使用了內存映射（mmap）技術，這樣便完全省掉了這些文件描述符在系統調用時複製的開銷。另外一個本質的改進在於epoll採用基於事件的就緒通知方式。在select/poll中，進程只有在調用必定的方法後，內核纔對全部監視的文件描述符進行掃描，而epoll事先經過epoll_ctl()來註冊一個文件描述符，一旦基於某個文件描述符就緒時，內核會採用相似callback的回調機制，迅速激活這個文件描述符，當進程調用epoll_wait()時便獲得通知。

 

epoll既然是對select和poll的改進，就應該能避免上述的三個缺點。那epoll都是怎麼解決的呢？在此以前，咱們先看一下epoll 和select和poll的調用接口上的不一樣，select和poll都只提供了一個函數——select或者poll函數。而epoll提供了三個函 數，epoll_create,epoll_ctl和epoll_wait，epoll_create是建立一個epoll句柄；epoll_ctl是注 冊要監聽的事件類型；epoll_wait則是等待事件的產生。

　　對於第一個缺點，epoll的解決方案在epoll_ctl函數中。每次註冊新的事件到epoll句柄中時（在epoll_ctl中指定 EPOLL_CTL_ADD），會把全部的fd拷貝進內核，而不是在epoll_wait的時候重複拷貝。epoll保證了每一個fd在整個過程當中只會拷貝 一次。

　　對於第二個缺點，epoll的解決方案不像select或poll同樣每次都把current輪流加入fd對應的設備等待隊列中，而只在 epoll_ctl時把current掛一遍（這一遍必不可少）併爲每一個fd指定一個回調函數，當設備就緒，喚醒等待隊列上的等待者時，就會調用這個回調 函數，而這個回調函數會把就緒的fd加入一個就緒鏈表）。epoll_wait的工做實際上就是在這個就緒鏈表中查看有沒有就緒的fd（利用 schedule_timeout()實現睡一會，判斷一會的效果，和select實現中的第7步是相似的）。

　　對於第三個缺點，epoll沒有這個限制，它所支持的FD上限是最大能夠打開文件的數目，這個數字通常遠大於2048,舉個例子, 在1GB內存的機器上大約是10萬左右，具體數目能夠cat /proc/sys/fs/file-max察看,通常來講這個數目和系統內存關係很大。

總結：

（1）select，poll實現須要本身不斷輪詢全部fd集合，直到設備就緒，期間可能要睡眠和喚醒屢次交替。而epoll其實也須要調用 epoll_wait不斷輪詢就緒鏈表，期間也可能屢次睡眠和喚醒交替，可是它是設備就緒時，調用回調函數，把就緒fd放入就緒鏈表中，並喚醒在 epoll_wait中進入睡眠的進程。雖然都要睡眠和交替，可是select和poll在「醒着」的時候要遍歷整個fd集合，而epoll在「醒着」的 時候只要判斷一下就緒鏈表是否爲空就好了，這節省了大量的CPU時間，這就是回調機制帶來的性能提高。

（2）select，poll每次調用都要把fd集合從用戶態往內核態拷貝一次，而且要把current往設備等待隊列中掛一次，而epoll只要 一次拷貝，並且把current往等待隊列上掛也只掛一次（在epoll_wait的開始，注意這裏的等待隊列並非設備等待隊列，只是一個epoll內 部定義的等待隊列），這也能節省很多的開銷。

select,poll,epoll

這三種IO多路複用模型在不一樣的平臺有着不一樣的支持，而epoll在windows下就不支持，好在咱們有selectors模塊，幫咱們默認選擇當前平臺下最合適的

#服務端
from socket import *
import selectors

sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr=server_fileobj.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()



server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind(('127.0.0.1',8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #設置socket的接口爲非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #至關於網select的讀列表裏append了一個文件句柄server_fileobj,而且綁定了一個回調函數accept

while True:
    events=sel.select() #檢測全部的fileobj，是否有完成wait data的
    for sel_obj,mask in events:
        callback=sel_obj.data #callback=accpet
        callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)

#客戶端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8088))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))