Python多进程中找到目标结果后如何终止所有进程？

方法1：使用共享事件（Event）传递终止信号 #

multiprocessing.Event是进程间通信的常用工具，可以传递"终止"信号。每个进程在循环猜PIN前先检查事件是否被触发，一旦触发就立即退出；当某个进程找到正确PIN时，触发事件再退出，其他进程会很快检测到并停止。

修改后的代码如下：

import requests
import multiprocessing
import time
import sys

def check_pin_range(start, end, stop_event, start_time):
    # 处理0-999的补零情况
    for n in range(start, end):
        # 先检查是否需要终止
        if stop_event.is_set():
            return
        guess = f'{n:04}' if start == 0 else str(n)
        try:
            answer = requests.post('https://www.guessthepin.com/prg.php', data={'guess': guess})
            answer.raise_for_status()  # 捕获请求异常
        except requests.exceptions.RequestException as e:
            print(f"请求出错: {e}")
            continue
        
        if "Holy Moley!" in answer.text:
            print(f"THE PIN IS: {guess}")
            end_time = time.time()
            print(f"Time taken: {(end_time - start_time) / 60:.2f} minutes")
            # 设置终止信号，通知其他进程停止
            stop_event.set()
            return
        print(f"The pin is not {guess}")

if __name__ == "__main__":
    start_time = time.time()
    # 创建共享终止事件
    stop_event = multiprocessing.Event()
    
    # 定义四个进程的范围，统一用一个函数处理，避免重复代码
    processes = []
    ranges = [(0, 1000), (1000, 4000), (4000, 7000), (7000, 10000)]
    for start, end in ranges:
        p = multiprocessing.Process(
            target=check_pin_range,
            args=(start, end, stop_event, start_time)
        )
        processes.append(p)
        p.start()
    
    # 等待任意一个进程触发终止事件，或者所有进程结束
    for p in processes:
        p.join()
        # 如果事件已设置，终止其他进程
        if stop_event.is_set():
            for other_p in processes:
                if other_p.is_alive():
                    other_p.terminate()
                    other_p.join()
            break
    
    print("Done!")
    end_time = time.time()
    print(f"Total time taken: {(end_time - start_time)/60:.2f} minutes")

代码说明： #

• 用check_pin_range统一处理所有范围的PIN检查，避免重复代码
• stop_event作为共享参数传递给每个进程，进程每次循环前检查is_set()，如果为True就立即退出
• 当找到正确PIN时，调用stop_event.set()触发终止信号，其他进程会在下一次循环时检测到并停止
• 主进程在join时，一旦发现事件被设置，就主动终止还在运行的子进程，确保快速停止

方法2：主进程主动终止子进程 #

如果不想用Event，也可以让主进程监控子进程的状态，当某个子进程找到PIN后，直接终止其他所有子进程。可以通过共享布尔变量标记是否找到PIN。

示例代码：

import requests
import multiprocessing
import time
import sys

# 共享布尔变量，标记是否找到PIN
found = multiprocessing.Value('b', False)

def check_pin_range(start, end, found_flag, start_time):
    for n in range(start, end):
        if found_flag.value:
            return
        guess = f'{n:04}' if start == 0 else str(n)
        answer = requests.post('https://www.guessthepin.com/prg.php', data={'guess': guess})
        if "Holy Moley!" in answer.text:
            print(f"THE PIN IS: {guess}")
            end_time = time.time()
            print(f"Time taken: {(end_time - start_time) / 60:.2f} minutes")
            found_flag.value = True
            return
        print(f"The pin is not {guess}")

if __name__ == "__main__":
    start_time = time.time()
    processes = []
    ranges = [(0, 1000), (1000, 4000), (4000, 7000), (7000, 10000)]
    for start, end in ranges:
        p = multiprocessing.Process(
            target=check_pin_range,
            args=(start, end, found, start_time)
        )
        processes.append(p)
        p.start()
    
    # 循环检查是否找到PIN
    while not found.value:
        time.sleep(0.1)
        # 检查是否所有进程都结束了（没找到的情况）
        all_done = all(not p.is_alive() for p in processes)
        if all_done:
            break
    
    # 终止所有还在运行的进程
    for p in processes:
        if p.is_alive():
            p.terminate()
            p.join()
    
    print("Done!")
    end_time = time.time()
    print(f"Total time taken: {(end_time - start_time)/60:.2f} minutes")

额外优化： #

• 修复了原代码中four函数打印未补零数字的问题，统一输出实际请求的四位PIN
• 方法1中加入了请求异常处理，避免网络问题导致进程卡住
• 移除了无用的全局start_time，改为直接在主进程计算总时间，结果更准确

内容的提问来源于stack exchange，提问作者Rohan P