40 協程 i/0多路複用

時間 2019-12-18

標籤多路简体版

原文原文鏈接

主要內容:python

1. yeild 實現狀態保存linux

import time
def func():
    sum = 0
    yield sum
    sum = 0
    yield sum
    sum = 0
    yield sum
def fff():
    g = func()      # 得到一個生成器函數, 並不會執行函數
    print('這是在fff函數中')
    print(next(g))  # 執行
    time.sleep(1)
    print('這是在fff函數中')
    print(next(g))
    time.sleep(1)
    print('這是在fff函數中')
    print(next(g))
fff()

2 . yield 實現併發的假象程序員

　　在單線程中, 若是存在多個函數, 若是某個函數存在i/o操做, 想讓程序立刻切換到另外一個函數去執行,以此實現一個假的並發現象.windows

　　總結 : yield 只能實現單純的切換函數和保存函數狀態的功能服務器

　　　不能實現: 當某一個函數遇到i/o操做時, 自動的切換到另外一個函數去執行, 若是能實現這個功能, 那麼每一個函數都是一個協程.網絡

　　　可是協程的本質仍是主要依靠yield實現的多線程

　　若是隻是拿yield去單純的實現一個切換的現象, 根本沒有程序串行執行效率高.併發

def consumer():
    while 1:
        x = yield
        # print(x)
def producer():
    g = consumer()
    next(g)
    for i in range(100000000):
        g.send(i)
start = time.time()
producer()
print('yield:',time.time() - start)

　　串行代碼:app

def consumer(l):
    # for i in l:
        # print(i)
    pass
def producer():
    l = []
    for i in range(100000000):
        l.append(i)
    return l
start = time.time()
l = producer()
consumer(l)
print(time.time() - start)

3 . 協程框架

　　a : 協程的定義 : 是一個比線程更加輕量級的單位, 是組成線程的各個函數,(單線程下的併發, 又稱)

　　b : 爲何要有協程　由於想要在單線程內實現併發的效果(由於cpython有GIL鎖,限制了在同一個時間點,只能執行一個線程. 因此想要在執行一個線程的期間, 充分利用cpu的性能, 因此想在單線程內實現併發的效果)

　　c : 併發 : 切換 + 保存狀態

　　d : cpu爲何要切換: 1.由於某個程序阻塞了. 2. 由於某個程序用完了時間片(該任務計算的時間過長)

　　e : 目標 : 因此想要實現單線程的併發, 就要解決在單線程內,多個任務函數中,某個任務函數遇到i/o操做, 立刻自動切換到其餘任務函數中去執行

協程是用戶本身去調度的

3 . greenlet模塊

　　a :定義能簡單的實現函數與函數的切換, 可是遇到i/o操做, 不能自動切換.

　　b :模塊的使用 : 該模塊是一個類, switch是類中的一個方法.

　　　　　　　　　註冊一下函數func, 將函數註冊成一個對象f1 f1 = greenlet(func)

　　　　　　　　　調用func, 使用f1.switch(), 若是函數須要傳參, 就在switch這裏傳參便可.

from greenlet import greenlet
import time
def eat(name):
    print('%s吃炸雞' % name)
    time.sleep(1)
    f2.switch('lili')
    print('%s吃雪糕' % name)
    f2.switch()
def drink(name):
    print('%s喝啤酒' % name)
    time.sleep(1)
    f1.switch()
    print('%s喝可樂' % name)
f1 = greenlet(eat)
f2 = greenlet(drink)
f1.switch('麗麗')

4. gevent 模塊

　　a : 定義能夠實如今某函數內部遇到io操做,就自動的切換到其餘函數內部去執行

　　b : 模塊的使用 : g = gevent.spawn(func，參數) 註冊一下函數func，返回一個對象g

　　　　　　　　　gevent.join(g) #等待g指向的函數func執行完畢，若是在執行過程當中，遇到IO，就切換

gevent.joinall([g1,g2,g3])#等待g1 g2 g3指向的函數func執行完畢

import gevent
def func1():
    print(1)
    gevent.sleep(0.5)
    print(2)

def func2():
    print(3)
    gevent.sleep(0.5)         #gevent不能識別其餘的io操做,只能識別本身的
    print(4)
g = gevent.spawn(func1)
g2 = gevent.spawn(func2)
g.join()
g2.join()

　　　　解決gevent不能識別其餘io操做的問題

import gevent
from gevent import monkey
monkey.patch_all()# 可讓gevent識別大部分經常使用的IO操做
import time
def func1():
    print('1 2 3 4')
    time.sleep(1)
    print('3 2 3 4')
    # gevent.sleep(1)

def func2():
    print('2 2 3 4')
    time.sleep(1)
    print('再來一次')
g1 = gevent.spawn(func1)
g2 = gevent.spawn(func2)
g1.join()# 等待g1指向的任務執行結束
g2.join()

　　串行與併發的效率對比:

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

def func(num):
    time.sleep(1)
    print(num)
start = time.time()
for i in range(10):
    func(i)
print('時間', time.time() - start)        #10s
if __name__ == '__main__':
    li = []
    start = time.time()
    for i in range(10):
        g = gevent.spawn(func, i)
        li.append(g)
    gevent.joinall(li)      # 等待g指向的函數執行完畢.
    print('時間',time.time() - start)   #1s

　　爬蟲事例 :

from gevent import monkey
import time
import requests
import gevent
def func(url):
    re = requests.get(url)
    print(url, re.status_code, len(re.text))
url_l = ['http://www.baidu.com',
         'https://www.jd.com',
         'http://www.taobao.com',
         'http://www.qq.com',
         'http://www.mi.com',
         'http://www.cnblogs.com']
def sync_func(url_l):
    for url in url_l:
        func(url)          #串行執行函數

def async_func(url_l):
    li = []
    for url in url_l:
        g = gevent.spawn(func, url)  #使用gevent協程併發去執行任務函數
        #當遇到每一個網頁請求比較大的網絡延遲時,自動切換到其餘的任務函數.
        li.append(g)
    gevent.joinall(li)   #等待g指向的任務函數執行完.

start = time.time()
sync_func(url_l)
print('使用串行消耗的時間爲', time.time() - start)


start = time.time()
async_func(url_l)
print('使用併發消耗的時間爲', time.time() - start)

5 . i/o多路複用

a : 用非阻塞io模型去解決阻塞io

import socket
sk = socket.socket()
sk.setblocking(False)
sk.bind(('127.0.0.1',8080))
sk.listen()

l = []
del_l = []
while 1:
    try:
        conn,addr = sk.accept()# 若是是阻塞IO模型，在這裏程序會一直等待。
        l.append(conn)# 將每一個請求鏈接的客戶端的conn添加到列表中
    except BlockingIOError:
        for conn in l:# 去遍歷全部客戶端的conn，看看有沒有客戶端給我發送數據了

            try:
                info = conn.recv(1024).decode('utf-8')# 嘗試接收，看看有沒有客戶端給我發數據
                if not info:# 若是客戶端正常執行了close，服務器會接收到一個空
                    del_l.append(conn)# 將已經結束的客戶端的conn，添加到要刪除的列表中
                    print('客戶端正常退出了!')
                    conn.close()# 由於客戶端已經主動close，因此服務器端的conn也要close
                else:
                    print(info)
                    conn.send(info.upper().encode('utf-8'))
            except BlockingIOError:
                continue# 是沒有接受到客戶端發來的數據而報錯
            except ConnectionResetError:
                pass# 是由於客戶端強制退出而報錯
        if del_l:
            for conn in del_l:
                l.remove(conn)
            del_l = []# 在刪除完主動關閉的客戶端的鏈接以後，應該把此列表清空，不然報錯

　　基於select的網絡io模型:

import select
import socket

sk = socket.socket()
sk.bind(('127.0.0.1',8080))
sk.listen()
del_l = []
rlist = [sk]# 是用來讓select幫忙監聽的 全部 接口
# select：windows/linux是監聽事件有沒有數據到來
# poll:  linux   也能夠作select的工做
# epoll: linux   也能夠作相似的工做
while 1:
    r,w,x = select.select(rlist,[],[])# 傳參給select，當rlist列表中哪一個接口有反應，就返回給r這個列表
    if r:
        for i in r:# 循環遍歷r，看看有反應的接口究竟是sk  仍是conn
            if i == sk:
                # 若是是sk，那就表示有客戶端的鏈接請求
                '''sk有數據要接收，表明着有客戶端要來鏈接'''
                conn,addr = i.accept()
                rlist.append(conn)# 把新的客戶端的鏈接，添加到rlist，繼續讓select幫忙監聽
            else:
                # 若是是conn，就表示有客戶端給我發數據了
                '''conn有數據要接收，表明要使用recv'''
                try:
                    msg_r = i.recv(1024).decode('utf-8')
                    if not msg_r:
                        '''客戶端執行了close，客戶端主動正常關閉鏈接'''
                        del_l.append(i)
                        i.close()
                    else:
                        print(msg_r)
                        i.send(msg_r.upper().encode('utf-8'))
                except ConnectionResetError:
                    pass
        if del_l:# 刪除那些主動斷開鏈接的客戶端的conn
            for conn in del_l:
                rlist.remove(conn)
            del_l.clear()

　　i/o 多路複用 : 阻塞i/o ; 非阻塞i/o ; 多路複用i/o ; 異步i/o : python實現不了, 可是有tornado框架,天生自帶異步.