关于gevent的协程间通信及队列和事件event用法

前言：

今天就写点gevent的高级点的用法，对于我来说这些也是常用的gevent模块。

gevent的AsyncResutl模块的用途，看字眼的意思是一个异步的任务的结果。其实官方的说法也让人有些发蒙。其实说白了就是协程间的通信，我是老板，让大哥和小弟同事去收账，小弟做完了后，会等大哥来问话。如果小弟没有完成，还在做着事情，那大哥会在一个时间里，等待小弟返回结果。一直等！

在实战中这个就很有意思了。我们同时做一个事情，但是我们又需要互相的帮助，或者是互相的通信。这个时候就可以用asyncresult。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

#coding:utf-8
#http://rfyiamcool.blog.51cto.com/1030776/1538367

import gevent

from gevent.event import AsyncResult
 

a = AsyncResult()
 

def xiaodi():

    """

    一会堵上10秒 ！

    """

    print "xiaodi 开始"

    gevent.sleep(10)

    a.set('hello world')

    print "xiaodi 结束"
 

def dage():

    """

    需要等待xiaodi完事了后，他才能i live

    """

    print 'dage 这里是先开始的...'

    print a.get() # blocking

    print 'I live!'
 
gevent.joinall([

    gevent.spawn(xiaodi),

    gevent.spawn(dage),
])

这里是gevent的事件用法。rawlink是注册一个回调，wait是等待任务的完成之后，才能后续进行。他的用法有些像twisted的一些思想，注册事件，事件和回调函数关联。 gevent最大的优势是用同步的写法，实现异步的功能。所以这功能不太实用。

e = Event()

e.rawlink(callback_def)

e.wait()

e.xxx

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

#coding:utf-8
#http://rfyiamcool.blog.51cto.com/1030776/1538367

import gevent

import time
 

def event_setter(e):

    print '开始搞了...'

    e.rawlink(event_callback)

    gevent.sleep(2)

    print '222...'

    e.set()
 

def event_waiter(e):

    print '等待...'

    e.wait()

    print '等待 end'
 

def event_callback(e):

    print "回调..."
 

def try_event():

    from gevent.event import Event

    e = Event()

    gevent.joinall([

        gevent.spawn(event_setter, e),

        gevent.spawn(event_waiter, e),

        gevent.spawn(event_waiter, e),

        gevent.spawn(event_waiter, e),

        gevent.spawn(event_waiter, e),

        gevent.spawn(event_waiter, e),

        gevent.spawn(event_waiter, e),

    ])
 
try_event()

gevent自己有个gevent.queue，我自己没做测试，倒是看一些老外再谈论，在数据大的数据的时候，要比pyhton queue本身的队列要抗用。

gevent queue 的队列功能很是丰富！

1
2
3
4
5
6
7
8


>>> queue = gevent.queue.Queue()

>>> queue.put(1)

>>> queue.put(2)
>>> queue.put(StopIteration)

>>> for item in queue:

...    print item
1
2

原文： http://rfyiamcool.blog.51cto.com/1030776/1538367

下面是一个比较完整的例子，大家跑跑：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16


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,25):

        tasks.put_nowait(i)
 
gevent.spawn(boss).join()
 
gevent.joinall([

    gevent.spawn(worker, 'fuck shencan'),

    gevent.spawn(worker, 'fuck zb'),

    gevent.spawn(worker, 'fuck liudehua'),
])

gevent队列的方法还是很牛叉的。比如put 推送队列，get取出队列，里面可以加几个选项堵塞和超时时间。

1
2
3
4
5
6
7
8
9


put(item, block=True, timeout=None) 往队列放入数据，可选是否阻塞和超时时间
put_nowait(item) 非阻塞的往队列放入数据，队列已满时抛出Full Exception

get(block=True, timeout=None) 从队列读出数据，可选是否阻塞和超时时间
get_nowait() 费阻塞地从队列读出数据，队列为空是抛出Empty Exception

peek(block=True, timeout=None) 和get()类似，但获取的数据不会从队列移除
peek_nowait() 类似get_nowait()

empty() 队列为空返回True

full() 队列已满返回True
qsize() 返回队列长度，即队列中的数据数，而非Queue(maxlength)初始化时的maxlength

原文：http://rfyiamcool.blog.51cto.com/1030776/1538367

外加一个gevent执行状态的判断。

1
2
3
4
5

started  表明是否gevent已经开始
ready()  表明是否gevent已经停止
successful()  表明是否gevent已经停止并且没有抛出异常

value --  任意，gevent返回的值

exception --  gevent异常,gevent内部实例没有被捕抓的异常

本文转自 rfyiamcool 51CTO博客，原文链接：http://blog.51cto.com/rfyiamcool/1538367，如需转载请自行联系原作者

关于gevent的协程间通信及队列和事件event用法

相关推荐