ITPub博客
首页 博客 专栏 学院 社区 IT新媒体评选
发博

首页 > 数据库 > 数据库开发技术 > python 并发编程-协程篇（2）greenle和gevent简单认识

python 并发编程-协程篇（2）greenle和gevent简单认识

原创 数据库开发技术 作者：程序员小钟同学 时间：2021-06-03 15:06:07 0 删除 编辑

上一篇中主要基于yield一种方式来实现一个基本雏形的协程的实现。从上面的yield的可以看到，其实我们的协程需要处理的问题，就是

❝

在一个线程内可以实现不同的函数的调用切换的执行，通俗的说其实就是可以实现多个任务在同一个线程里面并发的执行。

但是从上面我们的也可以得到一个关键点的问题是：

  • 协程是单线程的下的一种并发模式，这个就有一点的弊端，就是无法利用多核，但是多核的问题，其实我们的可以直接多开几个进程来解决。但是相对多线程模式来说节省了线程切换的开销

  • 协程任务切换其实类似一种程序中断，但是它与函数调用不同，中断通常是发生阻塞的情况下。才会去切换（需要不是需要手动切换的库的支持）

  • 上一篇也说道关于IO多路复用和协程的之间的对比，IO多路复用其实是在一个进程里面除了多个逻辑的流程，且依赖依赖于事件的回调通知，在某种程度上导致了我们的处理流程函数的分散，一个逻辑会被切割成多个函数去处理，但是协程上，则没有这种问题。

  • 协程需要保证是非堵塞的且没有相互依赖

  • 协程基本上不能同步通讯，多采用异步的消息通讯，效率比较高

  • 协程和多进程和多线程，再扯皮一次并行和并发：

    • 某种意义上：进程和多线程其实是真正的‘并行’（但是在单核的的情况下也是并发，只是宏观和微观一种区别）
    • 而协程因为有且只有一个线程处理多任务，其实是一种‘并发’

所以协程这玩意某种意义上肯定是非常好的，其实GO上也有协程的说话，后续有机会的话，也得研究学习一下。

下面的说的关于协程的其他库的补充的话，估计已经是老生常谈了，但是这里我还是补充到我们的系列梳理里面，主要为为自己重头再理解一次！

1 greenlet 库实现的协程

https://greenlet.readthedocs.io/en/latest/#introduction 最好的学习资料是官网文档

定义：

用于进程内顺序并发编程的轻量级协同器

greenlet的缺点：

  • 需要手动的进行切换（类似yield中的send）

运行方式

  • 1：首先定义一个需要执行任务函数
  • 2：把需要执行任务函数丢到协程里面执行
  • 3：需进程多任务的情况的切换时候进行.switch()来切换

几个方法

简单跑任务示例 

import greenlet

def run():
    print("Running in the greenlet function.")

glet = greenlet.greenlet(run)
print(glet.parent)
print(glet.getcurrent().parent)

Greenlet之间的切换：使用switch() 进行协程任务的调用和切换 

"""
Author = zyx
@version: v1.0.0
@File: greenlet.py
@文件功能描述:------
"""

import time
from greenlet import greenlet


def test1():
    while True:
        print("___test1--执行了___")
        g2.switch()
        print("___g2.切换到test2执行后，又切换回来后--执行了___")
        time.sleep(0.5)


def test2():
    while True:
        print("___test2执行了___")
        g1.switch()
        print("___ g1.切换到test1执行后，又切换回来后--执行了___")
        time.sleep(0.5)


g1 = greenlet(test1)
g2 = greenlet(test2)


# 切换到gr1中运行
g1.switch()

输出结果：

___test1--执行了___
___test2执行了___
___g2.切换到test2执行后，又切换回来后--执行了___
___test1--执行了___
___ g1.切换到test1执行后，又切换回来后--执行了___
___test2执行了___
___g2.切换到test2执行后，又切换回来后--执行了___
___test1--执行了___
___ g1.切换到test1执行后，又切换回来后--执行了___
___test2执行了___
___g2.切换到test2执行后，又切换回来后--执行了___
___test1--执行了___
___ g1.切换到test1执行后，又切换回来后--执行了___
___test2执行了___
___g2.切换到test2执行后，又切换回来后--执行了___

上面流程的说明:

  • 1:首先先把两个协程任务丢到我们的greenlet里面，创建出来协程的对象（仅仅是创建对象，不会执行函数的执行）

  • 2：调用协程对象自身的.switch()来进行任务的执行，首先执行了test1的任务。

  • 3：g1.switch()----开始执行我们的test1函数内部的任务

  • 4：test1函数内部开始执行，然后开始进行了g2.switch()任务的切换，此时我们的test1任务会被切断，暂停执行，然后我们的test2函数内部开始执行

  • 5：test2函数内部开始执行后，又遇到了切换到我们的g1.switch()，此时我们的test2任务会被切断暂停，然后返回我们的test1的任务暂停的地方继续执行，所以此时输出了

_g2.切换到test2执行后，又切换回来后--执行了___，

然后等到test1这个执行结束的时候，因为我们的test1第一次的循环完成了，然后再开始了第二次的循环---此时我们的test1函数内部【又开始】执行，然后【又开始】进行了g2.switch()任务的切换，test1又开始暂停了！

  • 6：因为我们的test2任务上一次只是被暂停了，还没完成第一次的循环，所以此时，因为g2.switch()之后，开始恢复上一次的暂停的执行的地方，然后又继续跑，所以输出了：
___ g1.切换到test1执行后，又切换回来后--执行了___

然后完成了test2任务的第一次的循环的执行，接着第二次的循环又切换到了g1.switch()。

····· 所以关键几个步骤就是上面的那些地方！因为这个地方，在我们的web开发部署的时候，其实也有用到，在使用我们的python gunicorn里面有配置默认使用greenlet的方式还是gevent的方式。

但是单独拎出来使用，目前还是相对少~~所以仅仅当做补充协程的一个认识就好！有兴趣的人可以继续看官网文档继续研究！


2 gevent 库实现的协程


这个gevent如果你再部署flask项目的时候，估计肯定是有遇到过，他的启动的模式是使用这个模型的话，则内部我们的FLask则是使用协程的并发模式。

还有如果你使用消息队列：Celery的话，应该也知道gevent还可以引入到celery里面。启动相关的worker的时候指定：

celery worker -P gevent

celery beat-P gevent
❝

题外话

不过之前我再部署的时候遇到过一个关于使用gevent的问题，就是单独的跑一个简单的FLask的项目的话，是没有什么问题！但是遇到复杂的项目的时候，突然发现在进行猴子布丁的时候，他就无法启动服务了！！这个让我很疑惑，大概估计是某个第三方库引发的 不支持的猴子布丁的模式，当时也么去认真的细究，直接换了另一种模式进行部署了！！！有机会再看看.

猴子补丁

猴子补丁其实就是gevent的一个牛逼的地方，它能够修改标准库里面大部分的阻塞式系统调用，包括socket、ssl、threading和 select等模块，让他们变为协作式运行。，一句话搞定所有同步的方法变为异步的方法。当然如我前面说的，仅限于大部分的标准库，第三方库的的话，有点难说！

gevent 和 greenlet

gevent，相比greenlet的话，主要区别就是比较智能化，在遇到IO(比如网络、文件操作等)操作时，可以自动切换到其他的greenlet，等到IO操作完成，在适当的时候再次的切换回来继续执行。由于IO操作非常耗时，如果我们的程序一直处于等待状态，那肯定不行，gevent可以为我们自动化管理IO阻塞态的时候，进行相关任务协程的切换 ，不需要再手动进行切换了。

关于gevent我们需要了解几点：

  • gevent，是通过封装了 libev（基于epoll的快速事件循环） 和 greenlet 两个库来实现协程的。

  • gevent主要是针对我们的IO阻塞的一个自动切换协程库。

  • gevent内部调度其实是所有其它的greenlet实例

gevent简单示例应用 

import gevent
import requests

def task_1(num):
    for i in range(num):
        print(gevent.getcurrent(), i)
        # 模拟一个网络请求耗时操作
        print(requests.get('http://www.baidu.com?q='+str(1)).text)


if __name__ == "__main__":
    g1 = gevent.spawn(task_1, 5)  # 创建协程
    g2 = gevent.spawn(task_1, 5)
    g3 = gevent.spawn(task_1, 5)

    # 等待协程运行完毕，再退出主进程
    g1.join()
    g2.join()
    g3.join()

如果感兴趣你还可以研究一下grequets这个库，这库主要是基于gevent实现了对requets的的异步支持。

pip install gevent grequests

示例：

import grequests

urls = [
    'http://www.heroku.com',
    'http://python-tablib.org',
    'http://httpbin.org',
    'http://python-requests.org',
    'http://fakedomain/',
    'http://kennethreitz.com'
]

rs = (grequests.get(u) for u in urls)
或者
res_list = grequests.map(urls)  # 并行发送
print(res_list[0].text) # 打印第一个请求的响应文本

使用猴子补丁的：

from gevent import monkey
monkey.patch_all()  # 给所有的耗时操作打上猴子补丁

import gevent
import requests



def task_1(num):
    for i in range(num):
        print(gevent.getcurrent(), i)
        # 模拟一个网络请求耗时操作
        print(requests.get('http://www.baidu.com?q='+str(1)).text)


if __name__ == "__main__":

    g1 = gevent.spawn(task_1, 5)  # 创建协程
    g2 = gevent.spawn(task_1, 5)
    g3 = gevent.spawn(task_1, 5)

    # 等待协程运行完毕，再退出主进程
    g1.join()
    g2.join()
    g3.join()

❝

PS：from gevent import monkey monkey.patch_all()  # 给所有的耗时操作打上猴子补丁 需要放在最开始的地方！

也不要这样的进行导入使用：

from gevent import monkey
import gevent
import requests
monkey.patch_all()  # 给所有的耗时操作打上补丁

上面会提示：

MonkeyPatchWarning: Monkey-patching ssl after ssl has already been imported may lead to errors, including RecursionError on Python 3.6. It may also silently lead to incorrect behaviour on Python 3.7. Please monkey-patch earlier. See https://github.com/gevent/gevent/issues/1016. Modules that had direct imports (NOT patched): ['urllib3.util (C:\\Users\\15944\\Desktop\\pyproj\\Python\\CSGFrame\\venv\\lib\\site-packages\\urllib3\\util\\__init__.py)', 'urllib3.util.ssl_ (C:\\Users\\15944\\Desktop\\pyproj\\Python\\CSGFrame\\venv\\lib\\site-packages\\urllib3\\util\\ssl_.py)']. 
  monkey.patch_all()  

标准的就是：

from gevent import monkey
monkey.patch_all()  # 给所有的耗时操作打上补丁
import gevent
import requests

约束任务的执行超时控制

前两种方式：

from gevent import monkey
monkey.patch_all()  # 给所有的耗时操作打上补丁
import gevent
import requests
from gevent import Timeout
import time


def task_1(num):
    # 模拟一个网络请求耗时操作
    # print(requests.get('http://www.baidu.com?q='+str(1)).text)
    # 没有引入的猴子补丁的情况下，不要使用time.sleep，而应该使用 gevent.sleep()模拟
    time.sleep(5)


if __name__ == "__main__":

    g1 = gevent.spawn(task_1, 5)  # 创建协程
    g2 = gevent.spawn(task_1, 5)
    g3 = gevent.spawn(task_1, 5)

    # 等待协程运行完毕，再退出主进程

    try:
        # 如果任务1的超时时间到了则直接的抛出异常
        # 第一种方式
        timer = Timeout(0.01).start() # 必须设置这个来定义的超时时间
        g1.join(timeout=timer)
        # 第二钟方式
        g1.get(timeout=timer)
      
    except Timeout:
        print('g1任务执行超时！')
    g2.join()
    g3.join()

with_timeout约束的方式：

from gevent import monkey
monkey.patch_all()  # 给所有的耗时操作打上补丁
import gevent
import requests
from gevent import Timeout
import time


def task_1(num):
    # 模拟一个网络请求耗时操作
    # print(requests.get('http://www.baidu.com?q='+str(1)).text)
    # 没有引入的猴子补丁的情况下，不要使用time.sleep，而应该使用 gevent.sleep()模拟
    time.sleep(5)


if __name__ == "__main__":

    # g1 = gevent.spawn(task_1, 5)  # 创建协程
    g1 =None
    try:
        g1 = gevent.with_timeout(1, task_1, 5)
        g1.join()
    except Timeout:
        print('task_1  timed out')
    g2 = gevent.spawn(task_1, 5)
    g3 = gevent.spawn(task_1, 5)

    # 等待协程运行完毕，再退出主进程

    g2.join()
    g3.join()

gevent的同步问题

但凡是涉及到多任务协作，其实我们会发现，不管是协程，还是线程，还是进程都会涉及到同步问题的处理。

相同的，他们的处理的同步的问题的方式，也是大同小异的，也无非是哪几种：

  • 事件（Event）

  • 队列(Quest)

  • 锁（Lock）和信号量

  • 组（类似障碍对象）和池 等等

gevent同步处理-Event简单示例

他们的其实也和我们的线程进程一样的，有set之类的。


import gevent
from gevent.event import Event

# 事件对象
evt = Event()


def xiaofeizhe():
    print(' 客人: 1:服务员给我来一个鸡腿！')
    # 模拟阻塞---阻塞后---我马上切换到fuwuyuan里面
    gevent.sleep(3)
    print("客人: 3：我最爱吃鸡腿啊！")
    evt.set()
    evt.wait()
    print("客人: 5：你还真会推销啊！")

def fuwuyuan():
    print("服务员：2:大早上的吃什么鸡腿啊！喝粥不香吗？")
    # 开始调用wait进行blocking阻塞等待
    evt.wait()
    print("服务员：4:要不来再来给鸭腿！")

    evt.set()
    evt.wait()
    print("服务员：6:呵呵哒！！！！")

def main():
    gevent.joinall([
        gevent.spawn(xiaofeizhe),
        gevent.spawn(fuwuyuan)
    ])


if __name__ == '__main__':
    main()

输出结果：

 客人: 1:服务员给我来一个鸡腿！
服务员：2:大早上的吃什么鸡腿啊！喝粥不香吗？
客人: 3：我最爱吃鸡腿啊！
服务员：4:要不来再来给鸭腿！
客人: 5：你还真会推销啊！
服务员：6:呵呵哒！！！！

其他的暂时不重复了~~~

gevent的一些实际应用：

  • gevent.wsgi.WSGIServer
  • gevent.pywsgi.WSGIServer
  • geventwebsocket
  • gevent.server.StreamServer

其中部署我们的Flask的时候我使用的一种方式就是：

from gevent import monkey, spawn

monkey.patch_all()

from flask import Flask, request, url_for, current_app, jsonify, g

app = Flask(__name__)


@app.route('/whhy', methods=['POST', 'GET'])
def returl():
    # 记录下所有的请提交的数据信息
    return jsonify({
        "code": 0,
        "message": "success"
    })


@app.route('/whhy/bakurl/callback', methods=['POST', 'GET'])
def bakurl():
    return jsonify({
        "code": 0,
        "message": "success"
    })

if __name__ == '__main__':
    print(app.url_map)
      # -------flask系统启动配置 ---------
    WSGIServer(('0.0.0.0', 7790),app).serve_forever()  # 配置gevent后的启动方式
    from gevent import pywsgi
    server = pywsgi.WSGIServer(('0.0.0.0', 20012), app)
    server.serve_forever()

参考资料：

https://www.jianshu.com/p/4dca99ffc0b4

❝

个人其他博客地址

❝

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

小钟同学 | 文 【原创】| QQ：308711822

  • 1：本文相关描述主要是个人的认知和见解，如有不当之处，还望各位大佬指正。
  • 2：关于文章内容，有部分内容参考自互联网整理，如有链接会声明标注；如没有及时标注备注的链接的，如有侵权请联系，我会立即删除处理哟。


原文作者：程序员小钟同学；如有侵权，请联系管理员删除。

0
0
上一篇： 一支 asyncio 穿云箭，多个协程来相见
下一篇： python 并发编程-协程篇（0）-协程一些概念基础认识
请登录后发表评论 登录
全部评论
程序员小钟同学

注册时间：2021-02-20

  • 博文量
    224
  • 访问量
    111723

最新文章

支持我们 作者招募 用户协议 FAQ Contact Us

北京盛拓优讯信息技术有限公司. 版权所有  京ICP备16024965号-8 北京市公安局海淀分局网监中心备案编号：11010802021510 niuxiaotong@pcpop.com 17352615567 未成年人举报专区

广播电视节目制作经营许可证(京) 字第1234号 中国互联网协会会员

违法和不良信息举报电话:18101376593