python - Python 3 非阻塞同步行为
问题描述
我正在使用Pygame在 python3 中制作经典的 atari 蛇游戏。我想生成一个子进程来监听击键,这样每当玩家输入一个键(向上、向下、向左或向右)时,子进程都会向父进程发送该键。但是这个管道不应该被阻塞,这样蛇就可以沿着它行进的方向行进,直到收到钥匙。
我找到了 Python 关于multi-processes的官方文档,但它没有描述我想要的行为,或者至少没有记录示例用法是否阻塞。有人可以给我一个如何实现这一目标的例子吗?
解决方案
你说:
我想为 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()
请注意,此实现使用无限队列。您想添加一些逻辑来清除队列,这样您的游戏就不会使用大量内存。
推荐阅读
- flutter - 如何使用 BuildContext 判断页面是否已被释放
- sql - PostgreSQL 中的无方向关系失败
- java - 使用嵌入式 H2 的 Spring Boot 测试执行 10 倍于 init 脚本
- sql - 使用 SQL 查询从新列中的表中提取值
- python - 将 curl 转换为 python 请求,包括 cookie
- python - 如何在给定条件下选择两个数组的元素?
- react-native - react-native-chart-kit 条形图图形切割
- node.js - 猫鼬试图填充
- mongodb - MongoDB 索引文本搜索仅适用于完全匹配
- java - Java 错误:在 Main 方法中调用不同的类但在使用 main 方法编译相同的类时运行良好