大家好~我是米洛

我正在从0到1打造一个开源的接口测试平台, 也在编写一套与之对应的教程,希望大家多多支持。

欢迎关注我的公众号米洛的测开日记,获取最新文章教程!

回顾

上一节我们的前置条件支持了python脚本类型,让我们能够更灵活处理数据了。

今天我们就玩点好玩的,和大家一起探讨:怎么停止一段python代码。之后我们将会运用到pity之中。

知识要点

本文需要大家对asyncio相关知识有一些了解,至于原理方面,大家可以自行查阅,因为我也没细看。

为什么要做这个?

针对测试任务执行非常久的时候,看起来会是阻塞的情况,举个例子:

某个python脚本里面写了time.sleep(1000),导致你的case一直好像没有执行完成,那我们想结束它,该咋整呢?

今天博主要聊的就是这方面的内容。

先看看”难题”

我们来写一个简单的demo。

  • 我们先定义一个异步无限循环的方法,让它每隔一秒就打印一行内容
  1. import asyncio
  2. async def run():
  3. while True:
  4. print("still alive")
  5. await asyncio.sleep(1)

试想一下,如果这个方法开启了,不结束它,它是不是会一直打印下去?没错,我们先试试直接运行之:

测试平台系列(96) 如何停止测试任务执行 - 图1

很显然,它是绝对不会停止的,除非你关闭这个py程序。

我们把它想象成同步方法,是不是也会遇到这个困难:

  1. import time
  2. def run():
  3. time.sleep(10000)
  4. if __name__ == "__main__":
  5. run()
  6. print("done")

可以发现run一旦开始,它就阻塞了整个程序,下面的done必须要等run结束了才能打印出来。

怎么停止run

我们都知道,在python3.4以后新增了异步编程相关的概念,最初是由@coroutine这样的装饰器放到方法上,把方法标注为异步方法,后面直接从语言层面支持了异步方法定义(async),那其实里面还有很多我们不太常用的部分,比如今天要说的create_task。

  • 我们改写下方法
  1. import asyncio
  2. async def run():
  3. while True:
  4. print("still alive")
  5. await asyncio.sleep(1)
  6. async def main():
  7. await asyncio.create_task(run())
  8. if __name__ == "__main__":
  9. asyncio.run(main())

这次我们包了一层方法,利用asyncio.create_task来创建异步任务,create_task接收一个coroutine并执行。

执行之后,可以发现这个和上面的情况,任务也会一直进行下去。

我们继续下一步改造:

测试平台系列(96) 如何停止测试任务执行 - 图2

我们去掉await,可以看到run方法确实执行了,但是still alive只打印了一次就结束了。

我对它的理解是,虽然create_task创建了一个异步任务,但没说要await,也就是说没有说要等它结束。

想象一下这些异步任务都由一个事件循环控制,当你执行main(main本身也是一个异步任务)的时候,asyncio.run默认是要执行到main方法执行完毕的,也就是说,现在事件循环等待main方法执行完毕,main方法里面又创建了一个异步任务,但没有强调需要该任务完成,创建完毕后,由于main任务已经完成了,就导致整个事件结束了。

梳理一下:

  • 有await
    线程开启 -> 执行main -> main里面创建异步任务run -> 等待异步任务run(一直等一直等)
    由于主线程没有结束,所以整个python程序一直在等待异步任务执行完毕,毕竟它是死循环,所以会一直等下去。
  • 无await
    线程开启 -> 执行main -> main里面创建异步任务run -> 不等待异步任务run -> main方法结束 -> 线程结束 -> 程序退出
    以上都是个人结合go的goroutine给出的理解。肯定会有一些差别的地方。
  • 再次改造
    其实create_task会返回一个task对象,里面有done和cancel方法,也就是说咱们可以取消他也可以完成他。
  1. import asyncio
  2. async def run():
  3. while True:
  4. print("still alive")
  5. await asyncio.sleep(1)
  6. async def main():
  7. task = asyncio.create_task(run())
  8. await asyncio.sleep(2)
  9. # 2秒后,就停掉这个任务
  10. task.cancel()
  11. print("task任务结束了,main也即将完成,整个程序即将退出")
  12. if __name__ == "__main__":
  13. asyncio.run(main())

一旦调用了create_task,那么任务就已经开始了,接着我们用await让main等待2秒,再调用task.cancel方法就可以取消这个task,这样所有事件都结束,程序也会退出了。

看看gif:

测试平台系列(96) 如何停止测试任务执行 - 图3

这时候有的同学可能会问了,await2秒以后,不管你是否调用cancel,因为没有await task,所以程序照样会退出啊,没法证明task真的被cancel了

仔细想想,确实说的有道理。那我们再来改造下:

测试平台系列(96) 如何停止测试任务执行 - 图4

可以看到,这个task死之前还做了垂死挣扎!!!

  • 加上await

测试平台系列(96) 如何停止测试任务执行 - 图5

可以看到加上await之后,它还是不会停止

那我们加上cancel试试:

测试平台系列(96) 如何停止测试任务执行 - 图6

很遗憾,报错了~~~不过没关系,我们继续改下。只需要加上异常处理,就可以完美实现2秒后自动取消任务了。

测试平台系列(96) 如何停止测试任务执行 - 图7

但这个例子依然不是很帅,如果只是想控制异步任务的执行时间,那我们可以用wait_for:

  1. import asyncio
  2. async def run():
  3. while True:
  4. print("still alive")
  5. await asyncio.sleep(1)
  6. async def main():
  7. try:
  8. await asyncio.wait_for(run(), 2)
  9. except asyncio.TimeoutError:
  10. print("run方法超过2秒仍未执行完成")
  11. if __name__ == "__main__":
  12. asyncio.run(main())

这样2秒后,任务还没结束也不会继续执行了。

测试平台系列(96) 如何停止测试任务执行 - 图8

去掉死循环以后,任务会在2秒内完成(因为只等待1秒),这时候就可以看到timeout异常不会被触发


emmmm, 今天的内容就介绍到这了,异步还有挺多玩法的,我也不是很清楚,大家可以互相交流交流~~~