協程和異步io

一. 並發、並行、同步、異步、阻塞、非阻塞

　　1.並發：是指一個時間段中有幾個程序都處於已啟動運行到運行完畢之間，且這幾個程序都是在同一個處理機（CPU）上運行，但任一個時刻點上只有一個程序在處理機上運行。

　　2.並行：是指任何時間點，有多個程序運行在多個CPU上（最多和CPU數量一致）。

　　3.並發和並行的區別：

　　　　並發和並行是即相似又有區別的兩個概念，並行是指兩個或者多個事件在同一時刻發生；而並發是指兩個或多個事件在同一時間間隔內發生。在多道程序環境下，並發性是指在一段時間內宏觀上有多個程序在同時運行，但在單處理機系統中，每一時刻卻僅能有一道程序執行，故微觀上這些程序只能是分時地交替執行。倘若在計算機系統中有多個處理機，則這些可以並發執行的程序便可被分配到多個處理機上，實現並行執行，即利用每個處理機來處理一個可並發執行的程序，這樣，多個程序便可以同時執行。

　　4.同步：是指代碼調用IO操作時，必須等待IO操作完成才能返回的調用方式。

　　5.異步：是指代碼調用IO操作時，不必等待IO操作完成就能返回的調用方式。

　　6.阻塞：是指調用函數的時候當前線程被掛起。

　　7.非阻塞：是指調用函數的時候當前線程不會被掛起，而是立即返回。

二. C10K問題和io多路復用(select、poll、epoll)

　　1.C10K問題：

　　　　謂c10k問題,指的是服務器同時支持成千上萬個客戶端的問題,也就是concurrent 10 000 connection(這也是c10k這個名字的由來)。由於硬件成本的大幅度降低和硬件技術的進步，如果一台服務器同時能夠服務更多的客戶端，那么也就意味着服務每一個客戶端的成本大幅度降低，從這個角度來看，問題顯得非常有意義。

　　2.五種I/O模型（詳情：https://www.cnblogs.com/findumars/p/6361627.html）：

　　　　

 

　　　　5.1阻塞I式/O：系統調用不會立即返回結果，當前線程會阻塞，等到獲得結果或報錯時在返回（問題：如在調用send()的同時，線程將被阻塞，在此期間，線程將無法執行任何運算或響應任何的網絡請求。）

 

　　　　5.2非阻塞式I/O：調用后立即返回結果（問題：不一定三次握手成功，recv() 會被循環調用，循環調用recv()將大幅度推高CPU 占用率），做計算任務或者再次發起其他連接就較有優勢

 

 

　　　　5.3I/O復用：它的基本原理就是select/epoll這個function會不斷的輪詢所負責的所有socket，當某個socket有數據到達了，就通知用戶進程。（阻塞式的方法，可以監聽多個socket狀態）（問題：將數據從內核復制到用戶空間的時間不能省）

 

　　　　5.4信號驅動式I/O：運用較少

 

　　　　5.5異步I/O：它就像是用戶進程將整個IO操作交給了他人（kernel）完成，然后他人做完后發信號通知。在此期間，用戶進程不需要去檢查IO操作的狀態，也不需要主動的去拷貝數據。

　　3.解決方法（參照：https://blog.csdn.net/wangtaomtk/article/details/51811011）：

　　　　3.1每個線程/進程處理一個連接：

　　　　　　但是由於申請進程/線程會占用相當可觀的系統資源，同時對於多進程/線程的管理會對系統造成壓力，因此這種方案不具備良好的可擴展性。因此，這一思路在服務器資源還沒有富裕到足夠程度的時候，是不可行的；即便資源足夠富裕，效率也不夠高。

　　　　　　問題：資源占用過多，可擴展性差。

　　　　3.2每個進程/線程同時處理多個連接（IO多路復用）：

　　　　　　3.2.1傳統思路

　　　　　　　　最簡單的方法是循環挨個處理各個連接，每個連接對應一個 socket，當所有 socket 都有數據的時候，這種方法是可行的。但是當應用讀取某個 socket 的文件數據不 ready 的時候，整個應用會阻塞在這里等待該文件句柄，即使別的文件句柄 ready，也無法往下處理。　　　　

　　　　　　　　思路：直接循環處理多個連接。問題：任一文件句柄的不成功會阻塞住整個應用。

　　　　　　3.2.2select：

 

　　　　　　　

　　　　　　　　思路：有連接請求抵達了再檢查處理。

　　　　　　　　問題：句柄上限+重復初始化+逐個排查所有文件句柄狀態效率不高。

　　　　　　3.2.3poll

 

　　　　　　　　思路：設計新的數據結構提供使用效率。

　　　　　　　　問題：逐個排查所有文件句柄狀態效率不高。

　　　　　　3.2.4epoll（nginx使用的是epoll）

 

　　　　　　　　思路：只返回狀態變化的文件句柄。

　　　　　　　　問題：依賴特定平台（Linux）。

　　　　注：epoll不一定比select好（在高並發的情況下，連接活躍度不是很高，epoll比select好；在並發性不高，同時連接很活躍select比epoll好（游戲））

三. epoll+回調+事件循環方式url

　　1. 通過非阻塞I/O實現http請求：

import socket
from urllib.parse import urlparse

def get_url(url):
    #通過socket請求html
    url=urlparse(url)
    host=url.netloc
    path=url.path
    if path=="":
        path="/"
    #建立socket連接
    client=socket.socket(socket.AF_INET,socket.SOCK_STREAM)
    #設置成非阻塞(拋異常：BlockingIOError: [WinError 10035] 無法立即完成一個非阻止性套接字操作。)
    client.setblocking(False)
    try:
        client.connect((host,80))
    except BlockingIOError as e:
        pass
    #向服務器發送數據(還未連接會拋異常)
    while True:
        try:
            client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(path, host).encode("utf8"))
            break
        except OSError as e:
            pass
    #將數據讀取完
    data=b""
    while True:
        try:
            d=client.recv(1024)
        except BlockingIOError as e:
            continue
        if d:
            data+=d
        else:
            break
    #會將header信息作為返回字符串
    data=data.decode('utf8')
    print(data.split('\r\n\r\n')[1])
    client.close()

if __name__=='__main__':
    get_url('http://www.baidu.com')

　　2.使用select完成http請求（循環回調）：

　　　　優點：並發性高（驅動整個程序主要是回調循環loop()，不會等待，請求操作系統有什么准備好了，准備好了就執行【沒有線程切換等，只有一個線程，當一個url連接建立完成后就會注冊，然后回調執行】，省去了線程切換和內存）

#自動根據環境選擇poll和epoll
from selectors import DefaultSelector,EVENT_READ,EVENT_WRITE
selector=DefaultSelector()
urls=[]
#全局變量
stop=False
class Fetcher:
    def connected(self, key):
        #取消注冊
        selector.unregister(key.fd)
        self.client.send("GET {} HTTP/1.1\r\nHost:{}\r\nConnection:close\r\n\r\n".format(self.path, self.host).encode("utf8"))
        selector.register(self.client.fileno(),EVENT_READ,self.readable)

    def readable(self,key):
        d = self.client.recv(1024)
        if d:
            self.data += d
        else:
            selector.unregister(key.fd)
            # 會將header信息作為返回字符串
            data = self.data.decode('utf8')
            print(data.split('\r\n\r\n')[1])
            self.client.close()
            urls.remove(self.spider_url)
            if not urls:
                global stop
                stop=True

    def get_url(self,url):
        self.spider_url = url
        url = urlparse(url)
        self.host = url.netloc
        self.path = url.path
        self.data = b""
        if self.path == "":
            self.path = "/"
        # 建立socket連接
        self.client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.client.setblocking(False)
        try:
            self.client.connect((self.host, 80))
        except BlockingIOError as e:
            pass

        #注冊寫事件,及回調函數
        selector.register(self.client.fileno(),EVENT_WRITE,self.connected)

def loop():
    #回調+事件循環+select（poll/epoll）
    #事件循環，不停的調用socket的狀態並調用對應的回調函數
    #判斷哪個可讀可寫，select本身不支持register模式
    #socket狀態變化后的回調使用程序員完成的
    if not stop:
        while True:
            ready=selector.select()
            for key,mask in ready:
                call_back=key.data
                call_back(key)


if __name__=='__main__':
    fetcher=Fetcher()
    fetcher.get_url('http://www.baidu.com')
    loop()

四. 回調之痛

　　1.可讀性差；2.共享狀態管理困難；3.異常處理困難

　　

協程能解決　　　　　　　　　　　　

五. C10M問題和協程

　　1.C10M問題：

　　　　如何利用8核心CPU，64G內存，在10gps的網絡上保持1000萬的並發連接。

　　2.協程：

　　　　2.1問題：回調模式編碼復雜度高；同步編程的並發性不高；多線程需要線程間同步，look會降低性能

　　　　2.2解決：

　　　　　　采用同步的方式去編寫異步的代碼；

　　　　　　采用單線程去解決任務：線程是由操作系統切換，單線程切換意味着需要我們自己去調度任務；不在需要鎖，並發性高，如果單線程內切換函數，性能遠高於線程切換，並發性更高。

　　　　2.3協程：

　　　　　　傳統函數調用 過程 A->B->C；

　　　　　　我們需要一個可以暫停的函數，並且可以在適當的時候恢復該函數的繼續執行；

　　　　　出現了協程 -> 有多個入口的函數， 可以暫停的函數， 可以暫停的函數(可以向暫停的地方傳入值)；

def get_url(url):
    #do someting 1
    html = get_html(url) #此處暫停，切換到另一個函數去執行
    # #parse html
    urls = parse_url(html)

def get_url(url):
    #do someting 1
    html = get_html(url) #此處暫停，切換到另一個函數去執行
    # #parse html
    urls = parse_url(html)

六. 生成器的send和yield from

　　1.生成器send和next方法：

　　　　啟動生成器方式有兩種：1.next()；2.send()；

　　　　生成器可以產出值；也可以接收值（調用方傳遞進來的值）；

　　　　send方法可以傳遞值進入生成器內部，同時還可以重啟生成器執行到下一個yield的位置（注：在調用send()發送非none之前，我們必須啟動一次生成器，否則會拋錯，方式有兩種gen.send(None)或者next(gen)）

　　2.close()方法：（關閉生成器）

　　　　自己處理的話會拋異常，gen.close()，RuntimeError: generator ignored GeneratorExit，如果是except Exception就不會拋異常，GeneratorExit是繼承至BaseException的，Exception也是繼承於BaseException的

def gen_func():
    #自己處理的話會拋異常，gen.close()，RuntimeError: generator ignored GeneratorExit
    try:
        yield 'https://www.baidu.com'
    #如果是except Exception就不會拋異常，GeneratorExit是繼承至BaseException的，Exception也是繼承於BaseException的
    except GeneratorExit as e:
        pass
    yield 1
    yield 2
    return 'LYQ'

if __name__=='__main__':
    #拋異常StopIteration：
    gen=gen_func()
    print(next(gen))
    gen.close()
    print(next(gen))

　　3.throw()方法：向生成器中扔異常，需要自己處理，否則會拋錯

def gen_func():
    try:
        yield 'https://www.baidu.com'
    except Exception:
        pass
    yield 1
    yield 2
    return 'LYQ'

if __name__=='__main__':
    #拋異常StopIteration：
    gen=gen_func()
    print(next(gen))
    #扔一個異常，是第一句的異常
    gen.throw(Exception,'download error')
    print(next(gen))
    #扔一個異常，是第二句的異常
    gen.throw(Exception,'download error')
    print(next(gen))

 　　4.yield from：（Python 3.3新加的語法）

　　　　4.1簡介：

from itertools import chain
my_list=[1,2,3]
my_dict={'name1':'LYQ1',
         'name2':'LYQ2'}
#將所有值遍歷輸出
# for value in chain(my_list,my_dict,range(5,10)):
#     print(value)

def g1(iterable):
    yield range(10)
#yield from iterable
def my_chain(*args,**kwargs):
    for my_iterable in args:
        #功能非常多
        yield from my_iterable
        # for value in my_iterable:
        #     yield value
for value in my_chain(my_list,my_dict,range(5,10)):
    print(value)

 

　　　  4.2main調用方 g1:委托生成器 gen:子生成器:

def g1(gen):
    yield from gen
gen=range(10)
def main():
    g=g1(gen)
    #直接發送給子生成器
    print(g.send(None))
#main:調用方 g1:委托生成器 gen:子生成器
#yield from會在調用方與子生成器之間建立一個雙向通道
main()

 

　　　  4.3例子：

final_result = {}


def sales_sum(pro_name):
    total = 0
    nums = []
    while True:
        x = yield
        print(pro_name + "銷量: ", x)
        if not x:
            break
        total += x
        nums.append(x)
    #直接返回到yield from sales_sum(key)
    return total, nums


def middle(key):
    while True:
        final_result[key] = yield from sales_sum(key)
        print(key + "銷量統計完成！！.")


def main():
    data_sets = {
        "面膜": [1200, 1500, 3000],
        "手機": [28, 55, 98, 108],
        "大衣": [280, 560, 778, 70],
    }
    for key, data_set in data_sets.items():
        print("start key:", key)
        m = middle(key)
        #直接send到子生成器里面（x = yield）
        m.send(None)  # 預激middle協程
        for value in data_set:
            m.send(value)  # 給協程傳遞每一組的值
        m.send(None)
    print("final_result:", final_result)


if __name__ == '__main__':
    main()

 

　　　　　　無yield from：

def sales_sum(pro_name):
    total = 0
    nums = []
    while True:
        x = yield
        print(pro_name + "銷量: ", x)
        if not x:
            break
        total += x
        nums.append(x)
    #直接返回到yield from sales_sum(key)
    return total, nums

if __name__ == "__main__":
    #直接與子生成器通信（沒用yield from就需要捕獲異常）
    my_gen = sales_sum("手機")
    my_gen.send(None)
    my_gen.send(1200)
    my_gen.send(1500)
    my_gen.send(3000)
    try:
        my_gen.send(None)
    #獲取返回值
    except StopIteration as e:
        result = e.value
        print(result)

 

　　4.4介紹yield from詳情：

#pep380

#1. RESULT = yield from EXPR可以簡化成下面這樣
#一些說明
"""
_i：子生成器，同時也是一個迭代器
_y：子生成器生產的值
_r：yield from 表達式最終的值
_s：調用方通過send()發送的值
_e：異常對象

"""

_i = iter(EXPR)      # EXPR是一個可迭代對象，_i其實是子生成器；
try:
    _y = next(_i)   # 預激子生成器，把產出的第一個值存在_y中；
except StopIteration as _e:
    _r = _e.value   # 如果拋出了`StopIteration`異常，那么就將異常對象的`value`屬性保存到_r，這是最簡單的情況的返回值；
else:
    while 1:    # 嘗試執行這個循環，委托生成器會阻塞；
        _s = yield _y   # 生產子生成器的值，等待調用方`send()`值，發送過來的值將保存在_s中；
        try:
            _y = _i.send(_s)    # 轉發_s，並且嘗試向下執行；
        except StopIteration as _e:
            _r = _e.value       # 如果子生成器拋出異常，那么就獲取異常對象的`value`屬性存到_r，退出循環，恢復委托生成器的運行；
            break
RESULT = _r     # _r就是整個yield from表達式返回的值。

"""
1. 子生成器可能只是一個迭代器，並不是一個作為協程的生成器，所以它不支持.throw()和.close()方法；
2. 如果子生成器支持.throw()和.close()方法，但是在子生成器內部，這兩個方法都會拋出異常；
3. 調用方讓子生成器自己拋出異常
4. 當調用方使用next()或者.send(None)時，都要在子生成器上調用next()函數，當調用方使用.send()發送非 None 值時，才調用子生成器的.send()方法；
"""
_i = iter(EXPR)
try:
    _y = next(_i)
except StopIteration as _e:
    _r = _e.value
else:
    while 1:
        try:
            _s = yield _y
        except GeneratorExit as _e:
            try:
                _m = _i.close
            except AttributeError:
                pass
            else:
                _m()
            raise _e
        except BaseException as _e:
            _x = sys.exc_info()
            try:
                _m = _i.throw
            except AttributeError:
                raise _e
            else:
                try:
                    _y = _m(*_x)
                except StopIteration as _e:
                    _r = _e.value
                    break
        else:
            try:
                if _s is None:
                    _y = next(_i)
                else:
                    _y = _i.send(_s)
            except StopIteration as _e:
                _r = _e.value
                break
RESULT = _r

 

　　　　看完代碼，我們總結一下關鍵點：

　　　　1. 子生成器生產的值，都是直接傳給調用方的；調用方通過.send()發送的值都是直接傳遞給子生成器的；如果發送的是 None，會調用子生成器的__next__()方法，如果不是 None，會調用子生成器的.send()方法；
　　　　2. 子生成器退出的時候，最后的return EXPR，會觸發一個StopIteration(EXPR)異常；
　　　　3. yield from表達式的值，是子生成器終止時，傳遞給StopIteration異常的第一個參數；
　　　　4. 如果調用的時候出現StopIteration異常，委托生成器會恢復運行，同時其他的異常會向上 "冒泡"；
　　　　5. 傳入委托生成器的異常里，除了GeneratorExit之外，其他的所有異常全部傳遞給子生成器的.throw()方法；如果調用.throw()的時候出現了StopIteration異常，那么就恢復委托生成器的運行，其他的異常全部向上 "冒泡"；
　　　　6. 如果在委托生成器上調用.close()或傳入GeneratorExit異常，會調用子生成器的.close()方法，沒有的話就不調用。如果在調用.close()的時候拋出了異常，那么就向上 "冒泡"，否則的話委托生成器會拋出GeneratorExit異常。

七. 生成器如何變成協程？

　　1.生成器可以暫停並獲取狀態：

#生成器是可以暫停的函數
import inspect
def gen():
    yield 1
    return True

if __name__=='__main__':
    g1=gen()
    #獲取生成器狀態 GEN_CREATED(創建)
    print(inspect.getgeneratorstate(g1))
    next(g1)
    #GEN_SUSPENDED暫停
    print(inspect.getgeneratorstate(g1))
    try:
        next(g1)
    except StopIteration:
        pass
    #GEN_CLOSED關閉
    print(inspect.getgeneratorstate(g1))

 

　　2.協程的調度依然是 事件循環+協程模式 ，協程是單線程模式：

#生成器是可以暫停的函數
import inspect
# def gen_func():
#     value=yield from
#     #第一返回值給調用方， 第二調用方通過send方式返回值給gen
#     return "bobby"
#1. 用同步的方式編寫異步的代碼， 在適當的時候暫停函數並在適當的時候啟動函數
import socket
def get_socket_data():
    yield 1

def downloader(url):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.setblocking(False)

    try:
        client.connect((host, 80))  # 阻塞不會消耗cpu
    except BlockingIOError as e:
        pass

    selector.register(self.client.fileno(), EVENT_WRITE, self.connected)
    #如果get_socket_data()中出現異常，會直接拋給downloader（向上拋）
    source = yield from get_socket_data()
    data = source.decode("utf8")
    html_data = data.split("\r\n\r\n")[1]
    print(html_data)

def download_html(html):
    html = yield from downloader()

if __name__ == "__main__":
    #協程的調度依然是 事件循環+協程模式 ，協程是單線程模式
    pass

八. async和await原生協程

　　1.python為了將語義變得更加明確，就引入了async和await關鍵字定義原生的協程：

　　　　生成器實現的協程又可以當生成器，又可以當協程，且代碼凌亂，不利於后期維護。原生的協程中不可以yield，否則會拋錯（讓協程更加明確）

可異步調用：實際實現了__await__魔法函數

　　　　await：將控制權交出去並等待結果返回,await只能接收awaitable對象，可以理解成yield from

# from collections import Awaitable
#如果是函數,就要使用coroutine裝飾器，實際將__await_指向___iter__
# import types
# @types.coroutine
# def downloader(url):
#     return "haha"

async def downloader(url):
    return "haha"
async def download_url(url):
    #將控制權交出去並等待結果返回,await只能接收awaitable對象，可以理解成yield from
    html=await downloader(url)
    return html

if __name__=='__main__':
    coro=download_url('www.baidu.com')
    #原生協程不能調用next
    coro.send(None)