协程(coroutine)是在Python编程中时常被提起的概念, 其的好处为人盛赞, 主要是CPU切换context的开销比进程, 线程都要小. Python中协程的技术原理是yield关键字与generator机制, 其的实现方式是让多个任务函数相互通过yield让出cpu执行权, 再通过send或者是next()+队列+sleep来返回原协程. 和一个实际游戏中多游戏世界同时运行的需求来探讨协程.
1. Python中协程的实现基础: yield与generator
协程的实现基础是python的generator机制和yield关键字.
generator机制: python中的generator机制指的是包含有yield关键字的generator function被执行后可以返回一个generator object的特性.
generator function: 指在函数体中写了yield关键字的一类函数--这类函数没有return, 只有yield. 其一旦被执行后返回的是一个generator object.
generator object: 被generator function返回的这个对象. 它能够传入作为入参被next(gen_obj)方法调用.
generator object本身属于iterator的一种. 而iterator又是对类似list, tuple, dict等实现了iter方法的iterable object进行遍历的一类对象.
关于iterable, iterator, generator相关知识, 可以更细致地参考这里https://anandology.com/python-practice-book/iterators.html
这里我提供了一个基本的例子来帮助理解yield和generator机制
# coding: utf-8
def create_generator(): # 这是一个generator function
mylist = [10, 20, 30]
for e in mylist:
yield e*e
def main():
g = create_generator() # 函数执行后执行的是yield而非return!
print g # 因此不会有常规函数的返回结果, 而是直接给出一个generator对象(g)
for val in g: # g是一个generator, 实现了next方法. 可以g.next() 或者 next(g)来调用.
print val
if __name__ == '__main__':
main()
利用yield和next, send实现协程: 生产者消费者问题
协程的实现的核心关键点是:
(1) cpu的使用权能够在不同的任务之间流转;
(2) 每个任务在切换出去和切换回来的时候能够正确保存和恢复上下文环境. 换句话说, 之前任务执行的进度要保留着.
只要能够做到这两点, 就是一个合格的协程.
我们可以用协程来实现一个经典的生产者消费者问题, 代码如下:
# coding: utf-8
""""代码参考了廖雪峰博客网站中的代码""""
import time
def consumer():
r = ''
while True:
n = yield r # 最重要的一行代码, 出去和回来都最在这里
if not n:
return
print('[CONSUMER] Consuming %s...' % n)
time.sleep(1)
r = '200 OK'
def produce(c):
c.next() # 一开始通过调用iterator的next方法来开始一个cycle. 注意这里第一个从consumer yield返回的r是被扔掉的.
n = 0
while n < 5: # Note: 协程最重要的特点, 与普通return函数调用的区别是, 它保存了函数内的临时变量的值, 也就是一个执行的进度, 而不是每次再进入该函数都重新创建!!!
n = n + 1
print('[PRODUCER] Producing %s...' % n)
r = c.send(n) # send 关键字使得我们可以向协程中的某个任务传递数据
print('[PRODUCER] Consumer return: %s' % r)
c.close()
if __name__=='__main__':
c = consumer()
produce(c)
代码中的注释比较详细, 通过PyCharm Debugger单步调试看一看程序是怎么走的, 结合注释后能对简单的协程有一个比较清晰的认识.
利用yield和next实现协程: 多个游戏世界同时运行问题
协程在实际中往往做的一个单进程多任务并行的应用.
假设我们要实现一个单进程但是能同时跑好多局游戏的服务器, 我们是可以利用协程+time.sleep()来实现CPU使用权的高效率流转.
这样的需求里, 每个正在进行的游戏局的小世界之间其实是完全相互独立的, 类似王者荣耀一样, 每个游戏世界之间不需要进行什么通信交流. 游戏小世界只有游戏初始化输入和结果输出的时候需要和大厅(或者是匹配服务器)做一个交互.
下面这份代码中, world.py表示的是我们游戏的小世界实例, 一局正在进行中的游戏就是一个小世界对象(小世界算是游戏开发中的一个术语). execution.py模块写的是我们协调执行各个world的调度代码, 并不复杂, 本质上就是一个队列, 一个任务类和一个装饰器. 最后, 我们还有一个demo.py, 里面就是放了一个让程序开始运行的main函数.
world.py
# coding: utf-8
"""运行中的游戏世界"""
import datetime
import time
class World(object):
serial_num = 0
def __init__(self):
World.serial_num += 1
self.id = World.serial_num
self.count = 0
self.game_over = False
def tick(self):
"""跑游戏每个turn逻辑的tick函数"""
self.count += 1
print '{} -- tick {} is done {} - time:{}'.format(self.id, self.count,
datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
time.time())
from coroutines.worlds_co.execution import loop_task
@loop_task
def run_world(self): # 是一个generator function
turn = 0
while not self.game_over:
turn += 1
print '-------Turn = w{}-{}--------'.format(self.id, turn)
self.tick()
yield 0 # 此处让出自己的执行权利
#由于我们不需要向调用generator的地方传递参数, 因此这里的yield [value]的value可以随便写一个占位置就行
execution.py
# coding=utf-8
import functools
import time
from collections import deque
class Executor(object):
"""
单例模式的Executor执行者
"""
instance = None
def __init__(self):
self.dq = deque()
@staticmethod
def get_instance():
if Executor.instance is None:
Executor.instance = Executor()
return Executor.instance
else:
return Executor.instance
def execute(self):
while True: # Executor的永恒循环, 取出第一个任务, 执行. 任务执行完后又会把自己塞回去deque尾部.
if len(self.dq) > 0:
task = self.dq[0]
task.run()
else:
print 'len = 0'
class Task(object):
def __init__(self, generator=None, delay=0.0):
super(Task, self).__init__()
self.generator = generator
self.delay = delay
if delay:
self.do_time = time.time() + delay
print 'self.do_time: ', self.do_time
def run(self):
if (not self.delay) or (self.do_time and self.do_time <= time.time()): # 如果时间到了, 就执行
print 'run...'
Executor.get_instance().dq.popleft() # 执行的任务要从队列DQ中移除
next(self.generator) # 此处让generator从上次执行的地方继续往下再走一个周期(一个周期=yield出一个值的周期)
next_task = Task(generator=self.generator, delay=0.4)
Executor.get_instance().dq.append(next_task)
else: # 否则直接睡到可以执行的那个时刻
sleep_time = self.do_time - time.time()
print 'sleep_time:', sleep_time
time.sleep(sleep_time)
def loop_task(func):
"""With executor, the decorated method becomes a looping task."""
# 这里work_func必须要由原先函数返回的instance_func而不能直接是func, 这是因为func是没有绑定过一个具体对象的, 而instance_func绑定了.
@functools.wraps(func)
def wrapper(*args, **kwargs): # 由于generator是一个while循环, 因此这个wrapper只会被调用一次, 后面循环内放task进入executor的工作是Task类完成的
generator = func(*args, **kwargs)
next_task = Task(generator=generator, delay=2.0)
Executor.get_instance().dq.append(next_task)
print 'dq len =', len(Executor.get_instance().dq)
return next_task # 记得要返回原先函数的结果
return wrapper
demo.py
import time
from coroutines.worlds_co.execution import Executor
from coroutines.worlds_co.world import World
if __name__ == '__main__':
w1 = World()
w2 = World()
w1.run_world()
time.sleep(0.2)
w2.run_world()
Executor.get_instance().execute()
不运用协程同样实现多个游戏世界同时运行问题
world.py
# coding: utf-8
"""运行中的游戏世界"""
import datetime
import time
class World(object):
serial_num = 0
def __init__(self):
World.serial_num += 1
self.id = World.serial_num
self.count = 0
def tick(self):
"""跑游戏每个turn逻辑的tick函数"""
self.count += 1
print '{} -- tick {} is done {} - time:{}'.format(self.id, self.count,
datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
time.time())
from coroutines.worlds_non_co.execution import loop_task
@loop_task
def run_world(self):
self.tick()
return self.run_world # 这里非常tricky, run_world需要能够返回出本函数, 且是绑定了一个instance实例的函数
execution.py
# coding=utf-8
import functools
import time
from collections import deque
class Executor(object):
"""
单例模式的Executor执行者
"""
instance = None
def __init__(self):
Executor.instance = self
self.dq = deque()
@staticmethod
def get_instance():
if Executor.instance is None:
Executor.instance = Executor()
return Executor.instance
else:
return Executor.instance
def execute(self):
while True:
# print 'executing...{}'.format(len(self.queue))
task = self.dq[0]
task.run()
class Task(object):
def __init__(self, work_func=None, delay=0.0):
super(Task, self).__init__()
self.work_func = work_func
self.delay = delay
if delay:
self.do_time = time.time() + delay
print 'self.do_time: ', self.do_time
def run(self):
if (not self.delay) or (self.do_time and self.do_time <= time.time()): # 如果时间到了, 就执行
Executor.get_instance().dq.popleft()
self.work_func()
else: # 否则再重新放回去队列中
sleep_time = self.do_time - time.time()
print 'sleep_time:', sleep_time
time.sleep(sleep_time)
def loop_task(func):
"""With executor, the decorated method becomes a looping task."""
# 这里work_func必须要由原先函数返回的instance_func而不能直接是func, 这是因为func是没有绑定过一个具体对象的, 而instance_func绑定了.
@functools.wraps(func)
def wrapper(*args, **kwargs):
instance_func = func(*args, **kwargs)
next_task = Task(work_func=instance_func, delay=2.0)
Executor.get_instance().dq.append(next_task)
return next_task # 记得要返回原先函数的结果
return wrapper
demo.py
# coding: utf-8
"""main: 不使用协程实现多个游戏世界同时运行"""
import time
from coroutines.worlds_non_co.execution import Executor
from coroutines.worlds_non_co.world import World
if __name__ == '__main__':
world1 = World()
world2 = World()
world1.run_world()
time.sleep(1.0)
world2.run_world()
Executor.get_instance().execute()
# Note: 后续可以改成executor一个线程, 网络收发一个线程, 由网络收发线程生成world, 并且执行world.run_world()这个最开始的启动.
通过这样一个协程执行框架, 我们可以在一个python进程中并行地跑很多个游戏小世界, 实现单进程多局游戏的业务需求. 在一台服务器上, 我们可以打开多个python进程(一个进程占一个socket), 每个进程又跑多个协程, 这样能够最大化利用多核心大内存服务器的性能.
不过, 这份执行框架的代码, 不是完美的. 由于采用deque做队列, 潜在的隐患是当deque队列中的所有任务被执行完使得队列为空之后, 再调用dq[0]会出现一个Empty的错误.
我专门写了一个Python Queue的改良版队列叫NiceQueue, 读者们可以参考使用NiceQueue来改进现在这个程序. https://github.com/imcheney/NiceQueue
