Python 3 非阻塞同步行为

问题描述

我正在使用

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

I'm making the classic atari snake game in python3 using Pygame. I want to spawn a subprocess to listen for key strokes so that whenever the player enters a key (UP, DOWN, LEFT, or RIGHT), the subprocess sends the parent process the key. But this pipe should not be blocking, so that the snake can travel in the direction it was traveling until the key is received.

I found Python's official documentation on multi-processes, but it does not describe the behavior I want, or at least doesn't document it as to whether if the example usages are blocking or not. Can someone give me an example of how this can be achieved?

解决方案

You said:

So to get what you want, you need an abstraction. In the example below, I created a Controller class that does that. KeyboardController handles keyboard input, while AsyncController starts a thread and uses the Queue class to pass the game state and the decision of the "AI" around. Note that you have to get pygame events on the main thread, so I do this in the main loop and simply pass the events down to the controller.

Your AI would have to be called by the worker function. As you can see, currently the "AI" in the worker function only acts every 0.5 second, while the framerate is 120. It doesn't matter to the game that the AI takes so long to make a decision.

Here's the code:

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

Note that this implementation uses infinite queues. You want to add some logic to clear the queues so your games does not use huge amounts of memory.

这篇关于Python 3 非阻塞同步行为的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！