python coroutine测试
Posted
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python coroutine测试相关的知识,希望对你有一定的参考价值。
目的:实现一个类似于asyn await的用法,来方便的编写callback相关函数
from __future__ import print_function
import time
import threading
def asyn_task(callback):
def SleepTask():
time.sleep(1)
callback("world")
# t = threading.Timer(1.0, lambda :callback("world"))
t = threading.Thread(target=SleepTask)
t.start() # after 30 seconds, "hello, world" will be printed
def main():
asyn_task(lambda s:print(s))
print ("hello")
def main2():
class myCoroutine:
def __init__(self):
self.evt = threading.Event()
self.co=self.myCOroutine()
self.co.next()
def myCOroutine(self):
while 1:
self.evt.clear()
line = (yield)
# print (‘yield:‘ + line)
self.line = line
self.evt.set()
def await(self):
self.evt.wait()
return self.line
def send(self, *args):
self.co.send(*args)
g = myCoroutine()
asyn_task(lambda s:g.send(s))
print ("hello")
s = g.await()
print (s)
main2()
以上是关于python coroutine测试的主要内容,如果未能解决你的问题,请参考以下文章
python asyncio 异步 I/O - 协程(Coroutine)与运行
Kotlin 协程单元测试错误:线程“main @coroutine#1 @coroutine#2”中的异常 java.lang.NullPointerException