网络编程之协程与池

Posted 2020-12-10 tonxin66

协程

本质：在单线程下，由用户自己控制一个任务遇到io阻塞了就切换另外一个任务去执行，以此来提升效率

目的：减少操作系统切换的开销，规避IO操作，将一条线程中的io操作 降到最低

定义

• 协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的。
• 特点：
  • 必须在只有一个单线程里实现并发
  • 修改共享数据不需加锁
  • 用户程序里自己保存多个控制流的上下文栈

协程的切换

    运行
   /    阻塞——就绪
    
# 进程为等待输入而阻塞
# 调度程序选择里一个进程
# 调度程序选择另一个进程
# 出现有效输入
--------------
# gevent =  利用了  greenlet    底层模块完成的切换 + 自动规避io的功能
# asyncio = 利用了  yield    	 底层语法完成的切换 + 自动规避io的功能

协程优点

1. 协程的切换开销更小，属于程序级别的切换，更加轻量级
2. 单线程内就可以实现并发的效果，最大限度地利用cpu
3. 用户级别，减轻了操作系统的负担

协程缺点

1. 无法利用多核，单线程下实现并发

Gevent模块

可以轻松通过gevent实现并发同步后异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程

用法 ：gevent.spawn(函数名，参数)

def func(a,b,c,x=3):pass
g1 = gevent.spawn(func,1,3,4,x=5)  # 创建一个协程对象g1

补丁

# gevent.sleep()模拟的是gevent可以识别的io阻塞，但time.sleep（2）或其他的阻塞不能被gevent直接识别，所以需要打补丁
from gevent import monkey
monkey.pstch_all()  # 最好放在文件开头，这样后面indeed所有阻塞全部能够识别了
import time,socket

• 协程是通过自己的程序（代码）来进行切换的，自己能够控制，只有遇到协程模块能够识别的IO操作的时候，程序才会进行任务切换，实现并发效果

asyncio模块

import asyncio

async def func(name):
    print(‘start‘,name)
    # await 可能会发生阻塞的方法
    # await 关键字必须写在一个async函数里
    await asyncio.sleep(2)
    print(‘end‘)

lo = asyncio.get_event_loop()
lo.run_until_complete(asyncio.wait([func(‘alex‘),func(‘太白‘)]))

池

在程序开始的时候，还没提交任务先创建几个线程或者进程放在池中

发展过程

1. threading模块 没有提供池
2. multiprocessing模块 仿照threading写的Pool
3. concurrent.futures模块 线程池，进程池都用相同的方式开启/使用

池的好处

1. 可直接使用这个池中的数据
2. 开好的线程和进程可以被多个任务反复利用，极大的减少了开启/关闭/调度线程和进程的时间开销
3. 控制了操作系统需要调度的任务个数，有利于提供操作系统的效率，减轻了操作系统的负担

线程池

import time
from threading import current_thread
from concurrent.futures import ThreadPoolExecutor
def func(a,b):
    print(current_thread().ident,‘start‘,a,b)
    time.sleep(2)
    pritn(current_thread().ident,‘end‘)
if __name__ == ‘__main__‘:
    p = ThreadPoolExecutor(4)
    for i in range(10):
        p.submit(func,i,i+1)  # 向池中提交任务,和向函数传参

进程池

import time
from threading import current_thread
from concurrent.futures import ProcessPoolExecutor
def func(a,b):
    print(current_thread().ident,‘start‘,a,b)
    time.sleep(2)
    pritn(current_thread().ident,‘end‘)
if __name__ == ‘__main__‘:
    p = ProcessPoolExecutor(4)
    for i in range(10):
        p.submit(func,i,i+1)  # 向池中提交任务,和向函数传参

回调函数

• 效率最高

import time,random
from threading import current_thread
from concurrent.futures import ThreadPoolExecutor

def func(a,b):
    print(current_thread().ident,‘start‘,a,b)
    time.sleep(random.randint(1,4))
    print(current_thread().ident,‘end‘,a)
    return (a,a*b)

def print_func(ret):       # 异步阻塞
    print(ret.result())

if __name__ == ‘__main__‘:
    tp = ThreadPoolExecutor(4)
    futrue_l = {}
    for i in range(20):         # 异步非阻塞的
        ret = tp.submit(func,i,b=i+1)
        ret.add_done_callback(print_func) # ret这个任务会在执行完毕的瞬间立即触发print_func函数
# 异步阻塞,给对象绑定一个回调函数,可以对结果立即处理,而不是按照顺序结果再处理结果