如何为Python线程分配内存限制？线程与子进程模块及并行问题

一、如何为Python线程分配内存限制？ #

首先得明确：Python的线程是共享所属进程的内存空间的，操作系统并没有提供直接给单个线程设置内存上限的机制。所以要实现线程级的内存限制，只能在代码层面手动监控和控制：

• 用tracemalloc模块跟踪线程内的内存使用：可以在线程函数里启动内存跟踪，定期检查已分配的内存量，一旦超过阈值就触发线程退出或资源释放。
  示例代码：

  import threading
  import tracemalloc
  import time
  
  def thread_func(limit_mb):
      tracemalloc.start()
      threshold = limit_mb * 1024 * 1024  # 转换为字节
      while True:
          current, peak = tracemalloc.get_traced_memory()
          if current > threshold:
              print(f"线程{threading.current_thread().name}内存超过限制，退出")
              tracemalloc.stop()
              return
          # 模拟内存占用操作
          data = [i for i in range(10000)]
          time.sleep(0.1)
  
  threading.Thread(target=thread_func, args=(50,), name="Thread-1").start()
  threading.Thread(target=thread_func, args=(50,), name="Thread-2").start()
  

• 手动监控关键变量的内存：如果线程里有大对象，比如列表、字典，可以定期计算它们的内存占用（用sys.getsizeof，注意嵌套对象要递归计算），超过限制就清理这些对象。

二、如何使用threading和subprocess模块限制对应进程的内存？ #

1. threading模块（线程所属进程的内存限制） #

因为线程共享进程内存，所以threading本身没法单独限制线程内存，但可以给整个进程设置内存上限：

• 在Linux/macOS下，可以用resource模块设置进程的虚拟内存限制：

  import resource
  import threading
  
  # 设置进程最大虚拟内存为500MB
  resource.setrlimit(resource.RLIMIT_AS, (500 * 1024 * 1024, 500 * 1024 * 1024))
  
  def thread_task():
      # 线程内的操作会受进程内存限制约束
      large_data = [i for i in range(10**7)]
  
  threading.Thread(target=thread_task).start()
  

  当进程总内存超过限制时，系统会发送内存不足信号，进程可能被终止。

2. subprocess模块（子进程的内存限制） #

subprocess启动的是独立进程，可以直接给子进程设置内存限制，不同操作系统有不同方式：

• Linux/macOS下，用preexec_fn结合resource模块：

  import subprocess
  import resource
  
  def set_memory_limit(limit_mb):
      # 在子进程启动前设置内存限制
      limit = limit_mb * 1024 * 1024
      resource.setrlimit(resource.RLIMIT_AS, (limit, limit))
  
  # 启动子进程并限制内存为200MB
  proc = subprocess.Popen(
      ["bash", "-c", "some_memory_intensive_command"],
      preexec_fn=lambda: set_memory_limit(200)
  )
  proc.wait()
  

• Windows下，可以借助Windows API或第三方库（如psutil）来设置进程内存限制，也可以在启动命令中通过系统工具配置。

三、线程启动bash进程呈串行运行而非并行，该如何处理？ #

你遇到的这个问题，大概率是代码里的细节没处理好，常见原因和解决方法如下：

1. 检查是否启动了多个线程 #

如果只启动了一个线程，那bash命令肯定是串行的。要确保创建多个threading.Thread实例并调用start()：
错误示例（串行）：

import threading
import subprocess

def run_bash():
    subprocess.run(["sleep", "2"])  # 模拟耗时bash命令

# 直接串行调用函数，没有启动线程
run_bash()
run_bash()

正确示例（并行）：

import threading
import subprocess

def run_bash():
    subprocess.run(["sleep", "2"])

# 启动两个线程并行执行
t1 = threading.Thread(target=run_bash)
t2 = threading.Thread(target=run_bash)
t1.start()
t2.start()
t1.join()
t2.join()

这样两个sleep 2会同时执行，总耗时约2秒而不是4秒。

2. 检查是否有不必要的同步锁 #

如果线程函数里用了threading.Lock且锁的范围过大，会导致线程串行执行。比如：

lock = threading.Lock()

def run_bash():
    with lock:  # 不必要的锁，导致所有线程排队执行
        subprocess.run(["sleep", "2"])

这种情况要去掉不必要的锁，确保bash命令的执行不受全局锁的阻塞。

3. 检查bash命令本身是否有串行依赖 #

如果你的bash命令需要读写同一个独占文件、使用同一个排他性资源，那即使线程并行启动，bash命令也会因为资源竞争而串行执行。这种情况要优化bash命令，比如给每个子进程分配独立的文件或资源。

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