迭代器
a = [1,2,3,45]
print(type(a))
a = iter(a) # 生成一个跌打器
print(type(a))
print(a.__iter__())
print(a)
class MyIterator(object):
"""迭代器对象是实现了__next__()方法和__iter__()方法的对象,其中他的__iter__()返回的是迭代器对象本身。
它只能实现一次迭代
"""
def __init__(self, start, end):
self.start = start
self.end = end
self.current = start
def __iter__(self):
return self
def __next__(self):
if self.current >= self.end:
raise StopIteration("haha")
self.current += 1
return self.current - 1
class MyIterable(object):
"""可迭代对象是实现了__iter__()方法的对象,且__iter__()返回一个迭代器对象。
要想实现多次迭代,只要可迭代对象返回一个新的迭代器即可,for循环时,python先调用__iter__()获得
迭代器,然后对该迭代器使用__next__()
for val in iterable:
print(val)
等效于:
iterator = iter(iterable) # 每次都是新的
while True:
try:
while True:
print(next(iterator))
except:
break
因为每次使用迭代工具迭代的时候都会调用__iter__()返回一个新的迭代器对象,这样就相当于创建多个迭代器了,自然可以看起来是重复迭代了!
"""
def __init__(self, start, end):
self.start = start
self.end = end
def __iter__(self):
return MyIterator(self.start, self.end)
if __name__ == "__main__":
# 迭代器
myIterator = MyIterator(3,5)
print(myIterator.__iter__() is myIterator)
for val in myIterator:
print(val)
# 迭代器对象不支持重新迭代
for val in myIterator:
print(val)
myIterator = MyIterator(3, 5)
print([i for i in myIterator])
# 迭代器对象不支持重新迭代
print([i for i in myIterator])
# 可迭代对象
myIterable = MyIterable(1,6)
print([i for i in myIterable])
# 可以重复迭代啦
print([i for i in myIterable])
iterator = iter(myIterable) # 每次都是新的
while True:
try:
while True:
print(next(iterator))
except StopIteration:
break
协程
from time import time as current_time
from random import random
event_list = []
class Event(object):
def __init__(self):
event_list.append(self)
self._callback = lambda: None
def is_ready(self):
ready = self._is_ready()
if ready:
self._callback()
return ready
def set_callback(self, callback):
self._callback = callback
class IOEvent(Event):
def __init__(self, duration):
super(IOEvent, self).__init__()
self._duration = duration
self._start_time = current_time()
def _is_ready(self):
return current_time() - self._start_time > self._duration
def testEvent():
e = IOEvent(3)
tic = current_time()
while not e.is_ready():
pass
toc = current_time()
print(toc - tic)
def io(name, times, duration=1):
for i in range(times):
yield IOEvent(duration)
print("{0} end: {1}".format(name, i))
def testIO():
print([event for event in io("yangkai", 3)])
# testIO()
class Dispatcher(object):
def __init__(self, tasks):
self._tasks = tasks
self._start()
def _start(self):
for task in self._tasks:
self._do(task)
def polling(self):
while len(event_list) > 0:
for e in event_list:
if e.is_ready():
event_list.remove(e)
break
def _do(self, task):
try:
event = next(task)
event.set_callback(lambda: self._do(task))
except StopIteration:
pass
def testDispatcher():
dispatcher = Dispatcher([io('yangkai', 3), io('Lizhen', 2), io('shushao', 4)])
dispatcher.polling()
import timeit
timeit_times = 10
avg_cost = timeit.timeit(lambda: testDispatcher(), number=timeit_times) / timeit_times
print('%.3f' % avg_cost)
网友评论