Python多进程下使用ctypes更新控制台标题避免重置的方法

数据老陈

2026-9-1

问题原因

你现在遇到的标题重置、计数不准问题本质是两个原因：

Windows平台multiprocessing默认采用spawn方式创建子进程，每个子进程拥有独立的内存空间，你定义的remain_scan/total_finish_global这类全局变量会在子进程启动时被拷贝一份，子进程里修改的只是自己内存里的副本，既不会同步到主进程，也不会和其他子进程共享，计数自然累加不准。
控制台窗口归属主进程，你让20个子进程同时调用SetConsoleTitleW修改同一个窗口的标题，多个进程并发写就会互相覆盖，出现标题跳变、被重置的现象。

实现思路

彻底解决的核心原则很简单：子进程只做扫描任务，所有计数累加、控制台标题更新的操作全部放到主进程统一执行。
具体落地逻辑：

新增专门的进度回传队列，子进程每处理完一个站点，就把处理结果（站点是否可访问、是否命中特征）塞进这个队列，子进程代码里完全删掉所有调用SetConsoleTitleW的逻辑，绝对不碰控制台标题。
所有统计变量（总站点数、有效站点数、剩余扫描量、已完成数、命中数）全部在主进程里维护，主进程循环读取进度队列的消息，每收到一条结果就更新对应计数，然后调用一次接口更新控制台标题，全程只有主进程改标题，根本不会出现被重置的问题。

完整参考代码

import multiprocessing
import ctypes
import requests
from requests.packages.urllib3.exceptions import InsecureRequestWarning
# 关闭SSL证书校验警告
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

NUM_PROCESSES = 20
# 仅在主进程加载控制台标题修改接口
set_console_title = ctypes.windll.kernel32.SetConsoleTitleW

def do_work(site, progress_queue):
    """子进程执行实际扫描逻辑，仅回传结果，不操作控制台"""
    is_valid = False
    is_hit = False
    try:
        resp = requests.head(site, timeout=3, stream=True, verify=False, allow_redirects=True)
        final_url = resp.url
        is_valid = resp.status_code < 400
        # 按需拉取页面内容判断特征
        if is_valid:
            page_resp = requests.get(final_url, timeout=5, verify=False, allow_redirects=True)
            if "LORE_IPSIUM" in page_resp.text:
                is_hit = True
                print(f"[+] 命中目标站点: {site}")
    except Exception:
        # 扫描失败直接记为无效站点即可
        pass
    # 把结果回传到主进程
    progress_queue.put((is_valid, is_hit))


def worker_main(task_queue, progress_queue):
    while True:
        item_site = task_queue.get(block=True)
        if item_site is None:
            # 收到进程退出信号，终止循环
            break
        do_work(item_site, progress_queue)


def main():
    # 替换为你实际的待扫描站点列表
    list_site_pars = [f"https://test{i}.com" for i in range(200)]
    total_site = len(list_site_pars)
    # 所有统计变量仅在主进程维护
    valid_count = 0
    finished_count = 0
    hit_count = 0
    remain_count = total_site

    task_queue = multiprocessing.Queue()
    progress_queue = multiprocessing.Queue()
    # 初始化进程池
    the_pool = multiprocessing.Pool(NUM_PROCESSES, worker_main, (task_queue, progress_queue))
            
    # 塞入所有待扫描任务
    for item_site in list_site_pars:
        task_queue.put(item_site)
    # 给每个进程塞退出信号
    for i in range(NUM_PROCESSES):
        task_queue.put(None)

    # 主进程循环读取进度，更新标题
    while finished_count < total_site:
        try:
            is_valid, is_hit = progress_queue.get(timeout=1)
            remain_count -= 1
            finished_count += 1
            if is_valid:
                valid_count += 1
            if is_hit:
                hit_count += 1
            # 仅主进程更新控制台标题，无并发冲突
            set_console_title(
                f'ALL SITE: {total_site}  |  SITE VALID: {valid_count}  |  REMAIN: {remain_count}  |  FINISHED: {finished_count}  |  HIT: {hit_count}'
            )
        except Exception:
            # 队列为空时超时跳过，继续等待新进度
            continue

    # 等待所有子进程退出
    the_pool.close()
    the_pool.join()
    # 扫描完成后最终更新标题
    set_console_title(
        f'ALL SITE: {total_site}  |  SITE VALID: {valid_count}  |  REMAIN: 0  |  FINISHED: {finished_count}  |  SCAN COMPLETED'
    )
    print("所有扫描任务执行完毕")

if __name__ == '__main__':
    main()

注意事项

不要在子进程中执行任何控制台UI相关的操作，所有界面更新逻辑收敛到主进程，从根源避免多进程资源抢占导致的显示异常。
跨进程数据传递优先用multiprocessing.Queue，不要依赖全局变量做跨进程计数——spawn模式下全局变量是各进程独立的副本，修改不会跨进程同步。
原错误代码中标题栏重复写了两个REMAIN字段，上述示例已经修正，你可以根据自己的需求调整标题展示的字段内容。
如果待扫描站点量级特别大，可以适当调整队列的缓冲大小，或者用multiprocessing.Value做共享计数，但单进程更新标题的原则不要变，否则依然会出现标题被覆盖的问题。

内容的提问来源于stack exchange，提问作者Massimo B.