您的位置：首页 > 编程语言 > Python开发

基础入门_Python—Gevent异步/状态获取/超时设置/猴子补丁

2017-02-27 00:00 471 查看

简单介绍:

说明: Gevent是一个基于libev的并发库,为各种并发和网络相关的任务提供了整洁的API

快速安装:

pipinstall --upgrade gevent

主要模式:

说明: Greenlet以C扩展模块形式接入PY轻量级协程,它们运行于主进程内部,被协作式的调度,且不同于multiprocessing和threading等真正的并行执行,它在同一时刻只能有一个协程在运行

公有方法
gevent.spawn(cls,args,*kwargs)	创建一个Greenlet对象,其实调用的是Greenlet.spawn(需要from geventimport Greenlet),返回greenlet对象
gevent.joinall(greenlets,timeout=None,raise_error=False,count=None)	等待所有greenlets全部执行完毕,greenlets为序列,timeout为超时计时器对象,返回执行完毕未出错的的greenlet序列
greenlet
g.join()	等待此协程执行完毕后

1. 使用基本封装类初始化Greenlets

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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 gevent

# 说明: 导入其它模块

def
 task001():

print
 'I'

gevent.sleep(
0
)

print
 '#=> switch to task002 run.'

print
 'You'

def
 task002():

print
 'LOVE'

gevent.sleep(
0
)

print
 '#=> switch to task001 run.'

if
 __name__
=
=
 '__main__'
:

gevent.joinall([

gevent.spawn(task001),

gevent.spawn(task002)

])

2. 继承Greenlets基类重载_run方法

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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 gevent

from
 gevent
import
 Greenlet

# 说明: 导入其它模块

class
 Gevent(Greenlet):

def
 __init__(
self
,msg,
*
args,
*
*
kwargs):

super
(Gevent,
self
).__init__(
*
args,
*
*
kwargs)

self
.msg
=
 msg

# 重载run方法,join后会自动调用此方法执行

def
 _run(
self
):

print
 self
.msg

gevent.sleep(
0
)

if
 __name__
=
=
 '__main__'
:

for
 _
in
 (
'I'
,
'LOVE'
,
'YOU'
):

g
=
 Gevent(_)

g.start()

g.join()

注意: 多条gevent.spawn(cls,*args,**kwargs).join()语句即使为阻塞调用也不会协程式调用,因为生成的Greenlet对象执行完就消失,所有要实现协程式调用可通过gevent.joinall(greenlets,timeout=None,raise_error=False,count=None)或是赋值到一变量后再对此对象调用.join(),其中一个特殊的方式就是for循环调用将隐式的赋值对象调用,会自动变为协程式运行.

同步异步:

说明: 并发的核心在于可拆分成小任务的大任务被分成子任务然后被上下文切换调度,在Gevent中,上下文切换调度是通过yielding来完成的,默认自动切换上下文,当然也可手动gevent.sleep(0)切换上下文

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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 gevent

# 说明: 导入其它模块

def
 task001():

print
 'I'

gevent.sleep(
0
)

print
 '#=> switch to task002 run.'

print
 'You'

def
 task002():

print
 'LOVE'

gevent.sleep(
0
)

print
 '#=> switch to task001 run.'

if
 __name__
=
=
 '__main__'
:

gevent.joinall([

gevent.spawn(task001),

gevent.spawn(task002)

])

说明: 如上实例手动调用gevent.sleep(0)来切换上下文,将生成的调度任务通过gevent.joinall加入调度队列,使得只有执行完调度队列的任务后才能继续往下走,一旦调度任务出现阻塞则立即切换到下一个上下文执行

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
35
36
37
38
39
40
41

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 socket

import
 gevent

from
 gevent
import
 select

# 说明: 导入其它模块

def
 handler(server,port):

r_list,w_list,e_list 
=
 [server],[],[]

while
 True
:

r,w,e 
=
 select.select(r_list,w_list,e_list)

for
 sock
in
 r:

if
 sock
is
 server:

c_sock,c_addr 
=
 sock.accept()

r_list.append(c_sock)

print
 'found notice: client #%s connectd to port #%s'
 %
 (c_addr,port)

else
:

data
=
 sock.recv(
1024
)

data
=
 data.strip()

if
 data
in
 (
'quit'
,
'exit'
):

sock.close()

r_list.remove(sock)

print
 'found notice: client #%s disconnectd'
 %
 (sock.getpeername(),)

else
:

print
 'found notice: client #%s send data#%s to port #%s'
 %
 (sock.getpeername(),data,port)

if
 __name__
=
=
 '__main__'
:

slist
=
 []

ports
=
 [
8001
,
8002
]

for
 p
in
 ports:

server
=
 socket.socket(socket.AF_INET,socket.SOCK_STREAM)

server.bind((
'0.0.0.0'
,p))

server.listen(
10000
)

slist.append(gevent.spawn(handler,server,p))

gevent.joinall(slist)

说明: Gevent使得函数协作式调度,对于我们隐藏的所有实现细节,保证网络库在可能的时候,隐式交出上下文执行权,上面实例中利用协程+多路复用I/O SELECT实现同时绑定多个端口处理数据的ECHO SERVER,我并没有显式的调用gevent.sleep(0),但是依然执行着上下文切换,这说明当我们在受限于网络或IO的函数中调用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
33
34

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 time

import
 pprint

import
 multiprocessing

from
 gevent.pool
import
 Pool

# 说明: 导入其它模块

def
 echo_data(data):

time.sleep(
0.001
)

return
 data

if
 __name__
=
=
 '__main__'
:

# 多进程 - 进程池

res
=
 []

p
=
 multiprocessing.Pool(
10
)

for
 _
in
 xrange
(
10
):

res.append(
list
(p.imap_unordered(echo_data,[_
for
 _
in
 xrange
(
10
)])))

p.close()

p.join()

pprint.pprint(res)

print
 '################################'

# 协程式 - 协程池

res
=
 []

p
=
 Pool(
20
)

for
 _
in
 xrange
(
10
):

res.append(
list
(p.imap_unordered(echo_data,[_
for
 _
in
 xrange
(
10
)])))

pprint.pprint(res)

说明: Gevent在相同调用相同输入相同SLEEP时产生的结果总是相同的,相对于多进程/多线程的即使在相同调用相同输入相同SLEEP下结果不总相同

注意: Gevent在处理不确定阻塞时间调用时,你的程序可能会出现不确定性,这是由于并发通病-竞争条件导致,最好的解决办法是尽量避免所有的全局状态.避免竞争访问或是修改.

状态获取:

说明: Greenlet可能由于不同的原因运行失败,但其并不会抛出异常,而是用状态变量/方法跟踪线程内部的状态

状态相关
g.started	协程是否已经启动
g.ready()	协程是否已经停止
g.successful()	同上,但是没有抛异常
g.value	协程返回的值
g.exception	协程内抛出的未捕捉的异常
g.get(block=True,timeout=None)	获取greenlet的返回值或重新抛出运行中异常,timeout设置计时器对象

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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 gevent

# 说明: 导入其它模块

# 谈情

def
 love():

print
 'ok'

# 做爱

def
 makelove():

raise
 Exception(
'fuck'
)

if
 __name__
=
=
 '__main__'
:

greenlets
=
 [gevent.spawn(love),gevent.spawn(makelove)]

gevent.joinall(greenlets)

for
 g
in
 greenlets:

print
 'started? %s'
 %
 g.started

print
 'ready? %s'
 %
 g.ready()

print
 'successful? %s'
 %
 g.successful()

print
 'exception? %s'
 %
 g.exception

超时设置:

说明: 超时是一种对代码块儿或Greenlet的运行时间约束

公有方法
gevent.Timeout(seconds,exception)	创建一个计时器对象
Timeout
t.start()	启动一个计时器对象
t.cancel()	取消一个计时器对象
t.start_new(timeout=None,exception=None,ref=True)	创建一个新的计时器对象,timeout为超时时间,默认单位为秒

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
35

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 gevent

# 说明: 导入其它模块

def
 run_forever(interval):

while
 True
:

gevent.sleep(interval)

if
 __name__
=
=
 '__main__'
:

timer
=
 gevent.Timeout(
1
).start()

g
=
 gevent.spawn(run_forever,
1000000
)

# 执行超时

try
:

g.join(timeout
=
timer)

except
 gevent.Timeout,e:

print
 'found error: thread 1 timeout.'

# 获取超时

timer
=
 gevent.Timeout.start_new(
1
)

g
=
 gevent.spawn(run_forever,
1000000
)

try
:

g.get(timeout
=
timer)

except
 gevent.Timeout,e:

print
 'found error: thread 2 timeout.'

# 执行超时

try
:

gevent.with_timeout(
1
,run_forever,
1000000
)

except
 gevent.Timeout,e:

print
 'found error: thread 3 timeout.'

猴子补丁:

说明: PY的默认运行环境允许我们运行中修改大部分对象,包括模块/类/函数等,而猴子补丁可以修改PY标准库里面大部分的阻塞式系统调用,实在不知哪些库可打补丁可gevent.monkey.patch_all()即可.

patch_all/patch_builtins/patch_dns/patch_item/patch_module/patch_os/patch_select/patch_signal/patch_socket/patch_ssl/patch_subprocess/patch_sys/patch_thread/patch_time

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

#!/usr/bin/env python

# -*- coding: utf-8 -*-

"""

# Authors: limanman

# OsChina: http://xmdevops.blog.51cto.com/[/code] # Purpose:

#

"""

# 说明: 导入公共模块

import
 select

# 说明: 导入其它模块

if
 __name__
=
=
 '__main__'
:

print
 '#select before path %s'
 %
 (select.select,)

from
 gevent
import
 monkey

monkey.patch_all()

print
 '#select after path %s'
 %
 (select.select,)

登录乐搏学院官网http://www.learnbo.com/

或关注我们的官方微博微信，还有更多惊喜哦~

本文出自 “满满李 - 运维开发之路” 博客，请务必保留此出处http://xmdevops.blog.51cto.com/11144840/1862085

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航