在前面几条里,我们以生命游戏为例,试着用各种方案解决I/O并行问题,这些方案在某些情况下确实可行,但如果同时需要执行的I/O任务有成千上万个,那么这些方案的效率就不太理想了
像这种在并发方面要求比较高的I/O需求,可以用Python的协程(coroutine)来解决。协程能够制造出一种效果,让我们觉得Python程序好像真的可以同时执行大量任务。这种效果需要使用async与await关键字来实现,它的基本原理与生成器(generator)类似,也就是不立刻给出所有的结果,而是等需要用到的时候再一项一项地获取
启动协程是有代价的,就是必须做一次函数调用。协程激活之后,只占用不到1KB内存,所以只要内存足够,协程稍微多一些也没关系。与线程类似,协程所要执行的任务也是用一个函数来表示的,在执行这个函数的过程中,协程可以从执行环境里面获取输入值,并把输出结果放到这个执行环境之中。协程与线程的区别在于,它不会把这个函数从头到尾执行完,而是每遇到一个await表达式,就暂停一次,下次继续执行的时候,它会先等待await所针对的那项awaitable操作有了结果(那项操作是用async函数表示的),然后再推进到下一个await表达式那里(这跟生成器函数的运作方式有点像,那种函数也是一遇到yield就暂停)。
Python系统可以让数量极多的async函数各自向前推进,看起来像很多条Python线程那样,能够并发地运行。然而,这些协程并不会像线程那样占用大量内存,启动和切换的开销也比较小,而且不需要用复杂的代码来实现加锁或同步。这种强大的机制是通过事件循环(event loop)打造的,只要把相关的函数写对,这种循环就可以穿插着执行许多个这样的函数,并且执行得相当快,从而高效地完成并发式的I/O任务。
现在就用协程来实现生命游戏。我们的目标是让游戏能够高效地执行game_logic函数里面的I/O操作,同时又不像前面提到的Thread方案与Queue方案那样,有那么多缺点。首先修改game_logic函数,这次必须在定义函数所用的那个def关键字前面,加上async,表示该函数是一个协程,这样我们就可以在函数里面用await做I/O了(例如从套接字(socket)之中异步读取一份数据)。同理,给step_cell函数也添上async关键字,把它变为协程,并在调用game_logic的那个地方使用await关键字。

ALIVE = '*'
EMPTY = '-'
class Grid:
def __init__(self, height, width):
self.height = height
self.width = width
self.rows = []
for _ in range(self.height):
self.rows.append([EMPTY] * self.width)
def get(self, y, x):
return self.rows[y % self.height][x % self.width]
def set(self, y, x, state):
self.rows[y % self.height][x % self.width] = state
def __str__(self):
output = ''
for row in self.rows:
for cell in row:
output += cell
output += '\n'
return output
def count_neighbors(y, x, get):
n_ = get(y - 1, x + 0) # North
ne = get(y - 1, x + 1) # Northeast
e_ = get(y + 0, x + 1) # East
se = get(y + 1, x + 1) # Southeast
s_ = get(y + 1, x + 0) # South
sw = get(y + 1, x - 1) # Southwest
w_ = get(y + 0, x - 1) # West
nw = get(y - 1, x - 1) # Northwest
neighbor_states = [n_, ne, e_, se, s_, sw, w_, nw]
count = 0
for state in neighbor_states:
if state == ALIVE:
count += 1
return count
# async def game_logic(state, neighbors):
# # Do some input/output in here:
# data = await my_socket.read(50)
async def game_logic(state, neighbors):
if state == ALIVE:
if neighbors < 2:
return EMPTY # Die: Too few
elif neighbors > 3:
return EMPTY # Die: Too many
else:
if neighbors == 3:
return ALIVE # Regenerate
return state
async def step_cell(y, x, get, set):
state = get(y, x)
neighbors = count_neighbors(y, x, get)
next_state = await game_logic(state, neighbors)
set(y, x, next_state)
simulate函数也同样需要变为协程。
import asyncio
async def simulate(grid):
next_grid = Grid(grid.height, grid.width)
tasks = []
for y in range(grid.height):
for x in range(grid.width):
task = step_cell(
y, x, grid.get, next_grid.set) # Fan out
tasks.append(task)
await asyncio.gather(*tasks) # Fan in
return next_grid
async版本的simulate函数,有以下几个地方需要解释:
▪ 第一,它在调用step_cell的时候,系统并不会立刻执行这个函数,而是会返回一个协程实例,稍后会把这个实例写在await表达式里面。这里的step_cell,好比那种用yield写的生成器函数一样,调用时并不立刻执行它,而是返回一个生成器实例。这样就可以实现任务fan-out(分派)模式了。
▪ 第二,内置的asyncio模块提供了gather函数,可以用来实现fan-in(归集)模式。把gather写在await表达式里面,可以让系统用事件循环去并发地执行那些step_cell协程,并在全部执行完之后,再往下推进simulate协程。
▪ 第三,由于这些代码都是在同一条线程里执行的,因此不需要给Grid(网格)实例加锁,至于怎样让这些I/O操作表现出平行的效果,则是由asyncio所提供的事件循环来负责的。
最后,要调整原范例之中用来推动游戏流程的那段代码。我们只需要修改一行代码,也就是把simulate(grid)这个协程交给asyncio.run去运行,从而利用事件循环机制去执行推进单元格状态所需的那些I/O操作。
class ColumnPrinter:
def __init__(self):
self.columns = []
def append(self, data):
self.columns.append(data)
def __str__(self):
row_count = 1
for data in self.columns:
row_count = max(
row_count, len(data.splitlines()) + 1)
rows = [''] * row_count
for j in range(row_count):
for i, data in enumerate(self.columns):
line = data.splitlines()[max(0, j - 1)]
if j == 0:
padding = ' ' * (len(line) // 2)
rows[j] += padding + str(i) + padding
else:
rows[j] += line
if (i + 1) < len(self.columns):
rows[j] += ' | '
return '\n'.join(rows)
logging.getLogger().setLevel(logging.ERROR)
grid = Grid(5, 9)
grid.set(0, 3, ALIVE)
grid.set(1, 4, ALIVE)
grid.set(2, 2, ALIVE)
grid.set(2, 3, ALIVE)
grid.set(2, 4, ALIVE)
columns = ColumnPrinter()
for i in range(5):
columns.append(str(grid))
grid = asyncio.run(simulate(grid)) # Run the event loop
print(columns)
协程的优点是,能够把那些与外部环境交互的代码(例如I/O调用)与那些实现自身需求的代码(例如事件循环)解耦。这让我们可以把重点放在实现需求所用的逻辑上面,而不用专门花时间去写一些代码来确保这些需求能够并发地执行。
我们同样按之前的测一下性能。
配置 性能(s) | 单线程 | 线程方案实现多线程 | 队列方案实现多线程 | ThreadPoolExecutor方案实现多线程 | 协程方式 |
Grid(500,900) step 100次 线程数 5 | 55.45792198181152 | 170.44810271263123 | 6410.6107659339905 | 382.2381818294525 | |
Grid(50,90) step 100次 线程数 5 | 0.6782510280609131 | 14.29249095916748 | 2.5422348976135254 | 5.175195217132568 | 2.2674009799957275 |