正文

Python的异步编程[0] -> 协程[1] -> 使用协程建立自己的异步非阻塞模型

IP云
IP云V管理员/6 阅读
08/05
提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

使用协程建立自己的异步非阻塞模型

接下来例子中，将使用纯粹的Python编码搭建一个异步模型，相当于自己构建的一个asyncio模块，这也许能对asyncio模块底层实现的理解有更大的帮助。主要参考为文末的链接，以及自己的补充理解。

完整代码

#!/usr/bin/python
# =============================================================
# File Name: async_base.py
# Author: LI Ke
# Created Time: 1/29/2018 09:18:50
# ============================================================= import types
import time @types.coroutine
def switch():
print('Switch: Start')
yield
print('Switch: Done') async def coro_1():
print('C1: Start')
await switch()
print('C1: Stop') async def coro_2():
print('C2: Start')
print('C2: 1')
print('C2: 2')
print('C2: 3')
print('C2: Stop') c_1 = coro_1()
c_2 = coro_2() try:
c_1.send(None)
except StopIteration:
pass
try:
c_2.send(None)
except StopIteration:
pass
try:
c_1.send(None)
except StopIteration:
pass print('--------------------------------') def run(coros):
coros = list(coros) while coros:
# Duplicate list for iteration so we can remove from original list
for coro in list(coros):
try:
coro.send(None)
except StopIteration:
coros.remove(coro) c_1 = coro_1()
c_2 = coro_2()
run([c_1, c_2]) print('--------------------------------') @types.coroutine
def action(t):
trace=[]
while True:
trace.append(time.time())
if trace[-1] - trace[0] > t:
break # This break will end this function and raise a StopIteration
yield async def coro_1():
print('C1: Start')
await action(2)
print('C1: Stop') async def coro_2():
print('C2: Start')
await action(3)
print('C2: Stop') def timeit(f):
def _wrapper(*args, **kwargs):
start = time.time()
re = f(*args, **kwargs)
end = time.time()
print('Time cost:', f.__name__, end-start)
return re
return _wrapper c_1 = coro_1()
c_2 = coro_2()
timeit(run)([c_1])
timeit(run)([c_2]) print('--------------------------------') c_1 = coro_1()
c_2 = coro_2()
timeit(run)([c_1, c_2])

分段解释

首先会导入需要的模块，这里仅仅使用types和time两个模块，放弃异步I/O的asyncio模块。

import types
import time

接下来定义一个switch函数，利用types.coroutine装饰器将switch装饰成一个协程，这个协程将完成一个切换功能。

@types.coroutine
def switch():
print('Switch: Start')
yield
print('Switch: Done')

随后定义第一个协程，协程启动后，会进入一个await，即切入刚才的switch协程，这里使用async和await关键字完成对协程的定义。

async def coro_1():
print('C1: Start')
await switch()
print('C1: Stop')

同样的，再定义第二个协程，第二个协程将从头到尾顺序执行。

async def coro_2():
print('C2: Start')
print('C2: 1')
print('C2: 2')
print('C2: 3')
print('C2: Stop')

有了上面的两个协程，但我们在异步时，希望在执行完C_1的start后，切换进协程C_2，执行完成后再切换回来。那么此时就需要一个对协程切换进行控制的程序，具体顺序如下，

  1. 启动协程c_1，启动后会切换进switch函数，
  2. Switch中由于yield而切出，并保留上下文环境
  3. c_1.send()将获得返回结果（如果有的话），并继续执行
  4. 此时c_1已经被中止，启动c_2，则完成所有执行步骤，捕获生成器的中止异常
  5. 这时c_2以执行完毕，再次切回c_1（此时会从switch yield之后开始执行）继续执行。

c_1 = coro_1()
c_2 = coro_2() try:
c_1.send(None)
except StopIteration:
pass
try:
c_2.send(None)
except StopIteration:
pass
try:
c_1.send(None)
except StopIteration:
pass

最终得到结果如下，可以看到，整个过程完全按期望的流程进行，

C1: Start
Switch: Start
C2: Start
C2: 1
C2: 2
C2: 3
C2: Stop
Switch: Done
C1: Stop

但是这里的协程运行部分仍需改善，于是接下来便定义一个run函数用于执行一个协程列表。

run函数首先会遍历协程列表的副本，并不断尝试启动列表中的协程，当协程结束后便将协程从协程列表中删除，直到所有的协程都执行完毕为止。

def run(coros):
coros = list(coros) while coros:
# Duplicate list for iteration so we can remove from original list
for coro in list(coros):
try:
coro.send(None)
except StopIteration:
coros.remove(coro) c_1 = coro_1()
c_2 = coro_2()
run([c_1, c_2])

测试一下run函数，得到结果与前面相同，

C1: Start
Switch: Start
C2: Start
C2: 1
C2: 2
C2: 3
C2: Stop
Switch: Done
C1: Stop

到目前为止，完成了一个简单的异步模型的搭建，即c_2无需等待c_1执行完成再继续执行，而是由c_1交出了控制权进行协作完成，同时也不存在多线程的抢占式任务，因为由始至终都只有一个线程在运行，而且也没有混乱的回调函数存在。

但是，还存在一个阻塞问题没有解决，也就是说，如果c_1中的switch函数是一个耗时的I/O操作或其他阻塞型操作，则此时需要等待switch的阻塞操作完成才能交出控制权，可如果希望在等待这个耗时操作时，先去执行c_2的任务，再回来检测c_1中的耗时操作是否完成，则需要使用非阻塞的方式。

首先，对刚才的switch进行改造，完成一个action协程，这个协程会根据传入的参数，执行对应时间后，再退出协程引发StopIteration，实现方式如下，每次切换进action中都会记录下时间，然后将时间和第一次进入的时间进行对比，如果超过了设置的时间便退出，如果没超过限制时间，则切出协程交还出控制权。

@types.coroutine
def action(t):
trace=[]
while True:
trace.append(time.time())
if trace[-1] - trace[0] > t:
break # This break will end this function and raise a StopIteration
yield

接着定义两个协程，分别执行action时间为2秒和3秒，同时定义一个计算时间的装饰器，用于时间记录。

async def coro_1():
print('C1: Start')
await action(2)
print('C1: Stop') async def coro_2():
print('C2: Start')
await action(3)
print('C2: Stop') def timeit(f):
def _wrapper(*args, **kwargs):
start = time.time()
re = f(*args, **kwargs)
end = time.time()
print('Time cost:', f.__name__, end-start)
return re
return _wrapper

然后我们先分别运行两个协程进行一个实验，

c_1 = coro_1()
c_2 = coro_2()
timeit(run)([c_1])
timeit(run)([c_2])

从输出的结果可以看到两个协程的耗时与action执行的时间基本相同，且顺序执行的时间为两者之和，

C1: Start
C1: Stop
Time cost: run 2.030202865600586
C2: Start
C2: Stop
Time cost: run 3.0653066635131836

接下来，利用异步非阻塞的方式来执行这两个协程，

c_1 = coro_1()
c_2 = coro_2()
timeit(run)([c_1, c_2])

最后得到结果

C1: Start
C2: Start
C1: Stop
C2: Stop
Time cost: run 3.0743072032928467

从结果中可以看到，此时的运行方式是异步的形式，c_1启动后由于进入一个耗时action，且action被我们设置为非阻塞形式，因此c_1交出了控制权，控制权回到run函数后，启动了c_2，而c_2同样也进入到action中，这时两个协程都在等待任务完成，而监视run则在两个协程中不停轮询，不断进入action中查看各自的action操作是否完成，当有协程完成后，将继续启动这个协程的后续操作，直到最终所有协程结束。

按照非阻塞异步协程的方式，可以以单线程运行，避免资源锁的建立，也消除了线程切换的开销，并且最终获得了类似多线程运行的时间性能。

相关阅读

1. 协程和 async / await

参考链接

http://www.oschina.net/translate/playing-around-with-await-async-in-python-3-5

Python的异步编程[0] -> 协程[1] -> 使用协程建立自己的异步非阻塞模型的更多相关文章

  1. Python的异步编程[0] -> 协程[0] -> 协程和 async / await

    协程 / Coroutine 目录 生产者消费者模型 从生成器到异步协程– async/await 协程是在一个线程执行过程中可以在一个子程序的预定或者随机位置中断,然后转而执行别的子程序,在适当的时 ...

  2. Python的网络编程[0] -> socket[1] -> socket 模块

    socket 1. 常量 / Constants AF_* 和 SOCK_* 分别属于 AddressFamily 和 SocketType 1.1 AF_*类常量 socket.AF_UNIX:  ...

  3. 异步编程系列第02章 你有什么理由使用Async异步编程

    p { display: block; margin: 3px 0 0 0; } --> 写在前面 在学异步,有位园友推荐了<async in C#5.0>,没找到中文版,恰巧也想提 ...

  4. Python 之并发编程之进程上&lpar;基本概念、并行并发、cpu调度、阻塞 &rpar;

    一: 进程的概念:(Process) 进程就是正在运行的程序,它是操作系统中,资源分配的最小单位. 资源分配:分配的是cpu和内存等物理资源 进程号是进程的唯一标识 同一个程序执行两次之后是两个进程 ...

  5. python IO非阻塞模型

    server端 import socket sk = socket.socket() sk.bind(('127.0.0.1', 8010)) sk.setblocking(False) # sk.l ...

  6. Python的Web编程&lbrack;0&rsqb; -&gt&semi; Web客户端&lbrack;1&rsqb; -&gt&semi; Web 页面解析

     Web页面解析 / Web page parsing 1 HTMLParser解析 下面介绍一种基本的Web页面HTML解析的方式,主要是利用Python自带的html.parser模块进行解析.其 ...

  7. Python的Web编程&lbrack;0&rsqb; -&gt&semi; Web客户端&lbrack;0&rsqb; -&gt&semi; 统一资源定位符 URL

    统一资源定位符 / URL 目录 URL 构成 URL 解析 URL 处理 1 URL构成 统一资源定位符(Uniform Resource Locator) 是对可以从互联网上得到的资源的位置和访问 ...

  8. Python的网络编程&lbrack;0&rsqb; -&gt&semi; socket&lbrack;0&rsqb; -&gt&semi; socket 与 TCP &sol; UDP

    Socket socket 简述 / socket Abstract 网络进程通信与 socket 网络中进程之间如何通信,首要解决的问题是如何唯一标识一个进程,否则通信无从谈起.在本地可以通过进程 ...

  9. Python的网络编程&lbrack;0&rsqb; -&gt&semi; socket&lbrack;2&rsqb; -&gt&semi; 利用 socket 建立 TCP&sol;UDP 通信

    Socket 目录 socket 的 TCP/IP 通信基本建立过程 socket 的 UDP 通信基本建立过程 socket 的 UDP 广播式通信基本建立过程 socket 的多线程通信建立过程 ...

随机推荐

  1. 20150912华为机考1之&quot&semi;输入一个字符串，将其中出现次数最多的字符输出&quot&semi;

    不吐槽华为的服务器了,直接上正文 输入:字符串(英文字母),长度不超过128 输出:出现频率最高的字母 思路写在注释文档 /* Input a string * Output the most fre ...

  2. 使用VS连接SQLServe时提示未能载入文件或程序集&OpenCurlyDoubleQuote;System&period;Data&period;OracleClient&comma; Version&equals;2&period;0&period;0&period;0&comma; Culture&equals;neutral&comma; PublicKey

    解决方法:         就是去微软主页下载两个Microsoft SQL Server 2012补丁包,SQLSysClrTypes.msi和SharedManagementObjects.msi ...

  3. &lbrack;0&period;0&rsqb;Analysis of Baidu search engine

    Rencently, my two teammates and I is doing a project, a simplified Chinese search engine for childre ...

  4. Latex及Beamer

    一 资源 Latex编辑部 Codecogs,latex在线编辑可见

  5. canvas绘图不清晰的解决方案

    现象描述 同样大小的图片(60x60px)用canvas和DOM绘制,结果发现canvas的画面质量要差很多.结果如下图所示. 上图中,左侧红框中的金币采用DOM绘制,右侧和下方的金币和文字等使用ca ...

  6. JS - 讨论 - 编码习惯 - JavaScript代码到底要不要写分号？

    如题:烦请大家在评论区给出原创意见!多谢!

  7. postgres 数据库的安装

    环境:Linux version 2.6.32-642.el6.x86_64   软件版本:postgresql-9.6.8.tar.gz 新项目要上线测试,要求安装一个PG 的数据库 我们进行的是源 ...

  8. 了解spring

    一.spring简介 Spring是一个JavaEE轻量级的一站式的开发框架(spring的可插拔特性,企业用于整合其他框架)轻量级:使用最少的代码启动程序,根据所需选择功能选择模块使用一站式:提供了 ...

  9. Spring boot 参数相关注解

    最近使用swagger的在线文档调试接口时发现老是报参数问题,最后发现是方法中参数上的注解有问题,今天把填的坑做一下总结. 1. RequestParam 该注解有两个属性: name/value:表 ...

  10. SocketAsyncEventArgs的释放问题

    起因是发现一个同事编写的程序运行两个月左右,占用了服务器20G左右的内存.用WinDbg查看发现存在大量的Async Pinned Handles,而它们的gcroot都来自于SocketAsyncE ...