Python 3 非阻塞同步行为

Question

我正在使用Pygame 在python3 中制作经典的atari 蛇游戏。我想生成一个子进程来监听击键，这样每当玩家输入一个键（向上、向下、向左或向右）时，子进程都会向父进程发送该键。但是这个管道不应该被阻塞，这样蛇就可以沿着它行进的方向行进，直到收到钥匙。

我在multi-processes 上找到了 Python 的官方文档，但它没有描述我想要的行为，或者至少没有记录示例用法是否阻塞。谁能给我一个如何实现的例子？

Answer 1

你说：

我想为 AI 创建一个界面来控制蛇。如果在每次迭代 b/c 时将游戏状态简单地传递给 AI，这将是不公平的，那么它可能只需要它想要计算下一步的时间。因此，为什么它应该是同步的和非阻塞的。

所以要得到你想要的，你需要一个抽象。在下面的示例中，我创建了一个 Controller 类来执行此操作。 KeyboardController 处理键盘输入，而AsyncController 启动一个线程并使用Queue 类传递游戏状态和“AI”的决定。请注意，您必须在主线程上获取 pygame 事件，因此我在主循环中执行此操作，并将事件传递给控制器​​。

您的 AI 必须由 worker 函数调用。可以看到，目前worker函数中的“AI”每0.5秒才行动一次，而帧率是120。对于游戏来说，AI需要这么长的时间来做决定并不重要。

代码如下：

import pygame
import time
import random
from queue import Queue, Empty
from threading import Thread

class Controller():
    def __init__(self, color, message, actor):
        self.color = color
        self.message = message
        if actor: self.attach(actor)

    def attach(self, actor):
        self.actor = actor
        self.actor.controller = self
        self.actor.image.fill(self.color)

class AsyncController(Controller):
    def __init__(self, actor=None):
        super().__init__(pygame.Color('orange'), "AI is in control.", actor)
        self.out_queue = Queue()
        self.in_queue  = Queue()
        t = Thread(target=self.worker)
        t.daemon = True
        t.start()

    def update(self, events, dt):
        for e in events:
            if e.type == pygame.KEYDOWN:
                if e.key == pygame.K_SPACE: self.actor.controller = KeyboardController(self.actor)

        self.out_queue.put_nowait((self.actor, events, dt))
        try: return self.in_queue.get_nowait()
        except Empty: pass

    def worker(self):
        while True:
            try:
                actor, events, dt = self.out_queue.get_nowait()
                if actor.rect.x < 100: self.in_queue.put_nowait(pygame.Vector2(1, 0))
                if actor.rect.x > 600: self.in_queue.put_nowait(pygame.Vector2(-1, 0))
                if actor.rect.y < 100: self.in_queue.put_nowait(pygame.Vector2(0, 1))
                if actor.rect.y > 400: self.in_queue.put_nowait(pygame.Vector2(0, -1))
                if random.randrange(1, 100) < 15:
                    self.in_queue.put_nowait(random.choice([
                        pygame.Vector2(1, 0),
                        pygame.Vector2(-1, 0),
                        pygame.Vector2(0, -1), 
                        pygame.Vector2(0, 1)]))

                time.sleep(0.5)
            except Empty:
                pass

class KeyboardController(Controller):
    def __init__(self, actor=None):
        super().__init__(pygame.Color('dodgerblue'), "You're in control.", actor)

    def update(self, events, dt):
        for e in events:
            if e.type == pygame.KEYDOWN:
                if e.key == pygame.K_SPACE: self.actor.controller = AsyncController(self.actor)
                if e.key == pygame.K_UP: return pygame.Vector2(0, -1)
                if e.key == pygame.K_DOWN: return pygame.Vector2(0, 1)
                if e.key == pygame.K_LEFT: return pygame.Vector2(-1, 0)
                if e.key == pygame.K_RIGHT: return pygame.Vector2(1, 0)

class Actor(pygame.sprite.Sprite):
    def __init__(self):
        super().__init__()
        self.image = pygame.Surface((32, 32))
        self.image.fill(pygame.Color('dodgerblue'))
        self.rect = self.image.get_rect(center=(100, 100))
        self.direction = pygame.Vector2(1, 0)
        self.pos = self.rect.center

    def update(self, events, dt):
        new_direction = self.controller.update(events, dt)
        if new_direction:
            self.direction = new_direction
        self.pos += (self.direction * dt * 0.2)
        self.rect.center = self.pos

def main():
    pygame.init()

    actor   = Actor()
    sprites = pygame.sprite.Group(actor)
    screen  = pygame.display.set_mode([800,600])
    clock   = pygame.time.Clock()
    font    = pygame.font.SysFont("consolas", 20, True)
    dt      = 0
    KeyboardController(actor)

    while True:
        events = pygame.event.get()
        for e in events:
            if e.type == pygame.QUIT:
                return

        sprites.update(events, dt)
        screen.fill(pygame.Color('grey12'))
        screen.blit(font.render(actor.controller.message + ' [SPACE] to change to keyboard control.', True, pygame.Color('white')), (10, 10))
        sprites.draw(screen)
        dt = clock.tick(120)
        pygame.display.update()

if __name__ == '__main__':
    main()

请注意，此实现使用无限队列。您想添加一些逻辑来清除队列，这样您的游戏就不会占用大量内存。