Python gevent学习笔记－2
2024-08-29 15:44:30
在上一篇里面介绍了gevent的最主要的功能，先来来了解一下gevent里面一些更加高级的功能。
事件
事件是一种可以让greenlet进行异步通信的手段。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
import geventfrom gevent.event import AsyncResulta = AsyncResult()def setter(): """ After 3 seconds set wake all threads waiting on the value of a. """ gevent.sleep(3) a.set()def waiter(): """ After 3 seconds the get call will unblock. """ a.get() # blocking print 'I live!'gevent.joinall([ gevent.spawn(setter), gevent.spawn(waiter),]) |
AsyncResult 是 event对象的扩展能够让你来发送值并且带有一定延迟。这种功能被成为feature或deferred，当它拿到一个未来的值的引用时，能够在任意安排好的时间内让它起作用。
队列
队列是一个有序的数据集合，通常有 put/get 的操作，这样能让队列在有在有greenletJ进行操作的时候能够进行安全的管理。
例如，如果greenlet从队列中取出了一项数据，那么这份数据就不能被另一个greenlet取出。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
import geventfrom gevent.queue import Queuetasks = 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, 'steve'), gevent.spawn(worker, 'john'), gevent.spawn(worker, 'nancy'),]) |
执行的结果如下：
|
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
|
Worker steve got task 1Worker john got task 2Worker nancy got task 3Worker steve got task 4Worker nancy got task 5Worker john got task 6Worker steve got task 7Worker john got task 8Worker nancy got task 9Worker steve got task 10Worker nancy got task 11Worker john got task 12Worker steve got task 13Worker john got task 14Worker nancy got task 15Worker steve got task 16Worker nancy got task 17Worker john got task 18Worker steve got task 19Worker john got task 20Worker nancy got task 21Worker steve got task 22Worker nancy got task 23Worker john got task 24Quitting time!Quitting time!Quitting time! |
队列的 put/get 操作在需要的情况下也可以阻塞程序的执行。
put 和 get 操作都有非阻塞的副本，就是 put_nowait 和 get_nowait。
在下面代码的例子里，运行一个叫boss的方法，同时运行worker方法，并且对队列有一个限制：队列的子项不能超过3个。这个限制意味着 put 操作在队列里面有足够空间之前会阻塞。相反，如果队列里没有任何子项，get操作会阻塞，同时也需要超时的机制，当一个操作在阻塞超过一定时间后会抛出异常。
|
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
|
import geventfrom gevent.queue import Queue, Emptytasks = Queue(maxsize=3)def worker(n): try: while True: task = tasks.get(timeout=1) # decrements queue size by 1 print('Worker %s got task %s' % (n, task)) gevent.sleep(0) except Empty: print('Quitting time!')def boss(): """ Boss will wait to hand out work until a individual worker is free since the maxsize of the task queue is 3. """ for i in xrange(1,10): tasks.put(i) print('Assigned all work in iteration 1') for i in xrange(10,20): tasks.put(i) print('Assigned all work in iteration 2')gevent.joinall([ gevent.spawn(boss), gevent.spawn(worker, 'steve'), gevent.spawn(worker, 'john'), gevent.spawn(worker, 'bob'),]) |
代码的执行结果如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
Worker steve got task 1Worker john got task 2Worker bob got task 3Worker steve got task 4Worker bob got task 5Worker john got task 6Assigned all work in iteration 1Worker steve got task 7Worker john got task 8Worker bob got task 9Worker steve got task 10Worker bob got task 11Worker john got task 12Worker steve got task 13Worker john got task 14Worker bob got task 15Worker steve got task 16Worker bob got task 17Worker john got task 18Assigned all work in iteration 2Worker steve got task 19Quitting time!Quitting time!Quitting time! |
组和池
组是一个由greenlet组成的集合，并且能够被统一管理。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
|
import geventfrom gevent.pool import Groupdef talk(msg): for i in xrange(3): 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() |
这在管理一组异步任务的时候会很有用。
Group还提供了一个API来分配成组的greenlet任务，并且通过不同的方法来获取结果。
|
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
|
import geventfrom gevent import getcurrentfrom gevent.pool import Groupgroup = Group()def hello_from(n): print('Size of group', len(group)) print('Hello from Greenlet %s' % id(getcurrent()))group.map(hello_from, xrange(3))def intensive(n): gevent.sleep(3 - n) return 'task', nprint('Ordered')ogroup = Group()for i in ogroup.imap(intensive, xrange(3)): print(i)print('Unordered')igroup = Group()for i in igroup.imap_unordered(intensive, xrange(3)): print(i) |
执行结果如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
Size of group 3Hello from Greenlet 10769424Size of group 3Hello from Greenlet 10770544Size of group 3Hello from Greenlet 10772304Ordered('task', 0)('task', 1)('task', 2)Unordered('task', 2)('task', 1)('task', 0) |
池是用来处理当拥有动态数量的greenlet需要进行并发管理（限制并发数）时使用的。
这在处理大量的网络和IO操作的时候是非常需要的。
|
1
2
3
4
5
6
7
8
9
|
import geventfrom gevent.pool import Poolpool = Pool(2)def hello_from(n): print('Size of pool', len(pool))pool.map(hello_from, xrange(3)) |
|
1
2
3
|
Size of pool 2Size of pool 2Size of pool 1 |
经常在创建gevent驱动程序的时候，整个服务需要围绕一个池的结构来执行。
锁和信号量
信号量是低级别的同步机制，能够让greenlet在执行的时候互相协调并且限制其并发数。信号量暴露了两个方法，acquire 和 release。如果信号量范围变成0，那么它会阻塞住直到另一个greenlet释放它的获得物。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
from gevent import sleepfrom gevent.pool import Poolfrom gevent.coros import BoundedSemaphoresem = 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))pool.map(worker2, xrange(3,6)) |
一下是代码的执行结果：
|
1
2
3
4
5
6
7
8
9
10
|
Worker 0 acquired semaphoreWorker 1 acquired semaphoreWorker 0 released semaphoreWorker 1 released semaphoreWorker 3 acquired semaphoreWorker 4 acquired semaphoreWorker 3 released semaphoreWorker 4 released semaphoreWorker 5 acquired semaphoreWorker 5 released semaphore |
如果把信号量的数量限制为1那么它就成为了锁。它经常会在多个greenlet访问相同资源的时候用到。
本地线程
Gevent还能够让你给gevent上下文来指定那些数据是本地的。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
import geventfrom gevent.local import localstash = 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]) |
以下是执行结果：
|
1
2
3
|
12x is not local to f2 |
很多集成了gevent的框架把HTTP的session对象存在gevent 本地线程里面。比如下面的例子：
|
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
|
from werkzeug.local import LocalProxyfrom werkzeug.wrappers import Requestfrom contextlib import contextmanagerfrom gevent.wsgi import WSGIServer_requests = local()request = LocalProxy(lambda: _requests.request)@contextmanagerdef sessionmanager(environ): _requests.request = Request(environ) yield _requests.request = Nonedef logic(): return "Hello " + request.remote_addrdef 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() |
子进程
在gevent 1.0版本中，gevent.subprocess 这个库被添加上。这个库能够让子进程相互协调地执行。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
import geventfrom gevent.subprocess import Popen, PIPEdef 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() |
执行结果：
|
1
2
3
4
5
6
|
croncroncroncroncronLinux |
最新文章
- javascript基础知识show
- LINQ标准查询操作符详解(转)
- Mongo技巧-连接数据库与修改表结构
- 快速傅里叶变换（FFT）
- 从零基础入门JavaScript(2)
- Linux下添加磁盘创建lvm分区
- java 集合类说明与区别
- DDR、DDR2、DDR3产品区别
- 教你使用破解无线路由器笔记本password
- ios-制作静态.a文件
- hdu--1258--Sum It Up（Map水过）
- javascript DOM事件总结
- Java IO(Properties/对象序列化/打印流/commons-io)
- jenkins主要目录用途
- densenet 中的shortcut connection
- net core 2.0 + Autofac的坑
- ES系列十六、集群配置和维护管理
- RAC集群安装校验输出信息
- angular.js前端分层开发（页面和js代码分离，并将js代码分层）
- Kubernetes 1.8火热出炉：稳定性、安全性与存储支持能力全面提升