前述

進(jìn)程 線程 協(xié)程 異步

并發(fā)編程（不是并行）目前有四種方式：多進(jìn)程、多線程、協(xié)程和異步。

• 多進(jìn)程編程在python中有類似C的os.fork,更高層封裝的有multiprocessing標(biāo)準(zhǔn)庫
• 多線程編程python中有Thread和threading
• 異步編程在linux下主+要有三種實(shí)現(xiàn)select，poll，epoll
• 協(xié)程在python中通常會說到y(tǒng)ield，關(guān)于協(xié)程的庫主要有g(shù)reenlet,stackless,gevent,eventlet等實(shí)現(xiàn)。

進(jìn)程

• 不共享任何狀態(tài)
• 調(diào)度由操作系統(tǒng)完成
• 有獨(dú)立的內(nèi)存空間（上下文切換的時(shí)候需要保存棧、cpu寄存器、虛擬內(nèi)存、以及打開的相關(guān)句柄等信息，開銷大）
• 通訊主要通過信號傳遞的方式來實(shí)現(xiàn)（實(shí)現(xiàn)方式有多種，信號量、管道、事件等，通訊都需要過內(nèi)核，效率低）

線程

• 共享變量（解決了通訊麻煩的問題，但是對于變量的訪問需要加鎖）
• 調(diào)度由操作系統(tǒng)完成（由于共享內(nèi)存，上下文切換變得高效）
• 一個(gè)進(jìn)程可以有多個(gè)線程，每個(gè)線程會共享父進(jìn)程的資源（創(chuàng)建線程開銷占用比進(jìn)程小很多，可創(chuàng)建的數(shù)量也會很多）
• 通訊除了可使用進(jìn)程間通訊的方式，還可以通過共享內(nèi)存的方式進(jìn)行通信（通過共享內(nèi)存通信比通過內(nèi)核要快很多）

協(xié)程

• 調(diào)度完全由用戶控制
• 一個(gè)線程（進(jìn)程）可以有多個(gè)協(xié)程
• 每個(gè)線程（進(jìn)程）循環(huán)按照指定的任務(wù)清單順序完成不同的任務(wù)（當(dāng)任務(wù)被堵塞時(shí)，執(zhí)行下一個(gè)任務(wù)；當(dāng)恢復(fù)時(shí)，再回來執(zhí)行這個(gè)任務(wù)；任務(wù)間切換只需要保存任務(wù)的上下文，沒有內(nèi)核的開銷，可以不加鎖的訪問全局變量）
• 協(xié)程需要保證是非堵塞的且沒有相互依賴
• 協(xié)程基本上不能同步通訊，多采用異步的消息通訊，效率比較高

總結(jié)

• 進(jìn)程擁有自己獨(dú)立的堆和棧，既不共享堆，亦不共享?xiàng)?，進(jìn)程由操作系統(tǒng)調(diào)度
• 線程擁有自己獨(dú)立的棧和共享的堆，共享堆，不共享?xiàng)?，線程亦由操作系統(tǒng)調(diào)度(標(biāo)準(zhǔn)線程是的)
• 協(xié)程和線程一樣共享堆，不共享?xiàng)?，協(xié)程由程序員在協(xié)程的代碼里顯示調(diào)度

聊聊協(xié)程

協(xié)程，又稱微線程，纖程。
Python的線程并不是標(biāo)準(zhǔn)線程，是系統(tǒng)級進(jìn)程，線程間上下文切換有開銷，而且Python在執(zhí)行多線程時(shí)默認(rèn)加了一個(gè)全局解釋器鎖（GIL），因此Python的多線程其實(shí)是串行的，所以并不能利用多核的優(yōu)勢，也就是說一個(gè)進(jìn)程內(nèi)的多個(gè)線程只能使用一個(gè)CPU。

def coroutine(func):
    def ret():
        f = func()
        f.next()
        return f
    return ret

@coroutine
def consumer():
    print "Wait to getting a task"
    while True:
        n = (yield)
        print "Got %s",n

import time
def producer():
    c = consumer()
    task_id = 0
    while True:
        time.sleep(1)
        print "Send a task to consumer" % task_id
        c.send("task %s" % task_id)

if __name__ == "__main__":
    producer()

結(jié)果：

Wait to getting a task
Send a task 0 to consumer
Got task 0
Send a task 1 to consumer
Got task 1
Send a task 2 to consumer
Got task 2
...

傳統(tǒng)的生產(chǎn)者-消費(fèi)者模型是一個(gè)線程寫消息，一個(gè)線程取消息，通過鎖機(jī)制控制隊(duì)列和等待，但容易死鎖。
如果改用協(xié)程，生產(chǎn)者生產(chǎn)消息后，直接通過yield跳轉(zhuǎn)到消費(fèi)者開始執(zhí)行，待消費(fèi)者執(zhí)行完畢后，切換回生產(chǎn)者繼續(xù)生產(chǎn)，效率極高。

Gevent

介紹

gevent是基于協(xié)程的Python網(wǎng)絡(luò)庫。特點(diǎn)：

• 基于libev的快速事件循環(huán)(Linux上epoll，F(xiàn)reeBSD上kqueue）。
• 基于greenlet的輕量級執(zhí)行單元。
• API的概念和Python標(biāo)準(zhǔn)庫一致(如事件，隊(duì)列)。
• 可以配合socket，ssl模塊使用。
• 能夠使用標(biāo)準(zhǔn)庫和第三方模塊創(chuàng)建標(biāo)準(zhǔn)的阻塞套接字(gevent.monkey)。
• 默認(rèn)通過線程池進(jìn)行DNS查詢,也可通過c-are(通過GEVENT_RESOLVER=ares環(huán)境變量開啟）。
• TCP/UDP/HTTP服務(wù)器
• 子進(jìn)程支持（通過gevent.subprocess）
• 線程池

安裝和依賴

依賴于greenlet library
支持python 2.6+ 、3.3+

核心部分

• Greenlets
• 同步和異步執(zhí)行
• 確定性
• 創(chuàng)建Greenlets
• Greenlet狀態(tài)
• 程序停止
• 超時(shí)
• 猴子補(bǔ)丁

Greenlets

gevent中的主要模式, 它是以C擴(kuò)展模塊形式接入Python的輕量級協(xié)程。 全部運(yùn)行在主程序操作系統(tǒng)進(jìn)程的內(nèi)部，但它們被程序員協(xié)作式地調(diào)度。

在任何時(shí)刻，只有一個(gè)協(xié)程在運(yùn)行。

區(qū)別于multiprocessing、threading等提供真正并行構(gòu)造的庫， 這些庫輪轉(zhuǎn)使用操作系統(tǒng)調(diào)度的進(jìn)程和線程，是真正的并行。

同步和異步執(zhí)行

并發(fā)的核心思想在于，大的任務(wù)可以分解成一系列的子任務(wù)，后者可以被調(diào)度成 同時(shí)執(zhí)行或異步執(zhí)行，而不是一次一個(gè)地或者同步地執(zhí)行。兩個(gè)子任務(wù)之間的 切換也就是上下文切換。

在gevent里面，上下文切換是通過yielding來完成的.

import gevent
def foo():
    print('Running in foo')
    gevent.sleep(0)
    print('Explicit context switch to foo again')
def bar():
    print('Explicit context to bar')
    gevent.sleep(0)
    print('Implicit context switch back to bar')
gevent.joinall([
    gevent.spawn(foo),
    gevent.spawn(bar),
])

執(zhí)行結(jié)果：

Running in foo
Explicit context to bar
Explicit context switch to foo again
Implicit context switch back to bar

網(wǎng)絡(luò)延遲或IO阻塞隱式交出greenlet上下文的執(zhí)行權(quán)。

import time
import gevent
from gevent import select
start = time.time()
tic = lambda: 'at %1.1f seconds' % (time.time() - start)
def gr1():
    print('Started Polling: %s' % tic())
    select.select([], [], [], 1)
    print('Ended Polling: %s' % tic())
def gr2():
    print('Started Polling: %s' % tic())
    select.select([], [], [], 2)
    print('Ended Polling: %s' % tic())
def gr3():
    print("Hey lets do some stuff while the greenlets poll, %s" % tic())
    gevent.sleep(1)
gevent.joinall([
    gevent.spawn(gr1),
    gevent.spawn(gr2),
    gevent.spawn(gr3),
])

執(zhí)行結(jié)果：

Started Polling: at 0.0 seconds
Started Polling: at 0.0 seconds
Hey lets do some stuff while the greenlets poll, at 0.0 seconds
Ended Polling: at 1.0 seconds
Ended Polling: at 2.0 seconds

同步vs異步

import gevent
import random
def task(pid):
    gevent.sleep(random.randint(0,2)*0.001)
    print('Task %s done' % pid)
def synchronous():
    for i in xrange(5):
        task(i)
def asynchronous():
    threads = [gevent.spawn(task, i) for i in xrange(5)]
    gevent.joinall(threads)
    print('Synchronous:')
    synchronous()
    print('Asynchronous:')
    asynchronous()

執(zhí)行結(jié)果：

Synchronous:
Task 0 done
Task 1 done
Task 2 done
Task 3 done
Task 4 done
Asynchronous:
Task 2 done
Task 0 done
Task 1 done
Task 3 done
Task 4 done

確定性

greenlet具有確定性。在相同配置相同輸入的情況下，它們總是會產(chǎn)生相同的輸出。

import time
def echo(i):
    time.sleep(0.001)
    return i
# Non Deterministic Process Pool
from multiprocessing.pool import Pool
p = Pool(10)
run1 = [a for a in p.imap_unordered(echo, xrange(10))]
run2 = [a for a in p.imap_unordered(echo, xrange(10))]
run3 = [a for a in p.imap_unordered(echo, xrange(10))]
run4 = [a for a in p.imap_unordered(echo, xrange(10))]
print(run1 == run2 == run3 == run4)
# Deterministic Gevent Pool
from gevent.pool import Pool
p = Pool(10)
run1 = [a for a in p.imap_unordered(echo, xrange(10))]
run2 = [a for a in p.imap_unordered(echo, xrange(10))]
run3 = [a for a in p.imap_unordered(echo, xrange(10))]
run4 = [a for a in p.imap_unordered(echo, xrange(10))]
print(run1 == run2 == run3 == run4)

執(zhí)行結(jié)果：

False
True

即使gevent通常帶有確定性，當(dāng)開始與如socket或文件等外部服務(wù)交互時(shí)， 不確定性也可能溜進(jìn)你的程序中。因此盡管gevent線程是一種“確定的并發(fā)”形式， 使用它仍然可能會遇到像使用POSIX線程或進(jìn)程時(shí)遇到的那些問題。

涉及并發(fā)長期存在的問題就是競爭條件(race condition)(當(dāng)兩個(gè)并發(fā)線程/進(jìn)程都依賴于某個(gè)共享資源同時(shí)都嘗試去修改它的時(shí)候， 就會出現(xiàn)競爭條件),這會導(dǎo)致資源修改的結(jié)果狀態(tài)依賴于時(shí)間和執(zhí)行順序。 這個(gè)問題，會導(dǎo)致整個(gè)程序行為變得不確定。

解決辦法: 始終避免所有全局的狀態(tài).

創(chuàng)建Greenlets

gevent對Greenlet初始化提供了一些封裝.

import gevent
from gevent import Greenlet
def foo(message, n):
    gevent.sleep(n)
    print(message)
    thread1 = Greenlet.spawn(foo, "Hello", 1)
    thread2 = gevent.spawn(foo, "I live!", 2)
    thread3 = gevent.spawn(lambda x: (x+1), 2)
    threads = [thread1, thread2, thread3]
    gevent.joinall(threads)

執(zhí)行結(jié)果：

Hello
I live!

除使用基本的Greenlet類之外，你也可以子類化Greenlet類，重載它的_run方法。

import gevent
from gevent import Greenlet
class MyGreenlet(Greenlet):
    def __init__(self, message, n):
        Greenlet.__init__(self)
        self.message = message
        self.n = n
    def _run(self):
        print(self.message)
        gevent.sleep(self.n)
g = MyGreenlet("Hi there!", 3)
g.start()
g.join()

執(zhí)行結(jié)果：

Hi there!

Greenlet狀態(tài)

greenlet的狀態(tài)通常是一個(gè)依賴于時(shí)間的參數(shù)：

• started – Boolean, 指示此Greenlet是否已經(jīng)啟動
• ready() – Boolean, 指示此Greenlet是否已經(jīng)停止
• successful() – Boolean, 指示此Greenlet是否已經(jīng)停止而且沒拋異常
• value – 任意值, 此Greenlet代碼返回的值
• exception – 異常, 此Greenlet內(nèi)拋出的未捕獲異常

程序停止

程序
當(dāng)主程序(main program)收到一個(gè)SIGQUIT信號時(shí)，不能成功做yield操作的 Greenlet可能會令意外地掛起程序的執(zhí)行。這導(dǎo)致了所謂的僵尸進(jìn)程， 它需要在Python解釋器之外被kill掉。

通用的處理模式就是在主程序中監(jiān)聽SIGQUIT信號，調(diào)用gevent.shutdown退出程序。

import gevent
import signal
def run_forever():
    gevent.sleep(1000)
    if __name__ == '__main__':
        gevent.signal(signal.SIGQUIT, gevent.shutdown)
        thread = gevent.spawn(run_forever)
        thread.join()

超時(shí)

通過超時(shí)可以對代碼塊兒或一個(gè)Greenlet的運(yùn)行時(shí)間進(jìn)行約束。

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

超時(shí)類

import gevent
from gevent import Timeout
time_to_wait = 5 # seconds
    class TooLong(Exception):
        pass
    with Timeout(time_to_wait, TooLong):
        gevent.sleep(10)

另外，對各種Greenlet和數(shù)據(jù)結(jié)構(gòu)相關(guān)的調(diào)用，gevent也提供了超時(shí)參數(shù)。

import gevent
from gevent import Timeout
def wait():
    gevent.sleep(2)
timer = Timeout(1).start()
thread1 = gevent.spawn(wait)
try:
    thread1.join(timeout=timer)
except Timeout:
    print('Thread 1 timed out')
# --
timer = Timeout.start_new(1)
thread2 = gevent.spawn(wait)
try:
    thread2.get(timeout=timer)
except Timeout:
    print('Thread 2 timed out')
# --
try:
    gevent.with_timeout(1, wait)
except Timeout:
    print('Thread 3 timed out')

執(zhí)行結(jié)果：

Thread 1 timed out
Thread 2 timed out
Thread 3 timed out

猴子補(bǔ)丁(Monkey patching)

gevent的死角.

import socket
print(socket.socket)
print("After monkey patch")
from gevent import monkey
monkey.patch_socket()
print(socket.socket)
import select
print(select.select)
monkey.patch_select()
print("After monkey patch")
print(select.select)

執(zhí)行結(jié)果：

class 'socket.socket'
After monkey patch
class 'gevent.socket.socket'

built-in function select
After monkey patch
function select at 0x1924de8

Python的運(yùn)行環(huán)境允許我們在運(yùn)行時(shí)修改大部分的對象，包括模塊，類甚至函數(shù)。 這是個(gè)一般說來令人驚奇的壞主意，因?yàn)樗鼊?chuàng)造了“隱式的副作用”，如果出現(xiàn)問題 它很多時(shí)候是極難調(diào)試的。雖然如此，在極端情況下當(dāng)一個(gè)庫需要修改Python本身 的基礎(chǔ)行為的時(shí)候，猴子補(bǔ)丁就派上用場了。在這種情況下，gevent能夠修改標(biāo)準(zhǔn)庫里面大部分的阻塞式系統(tǒng)調(diào)用，包括socket、ssl、threading和 select等模塊，而變?yōu)閰f(xié)作式運(yùn)行。

例如，Redis的python綁定一般使用常規(guī)的tcp socket來與redis-server實(shí)例通信。 通過簡單地調(diào)用gevent.monkey.patch_all()，可以使得redis的綁定協(xié)作式的調(diào)度 請求，與gevent棧的其它部分一起工作。

這讓我們可以將一般不能與gevent共同工作的庫結(jié)合起來，而不用寫哪怕一行代碼。 雖然猴子補(bǔ)丁仍然是邪惡的(evil)，但在這種情況下它是“有用的邪惡(useful evil)”。

數(shù)據(jù)結(jié)構(gòu)

• 事件

• 隊(duì)列

• 組和池

• 鎖和信號量

• 線程局部變量

• 子進(jìn)程

• Actors

  事件

  事件(event)是一個(gè)在Greenlet之間異步通信的形式。

import gevent
from gevent.event import Event
    
evt = Event()
    
def setter():
    print('A: Hey wait for me, I have to do something')
    gevent.sleep(3)
    print("Ok, I'm done")
    evt.set()
def waiter():
    print("I'll wait for you")
    evt.wait()  # blocking
    print("It's about time")
def main():
    gevent.joinall([
        gevent.spawn(setter),
        gevent.spawn(waiter),
        gevent.spawn(waiter),
        gevent.spawn(waiter)
    ])
if __name__ == '__main__': 
    main()

執(zhí)行結(jié)果：

A: Hey wait for me, I have to do something
I'll wait for you
I'll wait for you
I'll wait for you
Ok, I'm done
It's about time
It's about time
It's about time

事件對象的一個(gè)擴(kuò)展是AsyncResult，它允許你在喚醒調(diào)用上附加一個(gè)值。 它有時(shí)也被稱作是future或defered，因?yàn)樗钟幸粋€(gè)指向?qū)砣我鈺r(shí)間可設(shè)置為任何值的引用。

import gevent
from gevent.event import AsyncResult
a = AsyncResult()
def setter():
    gevent.sleep(3)
    a.set('Hello!')
def waiter():
    print(a.get())
gevent.joinall([
    gevent.spawn(setter),
    gevent.spawn(waiter),
])

隊(duì)列

隊(duì)列是一個(gè)排序的數(shù)據(jù)集合，它有常見的put / get操作， 但是它是以在Greenlet之間可以安全操作的方式來實(shí)現(xiàn)的。

import gevent
from gevent.queue import Queue
tasks = Queue()
def worker(n):
    while not tasks.empty():
        task = tasks.get()
        print('Worker %s got task %s' % (n, task))
        gevent.sleep(0)
    print('Quitting time!')
def boss():
    for i in xrange(1,10):
        tasks.put_nowait(i)
gevent.spawn(boss).join()
gevent.joinall([
    gevent.spawn(worker, 'steve'),
    gevent.spawn(worker, 'john'),
    gevent.spawn(worker, 'nancy'),
])

執(zhí)行結(jié)果：

Worker steve got task 1
Worker john got task 2
Worker nancy got task 3
Worker steve got task 4
Worker john got task 5
Worker nancy got task 6
Worker steve got task 7
Worker john got task 8
Worker nancy got task 9
Quitting time!
Quitting time!
Quitting time!

put和get操作都是阻塞的，put_nowait和get_nowait不會阻塞， 然而在操作不能完成時(shí)拋出gevent.queue.Empty或gevent.queue.Full異常。

組和池

組(group)是一個(gè)運(yùn)行中g(shù)reenlet集合，集合中的greenlet像一個(gè)組一樣會被共同管理和調(diào)度。 它也兼飾了像Python的multiprocessing庫那樣的平行調(diào)度器的角色，主要用在在管理異步任務(wù)的時(shí)候進(jìn)行分組。

import gevent
from gevent.pool import Group
def talk(msg):
    for i in xrange(2):
        print(msg)
g1 = gevent.spawn(talk, 'bar')
g2 = gevent.spawn(talk, 'foo')
g3 = gevent.spawn(talk, 'fizz')
group = Group()
group.add(g1)
group.add(g2)
group.join()
group.add(g3)
group.join()

執(zhí)行結(jié)果：

bar
bar
foo
foo
fizz
fizz

池(pool)是一個(gè)為處理數(shù)量變化并且需要限制并發(fā)的greenlet而設(shè)計(jì)的結(jié)構(gòu)。

import gevent
from gevent.pool import Pool
pool = Pool(2)
def hello_from(n):
    print('Size of pool %s' % len(pool))
    pool.map(hello_from, xrange(3))

執(zhí)行結(jié)果：

Size of pool 2
Size of pool 2
Size of pool 1

構(gòu)造一個(gè)socket池的類，在各個(gè)socket上輪詢。

from gevent.pool import Pool
class SocketPool(object):
    def __init__(self):
        self.pool = Pool(10)
        self.pool.start()
    def listen(self, socket):
        while True:
            socket.recv()
    def add_handler(self, socket):
        if self.pool.full():
            raise Exception("At maximum pool size")
        else:
            self.pool.spawn(self.listen, socket)
    def shutdown(self):
        self.pool.kill()

鎖和信號量

信號量是一個(gè)允許greenlet相互合作，限制并發(fā)訪問或運(yùn)行的低層次的同步原語。 信號量有兩個(gè)方法，acquire和release。在信號量是否已經(jīng)被 acquire或release，和擁有資源的數(shù)量之間不同，被稱為此信號量的范圍 (the bound of the semaphore)。如果一個(gè)信號量的范圍已經(jīng)降低到0，它會 阻塞acquire操作直到另一個(gè)已經(jīng)獲得信號量的greenlet作出釋放。

from gevent import sleep
from gevent.pool import Pool
from gevent.coros import BoundedSemaphore
sem = BoundedSemaphore(2)
def worker1(n):
    sem.acquire()
    print('Worker %i acquired semaphore' % n)
    sleep(0)
    sem.release()
    print('Worker %i released semaphore' % n)
def worker2(n):
    with sem:
        print('Worker %i acquired semaphore' % n)
        sleep(0)
    print('Worker %i released semaphore' % n)
pool = Pool()
pool.map(worker1, xrange(0,2))

執(zhí)行結(jié)果：

Worker 0 acquired semaphore
Worker 1 acquired semaphore
Worker 0 released semaphore
Worker 1 released semaphore

鎖(lock)是范圍為1的信號量。它向單個(gè)greenlet提供了互斥訪問。 信號量和鎖常被用來保證資源只在程序上下文被單次使用。

線程局部變量

Gevent允許程序員指定局部于greenlet上下文的數(shù)據(jù)。 在內(nèi)部，它被實(shí)現(xiàn)為以greenlet的getcurrent()為鍵， 在一個(gè)私有命名空間尋址的全局查找。

import gevent
from gevent.local import local
stash = local()
def f1():
    stash.x = 1
    print(stash.x)
def f2():
    stash.y = 2
    print(stash.y)
    try:
        stash.x
    except AttributeError:
        print("x is not local to f2")
g1 = gevent.spawn(f1)
g2 = gevent.spawn(f2)
gevent.joinall([g1, g2])

執(zhí)行結(jié)果：

1
2
x is not local to f2

很多集成了gevent的web框架將HTTP會話對象以線程局部變量的方式存儲在gevent內(nèi)。 例如使用Werkzeug實(shí)用庫和它的proxy對象，我們可以創(chuàng)建Flask風(fēng)格的請求對象。

from gevent.local import local
from werkzeug.local import LocalProxy
from werkzeug.wrappers import Request
from contextlib import contextmanager
from gevent.wsgi import WSGIServer
_requests = local()
request = LocalProxy(lambda: _requests.request)
@contextmanager
def sessionmanager(environ):
    _requests.request = Request(environ)
    yield
    _requests.request = None
def logic():
    return "Hello " + request.remote_addr
def application(environ, start_response):
    status = '200 OK'
    with sessionmanager(environ):
        body = logic()
    headers = [
        ('Content-Type', 'text/html')
    ]
    start_response(status, headers)
    return [body]
    WSGIServer(('', 8000), application).serve_forever()

子進(jìn)程

從gevent 1.0起，支持gevent.subprocess，支持協(xié)作式的等待子進(jìn)程。

import gevent
from gevent.subprocess import Popen, PIPE
def cron():
    while True:
        print("cron")
        gevent.sleep(0.2)
g = gevent.spawn(cron)
sub = Popen(['sleep 1; uname'], stdout=PIPE, shell=True)
out, err = sub.communicate()
g.kill()
print(out.rstrip())

執(zhí)行結(jié)果：

cron
cron
cron
cron
cron
Linux

很多人也想將gevent和multiprocessing一起使用。最明顯的挑戰(zhàn)之一 就是multiprocessing提供的進(jìn)程間通信默認(rèn)不是協(xié)作式的。由于基于 multiprocessing.Connection的對象(例如Pipe)暴露了它們下面的 文件描述符(file descriptor)，gevent.socket.wait_read和wait_write 可以用來在直接讀寫之前協(xié)作式的等待ready-to-read/ready-to-write事件。

import gevent
from multiprocessing import Process, Pipe
from gevent.socket import wait_read, wait_write
# To Process
a, b = Pipe()
# From Process
c, d = Pipe()
def relay():
    for i in xrange(5):
        msg = b.recv()
        c.send(msg + " in " + str(i))
def put_msg():
    for i in xrange(5):
        wait_write(a.fileno())
        a.send('hi')
def get_msg():
    for i in xrange(5):
        wait_read(d.fileno())
        print(d.recv())
if __name__ == '__main__':
    proc = Process(target=relay)
    proc.start()
    g1 = gevent.spawn(get_msg)
    g2 = gevent.spawn(put_msg)
    gevent.joinall([g1, g2], timeout=1)

執(zhí)行結(jié)果：

hi in 0
hi in 1
hi in 2
hi in 3
hi in 4

然而要注意，組合multiprocessing和gevent必定帶來 依賴于操作系統(tǒng)(os-dependent)的缺陷，其中有：

在兼容POSIX的系統(tǒng)創(chuàng)建子進(jìn)程(forking)之后， 在子進(jìn)程的gevent的狀態(tài)是不適定的(ill-posed)。一個(gè)副作用就是， multiprocessing.Process創(chuàng)建之前的greenlet創(chuàng)建動作，會在父進(jìn)程和子進(jìn)程兩方都運(yùn)行。

上例的put_msg()中的a.send()可能依然非協(xié)作式地阻塞調(diào)用的線程：一個(gè) ready-to-write事件只保證寫了一個(gè)byte。在嘗試寫完成之前底下的buffer可能是滿的。

上面表示的基于wait_write()/wait_read()的方法在Windows上不工作 (IOError: 3 is not a socket (files are not supported))，因?yàn)閃indows不能監(jiān)視 pipe事件。

Python包gipc以大體上透明的方式在 兼容POSIX系統(tǒng)和Windows上克服了這些挑戰(zhàn)。它提供了gevent感知的基于 multiprocessing.Process的子進(jìn)程和gevent基于pipe的協(xié)作式進(jìn)程間通信。

Actors

actor模型是一個(gè)由于Erlang變得普及的更高層的并發(fā)模型。 簡單的說它的主要思想就是許多個(gè)獨(dú)立的Actor，每個(gè)Actor有一個(gè)可以從 其它Actor接收消息的收件箱。Actor內(nèi)部的主循環(huán)遍歷它收到的消息，并根據(jù)它期望的行為來采取行動。

Gevent沒有原生的Actor類型，但在一個(gè)子類化的Greenlet內(nèi)使用隊(duì)列， 我們可以定義一個(gè)非常簡單的。

import gevent
from gevent.queue import Queue
class Actor(gevent.Greenlet):
    def __init__(self):
        self.inbox = Queue()
        Greenlet.__init__(self)
    def receive(self, message):
        """
        Define in your subclass.
        """
        raise NotImplemented()
    def _run(self):
        self.running = True
        while self.running:
            message = self.inbox.get()
            self.receive(message)

下面是一個(gè)使用的例子：

import gevent
from gevent.queue import Queue
from gevent import Greenlet
class Pinger(Actor):
    def receive(self, message):
        print(message)
        pong.inbox.put('ping')
        gevent.sleep(0)
class Ponger(Actor):
    def receive(self, message):
        print(message)
        ping.inbox.put('pong')
        gevent.sleep(0)
ping = Pinger()
pong = Ponger()
ping.start()
pong.start()
ping.inbox.put('start')
gevent.joinall([ping, pong])

實(shí)際應(yīng)用

• Gevent ZeroMQ
• 簡單server
• WSGI Servers
• 流式server
• Long Polling
• Websockets

簡單server

# On Unix: Access with ``$ nc 127.0.0.1 5000``
# On Window: Access with ``$ telnet 127.0.0.1 5000``
from gevent.server import StreamServer
def handle(socket, address):
    socket.send("Hello from a telnet!\n")
    for i in range(5):
        socket.send(str(i) + '\n')
    socket.close()
server = StreamServer(('127.0.0.1', 5000), handle)
server.serve_forever()

WSGI Servers And Websockets

Gevent為HTTP內(nèi)容服務(wù)提供了兩種WSGI server。從今以后就稱為 wsgi和pywsgi：

• gevent.wsgi.WSGIServer
• gevent.pywsgi.WSGIServer

glb中使用

import click
from flask import Flask
from gevent.pywsgi import WSGIServer
from geventwebsocket.handler import WebSocketHandler
import v1
from .settings import Config
from .sockethandler import handle_websocket
def create_app(config=None):
    app = Flask(__name__, static_folder='static')
    if config:
        app.config.update(config)
    else:
        app.config.from_object(Config)
    app.register_blueprint(
        v1.bp,
        url_prefix='/v1')
    return app
def wsgi_app(environ, start_response):
    path = environ['PATH_INFO']
    if path == '/websocket':
        handle_websocket(environ['wsgi.websocket'])
    else:
        return create_app()(environ, start_response)
@click.command()
@click.option('-h', '--host_port', type=(unicode, int),
    default=('0.0.0.0', 5000), help='Host and port of server.')
@click.option('-r', '--redis', type=(unicode, int, int),
    default=('127.0.0.1', 6379, 0),
    help='Redis url of server.')
@click.option('-p', '--port_range', type=(int, int),
    default=(50000, 61000),
    help='Port range to be assigned.')
def manage(host_port, redis=None, port_range=None):
    Config.REDIS_URL = 'redis://%s:%s/%s' % redis
    Config.PORT_RANGE = port_range
    http_server = WSGIServer(host_port,
                             wsgi_app, handler_class=WebSocketHandler)
    print '----GLB Server run at %s:%s-----' % host_port
    print '----Redis Server run at %s:%s:%s-----' % redis
    http_server.serve_forever()

缺陷

和其他異步I/O框架一樣,gevent也有一些缺陷:

• 阻塞(真正的阻塞,在內(nèi)核級別)在程序中的某個(gè)地方停止了所有的東西.這很像C代碼中monkey patch沒有生效
• 保持CPU處于繁忙狀態(tài).greenlet不是搶占式的,這可能導(dǎo)致其他greenlet不會被調(diào)度.
• 在greenlet之間存在死鎖的可能.

一個(gè)gevent回避的缺陷是,你幾乎不會碰到一個(gè)和異步無關(guān)的Python庫–它將阻塞你的應(yīng)用程序,因?yàn)榧働ython庫使用的是monkey patch的stdlib.