基础入门_Python-模块和包.Gevent异步/状态获取/超时设置/猴子补丁?

Posted 2020-08-14

简单介绍:

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

快速安装:

pip install --upgrade gevent

主要模式:

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

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

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

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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方法

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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中,上下文切换调度是通过yielding来完成的,默认自动切换上下文,当然也可手动gevent.sleep(0)切换上下文

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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加入调度队列,使得只有执行完调度队列的任务后才能继续往下走,一旦调度任务出现阻塞则立即切换到下一个上下文执行

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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时会隐式的切换上下文来实现协作式调度.

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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设置计时器对象

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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为超时时间,默认单位为秒

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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

---

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
#
# Authors: limanman
# OsChina: http://xmdevops.blog.51cto.com/
# 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://xmdevops.blog.51cto.com/11144840/1862085