PyInstaller打包的exe中多线程无法运行问题求助

问题描述 #

项目通过PyInstaller打包为exe文件后，线程无法启动，但在PyCharm中运行时一切正常。应用整合了Django与自定义应用，通过命令.\myapp.exe runserver localhost:40218 --noreload启动Django。

主程序代码 #

import os
import threading

import django
from django.core.management.commands.runserver import Command as runserver

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'rest_api.settings.base')
django.setup()
runserver.default_port = "40218"

from app.processes import start_processing  # noqa: E402
from app.threads_configuration import multiple_threads_creator_wrapper  # noqa: E402
from manage import main  # noqa: E402

if __name__ == '__main__':
    try:
        ctx = ctx_manager()
        threading.Thread(target=multiple_threads_creator_wrapper, args=(start_processing,)).start()
        main()
    except Exception as e:
        print(f'Application closed due to error: {str(e)}')

multiple_threads_creator_wrapper函数 #

def multiple_threads_creator_wrapper(target, *args, **kwargs):
    LimitedThread(target=target, *args, **kwargs).start()
    RepeatedTimer(target, *args, **kwargs)

线程配置代码 #

import time
import threading

class Config:
    def __init__(self) -> None:
        self.maxThreads = 10
        self.runEveryXSeconds = 5  


_config = Config()
_threadLimiter = threading.BoundedSemaphore(_config.maxThreads)


class LimitedThread(threading.Thread):
    def run(self):
        _threadLimiter.acquire()
        try:
            self._target(*self._args, **self._kwargs)
        finally:
            _threadLimiter.release()


class RepeatedTimer(object):
    def __init__(self, function, *args, **kwargs):
        self._timer = None
        self.interval = _config.runEveryXSeconds
        self.function = function
        self.is_running = False
        self.next_call = time.time()
        self._args = args
        self._kwargs = kwargs
        self.start()    

解决方案 #

针对PyInstaller打包后线程无法启动的问题，可从以下几个方向排查修复：

1. 修复RepeatedTimer的生命周期问题 #

PyInstaller打包后的环境中，未被引用的对象容易被垃圾回收机制提前清理，导致定时器线程终止。需完善RepeatedTimer的运行逻辑并保留实例引用：

修改后的RepeatedTimer类：

class RepeatedTimer(object):
    def __init__(self, function, *args, **kwargs):
        self._timer = None
        self.interval = _config.runEveryXSeconds
        self.function = function
        self.is_running = False
        self.next_call = time.time()
        self._args = args
        self._kwargs = kwargs
        self.start()  

    def _run(self):
        self.is_running = False
        self.function(*self._args, **self._kwargs)
        self.start()

    def start(self):
        if not self.is_running:
            self.next_call += self.interval
            self._timer = threading.Timer(self.next_call - time.time(), self._run)
            self._timer.start()
            self.is_running = True

    def stop(self):
        self._timer.cancel()
        self.is_running = False

同时在multiple_threads_creator_wrapper中保留全局引用：

# 全局存储定时器实例，避免被垃圾回收
_global_timers = []

def multiple_threads_creator_wrapper(target, *args, **kwargs):
    LimitedThread(target=target, *args, **kwargs).start()
    timer = RepeatedTimer(target, *args, **kwargs)
    _global_timers.append(timer)

2. 调整PyInstaller打包参数 #

Django项目打包需明确指定隐藏导入，避免模块遗漏：

pyinstaller --onefile --hidden-import=app.processes --hidden-import=app.threads_configuration --hidden-import=rest_api.settings.base --name myapp myapp.py

• --onefile生成单文件exe
• --hidden-import手动指定未被自动检测到的依赖模块
• --name指定生成的exe文件名

3. 调整线程启动时机 #

Django初始化流程可能干扰线程启动，可延迟线程启动确保Django加载完成：

if __name__ == '__main__':
    try:
        ctx = ctx_manager()
        def start_after_django():
            threading.Thread(target=multiple_threads_creator_wrapper, args=(start_processing,)).start()
        # 延迟1秒启动线程，确保Django初始化完成
        threading.Timer(1.0, start_after_django).start()
        main()
    except Exception as e:
        print(f'Application closed due to error: {str(e)}')

4. 添加线程内异常捕获 #

线程内部崩溃时无错误输出，难以排查，需添加异常捕获：

class LimitedThread(threading.Thread):
    def run(self):
        _threadLimiter.acquire()
        try:
            self._target(*self._args, **self._kwargs)
        except Exception as e:
            print(f'Thread error: {str(e)}')
        finally:
            _threadLimiter.release()

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