1. 程式人生 > >基於協程的Python網絡庫gevent

基於協程的Python網絡庫gevent

imp false n! eache 通過 建議 ces 喚醒 code

import gevent
 
def test1():
    print 12
    gevent.sleep(0)
    print 34
 
def test2():
    print 56
    gevent.sleep(0)
    print 78
 
gevent.joinall([
    gevent.spawn(test1),
    gevent.spawn(test2),
])

解釋下,”gevent.spawn()”方法會創建一個新的greenlet協程對象,並運行它。”gevent.joinall()”方法會等待所有傳入的greenlet協程運行結束後再退出,這個方法可以接受一個”timeout”參數來設置超時時間,單位是秒。運行上面的程序,執行順序如下:

  1. 先進入協程test1,打印12
  2. 遇到”gevent.sleep(0)”時,test1被阻塞,自動切換到協程test2,打印56
  3. 之後test2被阻塞,這時test1阻塞已結束,自動切換回test1,打印34
  4. 當test1運行完畢返回後,此時test2阻塞已結束,再自動切換回test2,打印78
  5. 所有協程執行完畢,程序退出

所以,程序運行下來的輸出就是:

12
56
34
78

greenlet一個協程運行完後,必須顯式切換,不然會返回其父協程。而在gevent中,一個協程運行完後,它會自動調度那些未完成的協程。

import gevent
import socket
 
urls = [
www.baidu.com, www.gevent.org, www.python.org] jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls] gevent.joinall(jobs, timeout=5) print [job.value for job in jobs]

我們通過協程分別獲取三個網站的IP地址,由於打開遠程地址會引起IO阻塞,所以gevent會自動調度不同的協程。另外,我們可以通過協程對象的”value”屬性,來獲取協程函數的返回值。

猴子補丁 Monkey patching

其實上面程序運行的時間同不用協程是一樣的,是三個網站打開時間的總和。可是理論上協程是非阻塞的,那運行時間應該等於最長的那個網站打開時間呀?其實這是因為Python標準庫裏的socket是阻塞式的,DNS解析無法並發,包括像urllib庫也一樣,所以這種情況下用協程完全沒意義。那怎麽辦?

一種方法是使用gevent下的socket模塊,我們可以通過”from gevent import socket”來導入。不過更常用的方法是使用猴子布丁(Monkey patching):

from gevent import monkey; monkey.patch_socket()
import gevent
import socket
 
urls = [www.baidu.com, www.gevent.org, www.python.org]
jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls]
gevent.joinall(jobs, timeout=5)
 
print [job.value for job in jobs]

上述代碼的第一行就是對socket標準庫打上猴子補丁,此後socket標準庫中的類和方法都會被替換成非阻塞式的,所有其他的代碼都不用修改,這樣協程的效率就真正體現出來了。Python中其它標準庫也存在阻塞的情況,gevent提供了”monkey.patch_all()”方法將所有標準庫都替換。

from gevent import monkey; monkey.patch_all()

使用猴子補丁褒貶不一,但是官網上還是建議使用”patch_all()”,而且在程序的第一行就執行。

獲取協程狀態

協程狀態有已啟動和已停止,分別可以用協程對象的”started”屬性和”ready()”方法來判斷。對於已停止的協程,可以用”successful()”方法來判斷其是否成功運行且沒拋異常。如果協程執行完有返回值,可以通過”value”屬性來獲取。另外,greenlet協程運行過程中發生的異常是不會被拋出到協程外的,因此需要用協程對象的”exception”屬性來獲取協程中的異常。下面的例子很好的演示了各種方法和屬性的使用。

#coding:utf8
import gevent
 
def win():
    return You win!
 
def fail():
    raise Exception(You failed!)
 
winner = gevent.spawn(win)
loser = gevent.spawn(fail)
 
print winner.started # True
print loser.started  # True
 
# 在Greenlet中發生的異常,不會被拋到Greenlet外面。
# 控制臺會打出Stacktrace,但程序不會停止
try:
    gevent.joinall([winner, loser])
except Exception as e:
    # 這段永遠不會被執行
    print This will never be reached
 
print winner.ready() # True
print loser.ready()  # True
 
print winner.value # ‘You win!‘
print loser.value  # None
 
print winner.successful() # True
print loser.successful()  # False
 
# 這裏可以通過raise loser.exception 或 loser.get()
# 來將協程中的異常拋出
print loser.exception

協程運行超時

在”gevent.joinall()”方法中可以傳入timeout參數來設置超時,我們也可以在全局範圍內設置超時時間:

import gevent
from gevent import Timeout
 
timeout = Timeout(2)  # 2 seconds
timeout.start()
 
def wait():
    gevent.sleep(10)
 
try:
    gevent.spawn(wait).join()
except Timeout:
    print(Could not complete)

上例中,我們將超時設為2秒,此後所有協程的運行,如果超過兩秒就會拋出”Timeout”異常。我們也可以將超時設置在with語句內,這樣該設置只在with語句塊中有效:

with Timeout(1):
    gevent.sleep(10)

此外,我們可以指定超時所拋出的異常,來替換默認的”Timeout”異常。比如下例中超時就會拋出我們自定義的”TooLong”異常。

class TooLong(Exception):
    pass
 
with Timeout(1, TooLong):
    gevent.sleep(10)

協程間通訊

greenlet協程間的異步通訊可以使用事件(Event)對象。該對象的”wait()”方法可以阻塞當前協程,而”set()”方法可以喚醒之前阻塞的協程。在下面的例子中,5個waiter協程都會等待事件evt,當setter協程在3秒後設置evt事件,所有的waiter協程即被喚醒。

#coding:utf8
import gevent
from gevent.event import Event
 
evt = Event()
 
def setter():
    print Wait for me
    gevent.sleep(3)  # 3秒後喚醒所有在evt上等待的協程
    print "Ok, I‘m done"
    evt.set()  # 喚醒
 
def waiter():
    print "I‘ll wait for you"
    evt.wait()  # 等待
    print Finish waiting
 
gevent.joinall([
    gevent.spawn(setter),
    gevent.spawn(waiter),
    gevent.spawn(waiter),
    gevent.spawn(waiter),
    gevent.spawn(waiter),
    gevent.spawn(waiter)
])

除了Event事件外,gevent還提供了AsyncResult事件,它可以在喚醒時傳遞消息。讓我們將上例中的setter和waiter作如下改動:

from gevent.event import AsyncResult
aevt = AsyncResult()
 
def setter():
    print Wait for me
    gevent.sleep(3)  # 3秒後喚醒所有在evt上等待的協程
    print "Ok, I‘m done"
    aevt.set(Hello!)  # 喚醒,並傳遞消息
 
def waiter():
    print("I‘ll wait for you")
    message = aevt.get()  # 等待,並在喚醒時獲取消息
    print Got wake up message: %s % message

隊列 Queue

gevent的隊列對象可以讓greenlet協程之間安全的訪問。運行下面的程序,你會看到3個消費者會分別消費隊列中的產品,且消費過的產品不會被另一個消費者再取到:

import gevent
from gevent.queue import Queue
 
products = Queue()
 
def consumer(name):
    while not products.empty():
        print %s got product %s % (name, products.get())
        gevent.sleep(0)
 
    print %s Quit
 
def producer():
    for i in xrange(1, 10):
        products.put(i)
 
gevent.joinall([
    gevent.spawn(producer),
    gevent.spawn(consumer, steve),
    gevent.spawn(consumer, john),
    gevent.spawn(consumer, nancy),
])

put和get方法都是阻塞式的,它們都有非阻塞的版本:put_nowait和get_nowait。如果調用get方法時隊列為空,則拋出”gevent.queue.Empty”異常。‘

信號量

信號量可以用來限制協程並發的個數。它有兩個方法,acquire和release。顧名思義,acquire就是獲取信號量,而release就是釋放。當所有信號量都已被獲取,那剩余的協程就只能等待任一協程釋放信號量後才能得以運行:

import gevent
from gevent.coros import BoundedSemaphore
 
sem = BoundedSemaphore(2)
 
def worker(n):
    sem.acquire()
    print(Worker %i acquired semaphore % n)
    gevent.sleep(0)
    sem.release()
    print(Worker %i released semaphore % n)
 
gevent.joinall([gevent.spawn(worker, i) for i in xrange(0, 6)])

上面的例子中,我們初始化了”BoundedSemaphore”信號量,並將其個數定為2。所以同一個時間,只能有兩個worker協程被調度。程序運行後的結果如下:

Worker 0 acquired semaphore
Worker 1 acquired semaphore
Worker 0 released semaphore
Worker 1 released semaphore
Worker 2 acquired semaphore
Worker 3 acquired semaphore
Worker 2 released semaphore
Worker 3 released semaphore
Worker 4 acquired semaphore
Worker 4 released semaphore
Worker 5 acquired semaphore
Worker 5 released semaphore

協程本地變量

同線程類似,協程也有本地變量,也就是只在當前協程內可被訪問的變量:

import gevent
from gevent.local import local
 
data = local()
 
def f1():
    data.x = 1
    print data.x
 
def f2():
    try:
        print data.x
    except AttributeError:
        print x is not visible
 
gevent.joinall([
    gevent.spawn(f1),
    gevent.spawn(f2)
])

通過將變量存放在local對象中,即可將其的作用域限制在當前協程內,當其他協程要訪問該變量時,就會拋出異常。不同協程間可以有重名的本地變量,而且互相不影響。因為協程本地變量的實現,就是將其存放在以的”greenlet.getcurrent()”的返回為鍵值的私有的命名空間內。

實際應用

基於Flask聊天室

https://github.com/sdiehl/minichat/blob/master/app.py

基於協程的Python網絡庫gevent