将多进程与PYGAME结合使用？

Using multiprocessing with pygame?(将多进程与PYGAME结合使用？)

问题描述

在我用pyGame制作的简单的蛇游戏中，我试图将输入循环与游戏逻辑分开，但是，我真的很难弄清楚为什么在运行程序时没有发生任何事情。

我尝试在子进程中导入pyGame，检查子进程中的错误，但一无所获。我在谷歌上查了一下，但找不到任何有用的例子，也找不到类似的问题。有人想过这些东西吗？

好的，代码如下：

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符号。

图形用户界面

通常在推荐答案应用程序中，希望将图形用户界面与逻辑分开是很常见的。 这样做有好处，因为这意味着即使您的逻辑 很忙。然而，为了同时运行，有许多缺点，包括 管理费用。同样重要的是要知道，Python不是"线程安全的"，所以您可以中断 如果你不小心的话(见竞争条件)。

无并发的简化示例

您的示例相当复杂，所以让我们从一个简单的示例开始： 移动点

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()

我们将把游戏逻辑从图形用户界面中分离出来

突变处理和其他选项：

因此，在Python中的多处理允许您通过多个解释器同时利用多个核心。 虽然这听起来不错，但就I/O而言：它带来了更高的管理费用，而且根本没有帮助(它很可能 损害了你的表现)。线程和异步都在一个内核上运行，也就是说它们不是"并行"计算。但 它们所允许的是在等待其他代码完成的同时完成代码。换句话说，您可以输入命令 当你的逻辑在别处快乐地运行时。

TLDR：一般规则：

• CPU绑定(100%内核)程序：使用多处理，
• I/O受限程序：使用线程或异步

线程版本

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图形用户界面也将保持被动反应。要在行动中看到这一点，我们可以将逻辑 循环进入睡眠状态，然后查看我们仍然可以移动图形用户界面、点击内容、输入命令等。
更改：

# 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结合使用？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！