Python3如何在线程已运行时避免重复启动新线程

核心逻辑是持久持有线程实例的引用，每次调用启动方法前先检查线程存活状态，仅当线程不存在或已退出时才创建并启动新线程。
Python 标准库threading.Thread自带is_alive()方法，可以直接判断线程是否处于运行状态。

基础全局变量实现 #

适合简单脚本场景，逻辑最直观：

from threading import Thread

# 全局持有线程实例引用，初始为None
kline_thread = None

def kline_func():
    # 此处填写远程服务器监听逻辑
    while True:
        pass

def start_my_thread():
    global kline_thread
    # 仅当线程未创建、或已终止（异常退出/运行结束）时才启动新线程
    if kline_thread is None or not kline_thread.is_alive():
        kline_thread = Thread(target=kline_func, daemon=True) # 建议设置daemon=True，避免主线程退出后线程残留
        kline_thread.start()

# 多次调用测试
start_my_thread() # 首次调用：启动线程
start_my_thread() # 第二次：检测到线程运行中，无操作
start_my_thread() # 第三次：检测到线程运行中，无操作

注：给Thread加daemon=True参数后，主线程退出时会自动终止该监听线程，避免程序退出后残留后台进程。

面向对象封装实现 #

适合项目工程场景，避免全局变量污染：

from threading import Thread

class KlineMonitor:
    def __init__(self):
        self._worker_thread = None

    def _listen_task(self):
        # 远程监听业务逻辑
        while True:
            pass

    def start(self):
        if self._worker_thread is None or not self._worker_thread.is_alive():
            self._worker_thread = Thread(target=self._listen_task, daemon=True)
            self._worker_thread.start()

# 使用方式
monitor = KlineMonitor()
monitor.start() # 启动监听
monitor.start() # 线程已运行，无操作
monitor.start() # 线程已运行，无操作

线程安全版本 #

如果存在多线程并发调用start_my_thread()的场景，加一把线程锁即可避免竞态条件导致重复创建线程：

from threading import Thread, Lock

kline_thread = None
_start_lock = Lock()

def kline_func():
    while True:
        pass

def start_my_thread():
    global kline_thread
    with _start_lock:
        if kline_thread is None or not kline_thread.is_alive():
            kline_thread = Thread(target=kline_func, daemon=True)
            kline_thread.start()

注意事项 #

• 不要每次调用启动方法时都新建Thread实例：Thread对象本身只能调用一次start()，且局部变量会被回收，根本无法追踪之前启动的线程状态。
• 上述实现天然支持故障自动恢复：如果监听线程因为网络中断、未捕获异常退出，下次调用启动方法时会自动拉起新的监听线程，不需要额外写重启逻辑。

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