如何实现主脚本触发多并行子脚本且不影响自身持续运行？

用户需要构建一款持续运行的补货通知机器人，需求如下：

• 每3分钟检查商品是否补货
• 补货时立即给优先收件组发送邮件
• 同时触发子任务延迟10分钟给次收件组发送相同邮件，子任务独立运行、完成即退出，主脚本无需追踪其状态
• 同一时间最多运行3个子任务

用户基于threading编写了示例代码，询问以下问题：

1. 该方案是否存在问题？
2. 长期运行是否会耗尽资源？
3. 应选用多线程还是多进程，或其他可行实现方案？

示例代码如下：

import threading
import time
import random

def delayed_email():
    time.sleep(10)
    print ("Delayed Email has been sent")

def immediate_email():
    print ("Immediate Email has been sent")

while True:
    number = random.randint(0,3)
    print (number)
    if number == 1: 
        immediate_email()
        print ("Starting Script 2")
        threading.Thread(target=delayed_email).start()
    time.sleep(3)

现有代码的问题 #

• 无并发限制：当前代码没控制同时运行的子线程数量，要是短时间内多次触发补货，会创建大量线程，可能占满系统资源，甚至拖慢主脚本的检查逻辑。
• 线程未设守护属性：如果手动终止主脚本，未设为守护线程的子线程会继续跑，导致程序没法正常退出。
• 无异常处理：发送邮件要是出问题（比如网络断了），线程会直接崩溃，连个错误记录都没有，排查问题全靠猜。

长期运行的资源情况 #

Python的线程是内核线程，每个线程会占一点内存（主要是栈空间）。如果不管并发数，长期跑下来确实会耗内存；但只要控制好最多3个并发，线程完成任务后会自动销毁释放资源，完全不会出现资源耗尽的情况。

多线程vs多进程的选择 #

• 优先选多线程：你的场景里，子任务主要是等待延迟和发邮件，属于IO密集型操作。Python的GIL在IO操作时会自动释放，不会影响并发效率，而且线程创建销毁的开销比进程小得多，适合这种轻量、频繁的子任务。
• 没必要用多进程：进程的创建、销毁和通信开销都很大，对于延迟10分钟发邮件这种简单任务来说，纯粹是浪费系统资源。

可行改进方案 #

方案1：用threading.Semaphore限制并发数 #

用信号量控制同时运行的子线程数量，确保最多3个：

import threading
import time
import random

# 限制最多3个并发子线程
semaphore = threading.Semaphore(3)

def delayed_email():
    with semaphore:
        try:
            time.sleep(10)
            print("Delayed Email has been sent")
        except Exception as e:
            print(f"Delayed email failed: {str(e)}")

def immediate_email():
    try:
        print("Immediate Email has been sent")
    except Exception as e:
        print(f"Immediate email failed: {str(e)}")

while True:
    number = random.randint(0,3)
    print(number)
    if number == 1: 
        immediate_email()
        print("Starting delayed email thread")
        # 设置为守护线程，主脚本退出时自动终止
        thread = threading.Thread(target=delayed_email, daemon=True)
        thread.start()
    time.sleep(3)

方案2：用concurrent.futures.ThreadPoolExecutor管理线程池 #

线程池可以更省心地管理线程生命周期，避免频繁创建销毁线程的开销：

import time
import random
from concurrent.futures import ThreadPoolExecutor

# 创建最多3个线程的线程池
executor = ThreadPoolExecutor(max_workers=3)

def delayed_email():
    try:
        time.sleep(10)
        print("Delayed Email has been sent")
    except Exception as e:
        print(f"Delayed email failed: {str(e)}")

def immediate_email():
    try:
        print("Immediate Email has been sent")
    except Exception as e:
        print(f"Immediate email failed: {str(e)}")

try:
    while True:
        number = random.randint(0,3)
        print(number)
        if number == 1: 
            immediate_email()
            print("Submitting delayed email task")
            executor.submit(delayed_email)
        time.sleep(3)
finally:
    # 程序退出时关闭线程池
    executor.shutdown()

额外优化建议 #

• 把print换成logging模块，方便后续查问题。
• 给发送邮件的逻辑加几次重试，避免一次网络波动就失败。
• 给主循环加异常捕获，防止单次检查出错导致整个程序崩掉。

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