时间戳

首页 > 技术文章 > python10-协程

python10-协程

tensorzhang 2021-07-30 10:00 原文

协程到底是什么？
python协程的本质

1，概念

  • 线程：

    • 正常的开发语言：多线程可以利用多核。
    • cpython解释器下的多个线程不能利用多核：这本质上是规避了所有IO操作的单线程。

  • 协程：

    • 是操作系统不可见的。
    • 协程本质就是一条线程，多个任务在一条线程上来回切换（因）以规避IO操作（果），以达到将一条线程中的IO操作降到最低的目的。

  • 进程、线程、协程之间的对比：

    操作 数据隔离/共享 数据是否安全 操作级别 开销 多核
    进程 数据隔离 数据不安全 操作系统级别(dis.dis) 非常大 能利用多核
    线程 数据共享 数据不安全 操作系统级别 开销小（几百倍） 不能利用多核
    协程 数据共享（一条线程，肯定共享） 数据安全 用户级别 更小（函数切换级别） 不能利用多核

  • 协程相比于线程的缺陷：

    • 协程的所有切换都基于用户，只有在用户级别能够感知到的IO操作才会用协程模块做切换以进行规避，比如：socket、请求网页等。
    • 但是一些和文件操作相关的IO只有操作系统能够感知到，此时只能用线程。
    • 线程的感知更加细腻。

  • 用户级别操作的协程有什么好处：

    • 减轻了操作系统的负担。
    • 一条线程如果开了多个协程，能够多争取一些时间片来被CPU执行，提高程序的效率。

2，模块

  • 切换并规避IO的两个模块：
    • gevent：利用了 greenlet 底层模块完成的切换+自动规避IO的功能；
    • asyncio：利用了 yield 底层语法完成的切换+自动规避IO的功能。
  • 基于python原生的协程概念的发展史：
    • tornado：一种异步的web框架，基于yield实现。
    • yield from，更好的实现协程。
    • send，更好的实现协程。
    • asyncio模块，基于python原生的协程的概念正式被成立。
  • 特殊的在python中提供协程功能的关键字：aysnc，await。

3，gevent模块

  • 带有IO操作的命令(conn.recv，使用gevent，务必将import gevent，from gevent import monkey，monkey.patch_all()三行语句放在其他所有的import语句之前)写在函数func中，然后提交func给gevent；如果主程序里没有阻塞，需要自己加上阻塞

    '''例子1'''
import gevent

def func():
    print('start func')
    gevent.sleep(1)
    print('end func')

g1 = gevent.spawn(func)
g2 = gevent.spawn(func)
g3 = gevent.spawn(func)
gevent.joinall([g1, g2, g3])
# g1.join()       #必须阻塞，因为协程只有在有IO操作的时候才切换
# g2.join()
# g3.join()

  • gevent.sleep与time.sleep不同；如果要想让time.sleep和gevent.sleep起到一样的效果，需要如下操作：

    '''例子2'''
import gevent
from gevent import monkey
monkey.patch_all(thread=False, select=False)
import time

def func():
    print('start func')
    time.sleep(1) #与time.sleep不同，如果想要让
    print('end func')

g1 = gevent.spawn(func)
g2 = gevent.spawn(func)
g3 = gevent.spawn(func)
gevent.joinall([g1, g2, g3])

  • 使用协程进行一个服务端与多个客户端的通信

    '''===============================server==============================='''
import gevent
from gevent import monkey
monkey.patch_all()
import socket

def func(conn):
    while True:
        msg = conn.recv(1024).decode('utf-8')
        MSG = msg.upper()
        conn.send(MSG.encode('utf-8'))

sk = socket.socket()
sk.bind(('127.0.0.1', 9000))
sk.listen()

while True:
    conn, _ = sk.accept()
    gevent.spawn(func, conn)

'''===============================client==============================='''
import socket
import time

sk = socket.socket()
sk.connect(('127.0.0.1', 9000))

while True:
    sk.send(b'hello')
    msg = sk.recv(1024)
    print(msg)
    time.sleep(0.5)

  • 协程的效率，起500个客户端

    4核CPU
    总的进程数 5个
    每个进程的线程数 20个
    每个线程的协程数 500个

    一个四核机器开5个进程，每个进程开20个线程，每个线程开500个协程，极限可以抗50000的并发。（一般每个机器抗30000的并发）

    '''===============================client==============================='''
import socket
import time
from threading import Thread

def client():
    sk = socket.socket()
    sk.connect(('127.0.0.1', 9000))

    while True:
        sk.send(b'hello')
        msg = sk.recv(1024)
        print(msg)
        time.sleep(0.5)

for i in range(500):
    Thread(target=client).start()

  • 测试gevent是否能可以对引起阻塞的操作起到协程的作用，可以进行如下判断：

    import socket
print(socket.socket)
import gevent
from gevent import monkey
monkey.patch_all()
import socket
print(socket.socket)

'''两次打印的结果相同，则gevent可以规避该IO操作；如果不同，gevent无法规避该IO操作；看patch_all源码，所有为True的都支持'''
def patch_all(socket=True, dns=True, time=True, select=True, thread=True, os=True, ssl=True,
              subprocess=True, sys=False, aggressive=True, Event=True,
              builtins=True, signal=True,
              queue=True, contextvars=True,
              **kwargs):

4，asyncio模块

import asyncio

async def func(name):
    print('start', name)
    await asyncio.sleep(1)  #await标志会在此处切走
    print('end')

loop = asyncio.get_event_loop()
# loop.run_until_complete(func('a'))
loop.run_until_complete(asyncio.wait([func('a'), func('b')]))

  • 原理

    • yield和next配合可以做到在执行函数的过程中从函数中切出去。

    • 使用python实现的协程：

      import time

def sleep(n):
    print('start sleep')
    yield time.time()+n
    print('end sleep')

def func(n):
    print(123)
    g = sleep(n)
    yield from g    #await就相当于yield from
    print(456)

g1 = func(2)
g2 = func(1)
ret1 = next(g1)
ret2 = next(g2)
timeDict = {ret1: g1, ret2: g2} #仅多1s就已经体现出了异步
print(timeDict)
while timeDict:
    minTime = min(timeDict)
    time.sleep(minTime - time.time())   #异步，同时睡了1s
    try:
        next(timeDict[minTime])
    except StopIteration:
        pass
    del timeDict[minTime]

推荐阅读