104.协程

Posted 2021-03-09 tk-tank

协程

• 定义：为非抢占式多任务产生子程序的计算机程序组件
• 协程允许不同入口点在不同位置暂停或开始执行程序
• 从技术角度讲，协程是可以暂停的函数，或直接理解为生成器
• 本质是单线程，相比多线程，可以及其节省系统资源
• 协程函数运行结束后，如果还执行.send()就会报错StopIteration
• yield即使返回值，也是可以是下一步的参数输入值

def xc():
    print("协程开始")
    #yield既是出口，也是入口
    a = yield
    print("协程从新开始：", a)

#实例一个协程
x = xc()

print("11111")
#此步骤可以使用next(x)
#开始执行，也叫预激
x.send(None)
print("22222")
x.send("haha")

inspect.getgeneratorstate(x) 查询协程的状态(inspect模块)

• ‘GEN_CREATED‘ 等待开始执行。
• ‘GEN_RUNNING‘ 解释器正在执行。
• ‘GEN_SUSPENDED‘ 在 yield 表达式处暂停。
• ‘GEN_CLOSED‘ 执行结束。

协程终止

• 协程中文处理的异常，会向上传递给next或.seed方法的调用方
• 止协程的另一种方法：发送哨符值，让协程推出，一内置的None或Ellipsis

yield from

• 委派生成器
  • 包含yield from的生成器函数

def A():
    for i in [1, 2]:
        yield i

print(list(A()))

def B():
    #相当于在list和[1, 2]间的中间地带
    yield from [1, 2]

print(list(B()))

asyncio包

• python3.4开始引入便准库，内置对异步IO的支持
• asyncio本身是一个消息循环
• 创建过程：
  1. 创建消息循环
  2. 把协程引入
  3. 关闭消息循环

import threading
import asyncio

@asyncio.coroutine
def hello():
    print('Hello world! (%s)' % threading.currentThread())
    #跳出协程，睡5秒
    yield from asyncio.sleep(1)
    print('Hello again! (%s)' % threading.currentThread())

#启动消息循环
loop = asyncio.get_event_loop()
#定义要执行的任务
tasks = [hello(), hello()]
#asyncio使用wati等待task执行完毕
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

#执行结果
Hello world! (<_MainThread(MainThread, started 2400)>)
Hello world! (<_MainThread(MainThread, started 2400)>)
Hello again! (<_MainThread(MainThread, started 2400)>)
Hello again! (<_MainThread(MainThread, started 2400)>)

async 和 await

• python3.5引入
• 更好的表示异步io
• 让协程代码更简洁
• 在语法上可以简单的替换
  • async 替换 @asyncio.coroutine
  • await 替换 yield from

aidhttp

• asyncio可以实现单线程并发IO操作
• 客户端用处不大
• 在服务器端，可以处理多用户的高并发用coroutine+单线程
• asyncio实现了TCP.UDP.SSL等协议
• aiohttp是给予asyncio的http框架
• 不是标准模块，需安装

concurrent.futures

• python3库
• 线程池
  -利用multiprocessiong实现真正的并行计算，前提是cup得是多核
• 原理已子进程的形式，并行运行多个python解释器，子进程和主进程是相互独立的，所以他们的全局解释器锁是相互独立的，每个子进程都能完整的使用一个cup内核
• from concurrent.futures import ThreadPoolExecutor
  • ThreadPoolExecutor(max_workers=数量) 创建线程池，并指定线程数量，根据需要选择
  • ProcesspoolExecutor(max_workers=数量) 创建进程池，并指定进程数量，根据需要选择
  • .submit(fn, args, kwargs) 创建线程或进程 fn线程或进程函数的函数名
  • .done 线程执行的状态 Ture执行完毕， False未执行完毕
  • .result 线程或进程函数执行的结果

import time
from concurrent import futures

def return_futures(a):
    time.sleep(2)
    return a

#创建一个线程池,max_workers工作线程数量
s = futures.ThreadPoolExecutor(max_workers=2)
#向线程池中加入两个任务
r1 = s.submit(return_futures, "hi")
r2 = s.submit(return_futures, "hello")

#判断两个线程执行状态
print(r1.done())
print(r2.done())
time.sleep(3)
print(r1.done())
print(r2.done())

#输出两个线程的结果
print(r1.result())
print(r2.result())

• .map()
  • 和map函数类似，映射
  • 异步执行

import time
from concurrent import futures

def wait_on(a):
    print(a)
    time.sleep(2)
    return "ok"

l = [1, 2]

t = futures.ThreadPoolExecutor(max_workers=2)
for i in t.map(wait_on, l):
    print(i)