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 ide pycharm.
• 下一篇 立法會五題：為自閉症兒童提供的支援服務 2012.