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)与运行

常用python日期日志获取内容循环的代码片段

Kotlin 协程单元测试错误:线程“main @coroutine#1 @coroutine#2”中的异常 java.lang.NullPointerException

Python 协程 - Coroutines

python3 coroutine

Python 协程(Coroutine)体验