使用 pygame 的多处理？
Using multiprocessing with pygame?
我试图在我用 pygame 制作的简单贪吃蛇游戏中将我的输入循环与我的游戏逻辑分开，但是，我真的很难弄清楚为什么当我运行程序。
我试过在子进程中导入 pygame，我检查了子进程中的错误，但一无所获。我查看了 google，但找不到任何可用的示例或类似问题。有人想过这些东西吗？
好的，这是代码：
import pygame
import time
import multiprocessing as mp
import random as rnd
pygame.init()
def event_to_dict(event: pygame.event) -> dict:
return {
'type': event.type,
'key': event.key if event.type == pygame.KEYDOWN else None,
}
class SnakeBoard:
def __init__(self, rows: int, columns: int):
self.rows = rows
self.columns = columns
self.vertices = []
self.odd_column = False
self.buff = []
for _ in range(self.rows):
self.buff.append([' ' for _ in range(self.columns)])
def initialize(self):
for r in range(self.rows):
for c in range(self.columns):
self.buff[r][c] = ' '
self.odd_column = (self.columns >> 1) % 2 == 1
self.buff[self.rows >> 1][self.columns >> 1] = '\u25cb'
self.vertices = [(self.rows >> 1, self.columns >> 1)]
def place_food(self):
while True:
r = rnd.randint(0, self.rows - 1)
c = rnd.randint(0, self.columns - 1)
codd = c % 2 == 1
if (codd and self.odd_column or not codd and not self.odd_column) and self.buff[r][c] != '\u25cb':
self.buff[r][c] = '\u25c9'
break
def tick(self, direction: int) -> bool:
nr, nc = self.vertices[-1]
if direction == 0:
nr -= 1
elif direction == 1:
nc += 1
elif direction == 2:
nr += 1
elif direction == 3:
nc -= 1
else:
print("Invalid direction for snake")
exit(1)
if nr >= self.rows or nc >= self.columns or nr < 0 or nc < 0 or self.buff[nr][nc] == '\u25cb':
return False
self.vertices.append((nr, nc))
self.vertices.pop(0)
return True
class SnakeGame(SnakeBoard):
def __init__(self, rows: int, columns: int):
super().__init__(rows, columns)
self.score = 0
self.direction = 0
self.initialize()
self.place_food()
def tick(self, direction: int = -1) -> bool:
v = super().tick(self.direction if direction < 0 else direction)
if self.buff[self.vertices[-1][0]][self.vertices[-1][1]] == '\u25c9':
self.score += 1
self.vertices.append(self.vertices[-1])
self.place_food()
for r in range(self.rows):
for c in range(self.columns):
if (r, c) in self.vertices:
self.buff[r][c] = '\u25cb'
elif self.buff[r][c] != '\u25c9' and self.buff[r][c] != ' ':
self.buff[r][c] = ' '
return v
class GameLoop(mp.Process):
def __init__(self, q: object, size: list):
super().__init__()
self.q = q
self.size = size
self.g = SnakeGame(size[1] // 10, size[0] // 10)
self.g.initialize()
self.g.place_food()
self.screen = None
self.game_surf = None
self.font = None
def run(self) -> None:
try:
import pygame
pygame.init()
self.screen = pygame.display.set_mode(self.size)
self.game_surf = pygame.Surface(self.size)
self.font = pygame.font.SysFont('roboto', 16)
is_running = True
while is_running:
if self.q.poll(0):
d = self.q.recv()
if d is not None:
if d['type'] == pygame.KEYDOWN:
if d['key'] == pygame.K_a:
self.g.direction = 3
elif d['key'] == pygame.K_s:
self.g.direction = 2
elif d['key'] == pygame.K_d:
self.g.direction = 1
elif d['key'] == pygame.K_w:
self.g.direction = 0
elif d['key'] == pygame.K_ESCAPE:
is_running = False
else:
is_running = False
self.game_surf.fill((255, 255, 255))
for ri, r in enumerate(self.g.buff):
for ci, c in enumerate(r):
if c == '\u25cb':
# print("Drawing a snake at {}, {}".format(ri * 10, ci * 10))
pygame.draw.circle(self.game_surf,
(0, 0, 255),
((ci * 10) + 5, (ri * 10) + 5),
5)
elif c == '\u25c9':
# wprint("Placing food at {}, {}".format(ci, ri))
pygame.draw.circle(self.game_surf,
(0, 127, 255),
((ci * 10) + 5, (ri * 10) + 5),
5)
timg = self.font.render("Score: {}, Level: {}".format(self.g.score, self.g.score // 10 + 1),
True,
(0, 0, 0))
self.screen.blit(self.game_surf, (0, 0))
self.screen.blit(timg, (0, 0))
pygame.display.flip()
if self.g.tick():
time.sleep(1 / ((int(self.g.score / 10 + 1)) * 10))
else:
timg = self.font.render("Game Over! Would you like to try again?", True, (0, 0, 0))
self.screen.blit(timg, ((self.size[0] >> 1) - 150, self.size[1] >> 1))
timg = self.font.render("Yes", True, (0, 0, 0))
btn_pos = ((self.size[0] >> 1) - 25, (self.size[1] >> 1) + 20)
self.screen.blit(timg, btn_pos)
pygame.display.flip()
while True:
event = pygame.event.wait()
if event.type == pygame.QUIT:
is_running = False
break
elif event.type == pygame.MOUSEBUTTONUP:
mx, my = pygame.mouse.get_pos()
if btn_pos[0] - 5 <= mx <= btn_pos[0] + 30 and btn_pos[1] - 5 <= my <= btn_pos[1] + 20:
self.g.initialize()
self.g.place_food()
self.g.score = 0
break
self.q.close()
except Exception as e:
print(e)
if __name__ == '__main__':
size = [800, 600]
parent, child = mp.Pipe()
p = GameLoop(child, size)
p.start()
running = True
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
elif event.type == pygame.KEYDOWN:
if event.key == pygame.K_ESCAPE:
running = False
ed = event_to_dict(event)
parent.send(ed)
parent.close()
p.join()
pygame.quit()
抱歉，有点奇怪，这是从控制台迁移到 pygame，所以一些逻辑仍在使用 unicode 符号。
通常在 GUI 应用程序中，通常希望将 GUI 与逻辑分开。
这样做有好处，因为这意味着您的 GUI 保持响应，即使您的逻辑
忙。但是，为了运行事物并发也有很多弊端，包括
开销。知道 python 不是 'thread safe' 也很重要，所以你可以打破
如果你不小心的话（见比赛条件）。
没有并发的简化示例
您的示例非常复杂，所以让我们从一个简单的示例开始：一个简单的 pygame 设置
一个移动的点
import pygame
import numpy as np
# Initialise parameters
#######################
size = np.array([800, 600])
position = size / 2
direction = np.array([0, 1]) # [x, y] vector
speed = 2
running = True
pygame.init()
window = pygame.display.set_mode(size)
pygame.display.update()
# Game loop
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
elif event.type == pygame.KEYDOWN:
if event.key == pygame.K_w:
direction = np.array([0, -1])
elif event.key == pygame.K_a:
direction = np.array([-1, 0])
elif event.key == pygame.K_s:
direction = np.array([0, 1])
elif event.key == pygame.K_d:
direction = np.array([1, 0])
position += direction * speed
if position[0] < 0 or position[0] > size[0] or position[1] < 0 or position[1] > size[1]:
running = False
pygame.time.wait(10) # Limit the speed of the loop
window.fill((0, 0, 0))
pygame.draw.circle(window, (0, 0, 255), position, 10)
pygame.display.update()
pygame.quit()
quit()
我们将从 gui 中分离出游戏逻辑
多重处理和其他选项：
因此 python 中的多处理允许您通过多个解释器同时利用多个内核。
虽然这听起来不错，但就 I/O 而言：它带来了更高的开销并且根本没有帮助（它可能会
伤害你的表现）。线程和异步都 运行 在单个内核上，即它们不是 'parrallel' 计算。但
他们允许的是在等待其他代码完成的同时完成代码。换句话说，您可以输入命令
虽然你的逻辑在其他地方 运行 很开心。
TLDR：作为一般规则：
- CPU绑定（100%的核心）程序：使用多处理，
- I/O绑定程序：使用线程或asyncio
线程版本
import pygame
import numpy as np
import threading
import time
class Logic:
# This will run in another thread
def __init__(self, size, speed=2):
# Private fields -> Only to be edited locally
self._size = size
self._direction = np.array([0, 1]) # [x, y] vector, underscored because we want this to be private
self._speed = speed
# Threaded fields -> Those accessible from other threads
self.position = np.array(size) / 2
self.input_list = [] # A list of commands to queue up for execution
# A lock ensures that nothing else can edit the variable while we're changing it
self.lock = threading.Lock()
def _loop(self):
time.sleep(0.5) # Wait a bit to let things load
# We're just going to kill this thread with the main one so it's fine to just loop forever
while True:
# Check for commands
time.sleep(0.01) # Limit the logic loop running to every 10ms
if len(self.input_list) > 0:
with self.lock: # The lock is released when we're done
# If there is a command we pop it off the list
key = self.input_list.pop(0).key
if key == pygame.K_w:
self._direction = np.array([0, -1])
elif key == pygame.K_a:
self._direction = np.array([-1, 0])
elif key == pygame.K_s:
self._direction = np.array([0, 1])
elif key == pygame.K_d:
self._direction = np.array([1, 0])
with self.lock: # Again we call the lock because we're editing
self.position += self._direction * self._speed
if self.position[0] < 0 \
or self.position[0] > self._size[0] \
or self.position[1] < 0 \
or self.position[1] > self._size[1]:
break # Stop updating
def start_loop(self):
# We spawn a new thread using our _loop method, the loop has no additional arguments,
# We call daemon=True so that the thread dies when main dies
threading.Thread(target=self._loop,
args=(),
daemon=True).start()
class Game:
# This will run in the main thread and read data from the Logic
def __init__(self, size, speed=2):
self.size = size
pygame.init()
self.window = pygame.display.set_mode(size)
self.logic = Logic(np.array(size), speed)
self.running = True
def start(self):
pygame.display.update()
self.logic.start_loop()
# any calls made to the other thread should be read only
while self.running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
self.running = False
elif event.type == pygame.KEYDOWN:
# Here we call the lock because we're updating the input list
with self.logic.lock:
self.logic.input_list.append(event)
# Another lock call to access the position
with self.logic.lock:
self.window.fill((0, 0, 0))
pygame.draw.circle(self.window, (0, 0, 255), self.logic.position, 10)
pygame.display.update()
pygame.time.wait(10)
pygame.quit()
quit()
if __name__ == '__main__':
game = Game([800, 600])
game.start()
那么取得了什么成果呢？
像这样轻巧的东西实际上不需要任何性能升级。但是，这确实允许的是
pygame GUI 将保持响应式，即使其背后的逻辑挂起。为了在行动中看到这一点，我们可以把逻辑
循环休眠，看看我们仍然可以移动 GUI、点击东西、输入命令等。
变化：
# Change this under _loop(self) [line 21]
time.sleep(0.01)
# to this
time.sleep(2)
# if we tried this in the original loop the program becomes glitchy
我试图在我用 pygame 制作的简单贪吃蛇游戏中将我的输入循环与我的游戏逻辑分开，但是，我真的很难弄清楚为什么当我运行程序。
我试过在子进程中导入 pygame，我检查了子进程中的错误，但一无所获。我查看了 google，但找不到任何可用的示例或类似问题。有人想过这些东西吗？
好的，这是代码：
import pygame
import time
import multiprocessing as mp
import random as rnd
pygame.init()
def event_to_dict(event: pygame.event) -> dict:
return {
'type': event.type,
'key': event.key if event.type == pygame.KEYDOWN else None,
}
class SnakeBoard:
def __init__(self, rows: int, columns: int):
self.rows = rows
self.columns = columns
self.vertices = []
self.odd_column = False
self.buff = []
for _ in range(self.rows):
self.buff.append([' ' for _ in range(self.columns)])
def initialize(self):
for r in range(self.rows):
for c in range(self.columns):
self.buff[r][c] = ' '
self.odd_column = (self.columns >> 1) % 2 == 1
self.buff[self.rows >> 1][self.columns >> 1] = '\u25cb'
self.vertices = [(self.rows >> 1, self.columns >> 1)]
def place_food(self):
while True:
r = rnd.randint(0, self.rows - 1)
c = rnd.randint(0, self.columns - 1)
codd = c % 2 == 1
if (codd and self.odd_column or not codd and not self.odd_column) and self.buff[r][c] != '\u25cb':
self.buff[r][c] = '\u25c9'
break
def tick(self, direction: int) -> bool:
nr, nc = self.vertices[-1]
if direction == 0:
nr -= 1
elif direction == 1:
nc += 1
elif direction == 2:
nr += 1
elif direction == 3:
nc -= 1
else:
print("Invalid direction for snake")
exit(1)
if nr >= self.rows or nc >= self.columns or nr < 0 or nc < 0 or self.buff[nr][nc] == '\u25cb':
return False
self.vertices.append((nr, nc))
self.vertices.pop(0)
return True
class SnakeGame(SnakeBoard):
def __init__(self, rows: int, columns: int):
super().__init__(rows, columns)
self.score = 0
self.direction = 0
self.initialize()
self.place_food()
def tick(self, direction: int = -1) -> bool:
v = super().tick(self.direction if direction < 0 else direction)
if self.buff[self.vertices[-1][0]][self.vertices[-1][1]] == '\u25c9':
self.score += 1
self.vertices.append(self.vertices[-1])
self.place_food()
for r in range(self.rows):
for c in range(self.columns):
if (r, c) in self.vertices:
self.buff[r][c] = '\u25cb'
elif self.buff[r][c] != '\u25c9' and self.buff[r][c] != ' ':
self.buff[r][c] = ' '
return v
class GameLoop(mp.Process):
def __init__(self, q: object, size: list):
super().__init__()
self.q = q
self.size = size
self.g = SnakeGame(size[1] // 10, size[0] // 10)
self.g.initialize()
self.g.place_food()
self.screen = None
self.game_surf = None
self.font = None
def run(self) -> None:
try:
import pygame
pygame.init()
self.screen = pygame.display.set_mode(self.size)
self.game_surf = pygame.Surface(self.size)
self.font = pygame.font.SysFont('roboto', 16)
is_running = True
while is_running:
if self.q.poll(0):
d = self.q.recv()
if d is not None:
if d['type'] == pygame.KEYDOWN:
if d['key'] == pygame.K_a:
self.g.direction = 3
elif d['key'] == pygame.K_s:
self.g.direction = 2
elif d['key'] == pygame.K_d:
self.g.direction = 1
elif d['key'] == pygame.K_w:
self.g.direction = 0
elif d['key'] == pygame.K_ESCAPE:
is_running = False
else:
is_running = False
self.game_surf.fill((255, 255, 255))
for ri, r in enumerate(self.g.buff):
for ci, c in enumerate(r):
if c == '\u25cb':
# print("Drawing a snake at {}, {}".format(ri * 10, ci * 10))
pygame.draw.circle(self.game_surf,
(0, 0, 255),
((ci * 10) + 5, (ri * 10) + 5),
5)
elif c == '\u25c9':
# wprint("Placing food at {}, {}".format(ci, ri))
pygame.draw.circle(self.game_surf,
(0, 127, 255),
((ci * 10) + 5, (ri * 10) + 5),
5)
timg = self.font.render("Score: {}, Level: {}".format(self.g.score, self.g.score // 10 + 1),
True,
(0, 0, 0))
self.screen.blit(self.game_surf, (0, 0))
self.screen.blit(timg, (0, 0))
pygame.display.flip()
if self.g.tick():
time.sleep(1 / ((int(self.g.score / 10 + 1)) * 10))
else:
timg = self.font.render("Game Over! Would you like to try again?", True, (0, 0, 0))
self.screen.blit(timg, ((self.size[0] >> 1) - 150, self.size[1] >> 1))
timg = self.font.render("Yes", True, (0, 0, 0))
btn_pos = ((self.size[0] >> 1) - 25, (self.size[1] >> 1) + 20)
self.screen.blit(timg, btn_pos)
pygame.display.flip()
while True:
event = pygame.event.wait()
if event.type == pygame.QUIT:
is_running = False
break
elif event.type == pygame.MOUSEBUTTONUP:
mx, my = pygame.mouse.get_pos()
if btn_pos[0] - 5 <= mx <= btn_pos[0] + 30 and btn_pos[1] - 5 <= my <= btn_pos[1] + 20:
self.g.initialize()
self.g.place_food()
self.g.score = 0
break
self.q.close()
except Exception as e:
print(e)
if __name__ == '__main__':
size = [800, 600]
parent, child = mp.Pipe()
p = GameLoop(child, size)
p.start()
running = True
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
elif event.type == pygame.KEYDOWN:
if event.key == pygame.K_ESCAPE:
running = False
ed = event_to_dict(event)
parent.send(ed)
parent.close()
p.join()
pygame.quit()
抱歉，有点奇怪，这是从控制台迁移到 pygame，所以一些逻辑仍在使用 unicode 符号。
通常在 GUI 应用程序中，通常希望将 GUI 与逻辑分开。 这样做有好处，因为这意味着您的 GUI 保持响应，即使您的逻辑 忙。但是，为了运行事物并发也有很多弊端，包括 开销。知道 python 不是 'thread safe' 也很重要，所以你可以打破 如果你不小心的话（见比赛条件）。
没有并发的简化示例
您的示例非常复杂，所以让我们从一个简单的示例开始：一个简单的 pygame 设置 一个移动的点
import pygame
import numpy as np
# Initialise parameters
#######################
size = np.array([800, 600])
position = size / 2
direction = np.array([0, 1]) # [x, y] vector
speed = 2
running = True
pygame.init()
window = pygame.display.set_mode(size)
pygame.display.update()
# Game loop
while running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
running = False
elif event.type == pygame.KEYDOWN:
if event.key == pygame.K_w:
direction = np.array([0, -1])
elif event.key == pygame.K_a:
direction = np.array([-1, 0])
elif event.key == pygame.K_s:
direction = np.array([0, 1])
elif event.key == pygame.K_d:
direction = np.array([1, 0])
position += direction * speed
if position[0] < 0 or position[0] > size[0] or position[1] < 0 or position[1] > size[1]:
running = False
pygame.time.wait(10) # Limit the speed of the loop
window.fill((0, 0, 0))
pygame.draw.circle(window, (0, 0, 255), position, 10)
pygame.display.update()
pygame.quit()
quit()
我们将从 gui 中分离出游戏逻辑
多重处理和其他选项：
因此 python 中的多处理允许您通过多个解释器同时利用多个内核。 虽然这听起来不错，但就 I/O 而言：它带来了更高的开销并且根本没有帮助（它可能会 伤害你的表现）。线程和异步都 运行 在单个内核上，即它们不是 'parrallel' 计算。但 他们允许的是在等待其他代码完成的同时完成代码。换句话说，您可以输入命令 虽然你的逻辑在其他地方 运行 很开心。
TLDR：作为一般规则：
- CPU绑定（100%的核心）程序：使用多处理，
- I/O绑定程序：使用线程或asyncio
线程版本
import pygame
import numpy as np
import threading
import time
class Logic:
# This will run in another thread
def __init__(self, size, speed=2):
# Private fields -> Only to be edited locally
self._size = size
self._direction = np.array([0, 1]) # [x, y] vector, underscored because we want this to be private
self._speed = speed
# Threaded fields -> Those accessible from other threads
self.position = np.array(size) / 2
self.input_list = [] # A list of commands to queue up for execution
# A lock ensures that nothing else can edit the variable while we're changing it
self.lock = threading.Lock()
def _loop(self):
time.sleep(0.5) # Wait a bit to let things load
# We're just going to kill this thread with the main one so it's fine to just loop forever
while True:
# Check for commands
time.sleep(0.01) # Limit the logic loop running to every 10ms
if len(self.input_list) > 0:
with self.lock: # The lock is released when we're done
# If there is a command we pop it off the list
key = self.input_list.pop(0).key
if key == pygame.K_w:
self._direction = np.array([0, -1])
elif key == pygame.K_a:
self._direction = np.array([-1, 0])
elif key == pygame.K_s:
self._direction = np.array([0, 1])
elif key == pygame.K_d:
self._direction = np.array([1, 0])
with self.lock: # Again we call the lock because we're editing
self.position += self._direction * self._speed
if self.position[0] < 0 \
or self.position[0] > self._size[0] \
or self.position[1] < 0 \
or self.position[1] > self._size[1]:
break # Stop updating
def start_loop(self):
# We spawn a new thread using our _loop method, the loop has no additional arguments,
# We call daemon=True so that the thread dies when main dies
threading.Thread(target=self._loop,
args=(),
daemon=True).start()
class Game:
# This will run in the main thread and read data from the Logic
def __init__(self, size, speed=2):
self.size = size
pygame.init()
self.window = pygame.display.set_mode(size)
self.logic = Logic(np.array(size), speed)
self.running = True
def start(self):
pygame.display.update()
self.logic.start_loop()
# any calls made to the other thread should be read only
while self.running:
for event in pygame.event.get():
if event.type == pygame.QUIT:
self.running = False
elif event.type == pygame.KEYDOWN:
# Here we call the lock because we're updating the input list
with self.logic.lock:
self.logic.input_list.append(event)
# Another lock call to access the position
with self.logic.lock:
self.window.fill((0, 0, 0))
pygame.draw.circle(self.window, (0, 0, 255), self.logic.position, 10)
pygame.display.update()
pygame.time.wait(10)
pygame.quit()
quit()
if __name__ == '__main__':
game = Game([800, 600])
game.start()
那么取得了什么成果呢？
像这样轻巧的东西实际上不需要任何性能升级。但是，这确实允许的是
pygame GUI 将保持响应式，即使其背后的逻辑挂起。为了在行动中看到这一点，我们可以把逻辑
循环休眠，看看我们仍然可以移动 GUI、点击东西、输入命令等。
变化：
# Change this under _loop(self) [line 21]
time.sleep(0.01)
# to this
time.sleep(2)
# if we tried this in the original loop the program becomes glitchy