Python全棧之路系列----之-----協程(單線程併發)/Greenlet

時間 2019-12-20

標籤 python 之路系列單線併發 greenlet 欄目 Python 简体版

原文原文鏈接

協程

概念

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

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

對於單線程下，咱們不可避免程序中出現io操做，但若是咱們能在本身的程序中（即用戶程序級別，而非操做系統級別）控制單線程下的多個任務能在一個任務遇到io阻塞時就切換到另一個任務去計算github

這樣就保證了該線程可以最大限度地處於就緒態，即隨時均可以被cpu執行的狀態，至關於咱們在用戶程序級別將本身的io操做最大限度地隱藏起來編程

從而能夠迷惑操做系統，讓其看到：該線程好像是一直在計算，io比較少，從而更多的將cpu的執行權限分配給咱們的線程。併發

單線程實現併發即在一個主線程內實現併發;本質是:切換+保存狀態,這種併發只基於io阻塞app

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

2. 能夠檢測io操做，在遇到io操做的狀況下才發生切換

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

#1 yiled能夠保存狀態，yield的狀態保存與操做系統的保存線程狀態很像，可是yield是代碼級別控制的，更輕量級
#2 send能夠把一個函數的結果傳給另一個函數，以此實現單線程內程序之間的切換 



單純地切換反而會下降運行效率
#串行執行
import time
def consumer(res):
    '''任務1:接收數據,處理數據'''
    pass

def producer():
    '''任務2:生產數據'''
    res=[]
    for i in range(10000000):
        res.append(i)
    return res

start=time.time()
#串行執行
res=producer()
consumer(res) #寫成consumer(producer())會下降執行效率
stop=time.time()
print(stop-start) #1.5536692142486572



#基於yield併發執行
import time
def consumer():
    '''任務1:接收數據,處理數據'''
    while True:
        x=yield

def producer():
    '''任務2:生產數據'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)

start=time.time()
#基於yield保存狀態,實現兩個任務直接來回切換,即併發的效果
#PS:若是每一個任務中都加上打印,那麼明顯地看到兩個任務的打印是你一次我一次,即併發執行的.
producer()

stop=time.time()
print(stop-start) #2.0272178649902344

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

import time
def consumer():
    '''任務1:接收數據,處理數據'''
    while True:
        x=yield

def producer():
    '''任務2:生產數據'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
        time.sleep(2)

start=time.time()
producer() #併發執行,可是任務producer遇到io就會阻塞住,並不會切到該線程內的其餘任務去執行

stop=time.time()
print(stop-start)

yield並不能實現遇到io切換

協程特色

協程的優缺點：ide

優勢函數

無需線程上下文切換的開銷
無需原子操做鎖定及同步的開銷(更改一個變量)
方便切換控制流，簡化編程模型
高併發+高擴展性+低成本：一個CPU支持上萬的協程都不是問題。因此很適合用於高併發處理。

缺點：高併發

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

Greenlet

greenlet模塊能夠很是簡單地實現多個任務的切換,可是檢測不到io阻塞,須要手動添加

實現協程實例

def consumer(name): print("--->starting eating baozi...") while True: new_baozi = yield  # 直接返回
        print("[%s] is eating baozi %s" % (name, new_baozi)) 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()

yield

Greenlet

安裝greenlet >>> pip3 install greenlet

from greenlet import greenlet

def func1():
    print(12)
    #遇到switch時切換,手動切換
    gr2.switch()
    print(34)
    gr2.switch()
def func2():
    print(56)
    gr1.switch()
    print(78)

#建立兩個協程
gr1=greenlet(func1)
gr2=greenlet(func2)
gr1.switch()

Greenlet

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

單純的切換

Gevent

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

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

用法

遇到IO阻塞時會自動切換任務

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

gevent

gevent.sleep(2)模擬的是gevent能夠識別的io阻塞,補丁必須放在開頭位置

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

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

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_phone)
gevent.joinall([g1,g2])
print('主')

補丁

#協程：單線程下實現併發，用戶從應用程序級別控制單線程下任務的切換，注意必定是遇到IO才切


# import gevent
# #1.檢測IO
# #2.自動切換
# import time
# 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)
#
# start=time.time()
# g1=gevent.spawn(eat,'alex')
# g2=gevent.spawn(play,'egon')
#
# # g1.join()
# # g2.join()
# gevent.joinall([g1,g2])
# stop=time.time()
# print(stop-start)





# import gevent
# import os
# #1.檢測IO
# #2.自動切換
# import time
# def eat():
#     print('%s eat 1' %os.getpid())
#     gevent.sleep(2)
#     print('%s eat 2' %os.getpid())
# def play():
#     print('%s play 1' %os.getpid())
#     gevent.sleep(1)
#     print('%s play 2' %os.getpid())
#
# start=time.time()
# g1=gevent.spawn(eat,)
# g2=gevent.spawn(play,)
#
# # g1.join()
# # g2.join()
# gevent.joinall([g1,g2])
# stop=time.time()
# print(stop-start)



# import gevent
# import os
# from threading import current_thread
# #1.檢測IO
# #2.自動切換
# import time
# def eat():
#     print('%s eat 1' %current_thread().getName())
#     gevent.sleep(2)
#     print('%s eat 2' %current_thread().getName())
# def play():
#     print('%s play 1' %current_thread().getName())
#     gevent.sleep(1)
#     print('%s play 2' %current_thread().getName())
#
# start=time.time()
# g1=gevent.spawn(eat,)
# g2=gevent.spawn(play,)
#
# # g1.join()
# # g2.join()
# gevent.joinall([g1,g2])
# stop=time.time()
# print(stop-start)


from gevent import monkey;monkey.patch_all()
import gevent
import os
from threading import current_thread
#1.檢測IO
#2.自動切換
import time
def eat():
    print('%s eat 1' %current_thread().getName())
    time.sleep(2)
    print('%s eat 2' %current_thread().getName())
def play():
    print('%s play 1' %current_thread().getName())
    time.sleep(1)
    print('%s play 2' %current_thread().getName())

start=time.time()
g1=gevent.spawn(eat,)
g2=gevent.spawn(play,)

# g1.join()
# g2.join()
gevent.joinall([g1,g2])
stop=time.time()
print(stop-start)

總結性代碼

例子

from urllib import request
from gevent import monkey
import gevent
import time

monkey.patch_all()  # 當前程序中只要設置到IO操做的都作上標記

def wget(url):
    print('GET: %s' % url)
    resp = request.urlopen(url)
    data = resp.read()
    print('%d bytes received from %s.' % (len(data), url))

urls = [
    'https://www.python.org/',
    'https://www.python.org/',
    'https://github.com/',
    'https://blog.ansheng.me/',
]

# 串行抓取
start_time = time.time()
for n in urls:
    wget(n)
print("串行抓取使用時間：", time.time() - start_time)

# 並行抓取
ctrip_time = time.time()
gevent.joinall([
    gevent.spawn(wget, 'https://www.python.org/'),
    gevent.spawn(wget, 'https://www.python.org/'),
    gevent.spawn(wget, 'https://github.com/'),
    gevent.spawn(wget, 'https://blog.ansheng.me/'),
])
print("並行抓取使用時間：", time.time() - ctrip_time)



輸出
GET: https://www.python.org/
47424 bytes received from https://www.python.org/.
GET: https://www.python.org/
47424 bytes received from https://www.python.org/.
GET: https://github.com/
25735 bytes received from https://github.com/.
GET: https://blog.ansheng.me/
82693 bytes received from https://blog.ansheng.me/.
串行抓取使用時間： 15.143015384674072
GET: https://www.python.org/
GET: https://www.python.org/
GET: https://github.com/
GET: https://blog.ansheng.me/
25736 bytes received from https://github.com/.
47424 bytes received from https://www.python.org/.
82693 bytes received from https://blog.ansheng.me/.
47424 bytes received from https://www.python.org/.
並行抓取使用時間： 3.781306266784668

頁面抓取

from gevent import monkey;monkey.patch_all()
import gevent
from multiprocessing import Process
from socket import *

def server(ip,port):
    s = socket(AF_INET, SOCK_STREAM)
    s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
    s.bind((ip,port))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        print('%s:%s' % (addr[0], addr[1]))
        g1=gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    while True:
        try:
            data=conn.recv(1024)
            print('%s:%s [%s]' %(addr[0],addr[1],data))
            if not data:break
            conn.send(data.upper())
        except ConnectionResetError:
            break
    conn.close()
if __name__ == '__main__':
    server('127.0.0.1',8091)

服務端

# from socket import *
# c=socket(AF_INET,SOCK_STREAM)
# c.connect(('127.0.0.1',8090))
#
# while True:
#     msg=input('>>: ').strip()
#     if not msg:continue
#     c.send(msg.encode('utf-8'))
#     data=c.recv(1024)
#     print(data.decode('utf-8'))


from threading import Thread
from socket import *

def client():
    c=socket(AF_INET,SOCK_STREAM)
    c.connect(('127.0.0.1',8091))

    while True:
        c.send('hello'.encode('utf-8'))
        data=c.recv(1024)
        print(data.decode('utf-8'))

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