Python国际象棋引擎：如何在搜索时即时响应stop命令？

你当前的核心问题是主循环被search()函数阻塞，导致stop命令必须等当前深度的搜索完全结束后才能被处理。要实现即时响应，必须让命令处理和搜索逻辑并发执行，下面是几个实用的解决方案：

方案1：用线程分离搜索与命令处理 #

把搜索逻辑放到独立线程中，主循环专注监听控制台命令，这样stop命令可以直接设置终止标志，让搜索线程即时停止。

示例代码：

import threading

# 全局终止标志（也可以封装到类中，更优雅）
stop_search = False

def search(max_depth):
    global stop_search
    current_depth = 1
    while current_depth <= max_depth and not stop_search:
        # 执行当前深度的搜索逻辑（比如Alpha-Beta剪枝）
        print(f"正在搜索深度 {current_depth}")
        # 模拟搜索耗时
        import time
        time.sleep(1)
        current_depth += 1
    print("搜索已终止")

# 主命令处理循环
while True:
    command = input().strip()
    if command == "go":
        stop_search = False
        # 启动搜索线程，不阻塞主循环
        threading.Thread(target=search, args=(10,), daemon=True).start()
    elif command == "stop":
        stop_search = True
        print("已触发终止命令")

注意：如果你的搜索逻辑包含递归（比如经典的Alpha-Beta递归实现），必须在每一层递归中检查stop_search标志，否则会等当前递归栈走完才会终止。

方案2：用生成器实现可中断的迭代式搜索 #

把search()改写成生成器，每完成一层深度搜索就yield一次，主循环可以在每次yield间隙检查新命令，无需线程也能实现中断。

示例代码：

import sys
import select

def search_generator(max_depth):
    current_depth = 1
    while current_depth <= max_depth:
        # 执行当前深度的搜索逻辑
        print(f"正在搜索深度 {current_depth}")
        # 模拟搜索耗时
        import time
        time.sleep(1)
        yield current_depth
        current_depth += 1

# 主命令循环
active_search = None
while True:
    # 非阻塞读取控制台输入，避免被input()阻塞
    if select.select([sys.stdin], [], [], 0.1)[0]:
        command = sys.stdin.readline().strip()
        if command == "go":
            active_search = search_generator(10)
            print("搜索启动")
        elif command == "stop":
            active_search = None
            print("搜索已终止")
    # 推进搜索进度
    if active_search is not None:
        try:
            next(active_search)
        except StopIteration:
            active_search = None
            print("搜索完成")

这个方案不需要线程，但需要把递归式搜索改成迭代式，同时处理非阻塞输入，确保主循环不会卡在等待命令的状态。

方案3：用系统信号（仅Unix/Linux适用） #

在类Unix系统下，可以通过信号机制触发终止，不过这种方式兼容性较差，Windows无法使用。

示例代码：

import signal
import os

stop_search = False

def handle_stop_signal(signum, frame):
    global stop_search
    stop_search = True
    print("收到终止信号")

# 注册自定义信号处理函数
signal.signal(signal.SIGUSR1, handle_stop_signal)

def search(max_depth):
    global stop_search
    current_depth = 1
    while current_depth <= max_depth and not stop_search:
        print(f"正在搜索深度 {current_depth}")
        import time
        time.sleep(1)
        current_depth += 1

# 主循环
while True:
    command = input().strip()
    if command == "go":
        stop_search = False
        search(10)
    elif command == "stop":
        # 向当前进程发送终止信号
        os.kill(os.getpid(), signal.SIGUSR1)

关键注意事项 #

无论选择哪种方案，核心都是让搜索逻辑定期检查终止标志，尤其是在耗时的子过程或递归调用中，必须插入检查点，才能真正实现即时终止。如果使用线程，注意共享数据的线程安全（比如局面数据的读写）。

内容的提问来源于stack exchange，提问作者Boo Who