pyuv+python35的超简化异步编程方案，参考asyncio
jopen
9年前
实现原理类似参考：
测试代码：
import pyuv import sys import functools class Futear: _PENDING = 0 _CANCELLED = 1 _FINISHED = 2 def __init__(self): self._result_args = [] self._result_kwargs = {} self._state = Futear._PENDING self._coro = Task._current_task._coro if Task._current_task else None #todo: assert self._coro is not None def cancel(self): self._state = Futear._CANCELLED def set_result(self, *args): self._result_args = args self._state = Futear._FINISHED if self._coro is not None: try: self._coro.send(None) except StopIteration as si: pass def result(self): if self._result_kwargs: return (*self._result_args, self._result_kwargs) else: return (*self._result_args, ) def is_pending(self): return self._state == Futear._PENDING def is_done(self): return self._state == Futear._FINISHED def is_running(self): return self._state == Futear._PENDING def __iter__(self, *args): if self.is_running(): yield self return self.result() # May raise too. if sys.version_info >= (3, 5): __await__ = __iter__ # make compatible with 'await' expression pass class Task(Futear): #_tasks = [] _current_task = None def __init__(self, coro, timeout=None): super().__init__() Task.add_task(self) self._state = None #self._loop = loop if coro is not None: self._coro = coro self._start(timeout) def _start(self, timeout=None): if self._coro is not None: prev_task = Task._current_task Task._current_task = self try: self._state = Futear._PENDING self._coro.send(None) except StopIteration as si: pass finally: Task._current_task = prev_task pass @classmethod def add_task(cls, task): #cls._tasks.append(task) pass loop = pyuv.Loop() async def sleep(sec): timer = pyuv.Timer(loop) waiter = Futear() def callback(result, timer_handle): timer_handle.stop() waiter.set_result(result) cb = functools.partial(callback, "abc") timer.start(cb, sec, sec) return await waiter; async def fork(x): print("sleeping...") res = await sleep(x) print("wakeup! Has got a result:%s" % res) def main(): t = Task(fork(1)) print("after creating task.") if __name__ == "__main__": main() loop.run()
1，异常处理
2，运行时信息处理
3，使用统一方式封装libuv。