如何查看Python threading库中各线程的CPU使用率并输出到控制台？

我的Python应用把单个核心的CPU占满了，我想搞清楚应用里每个线程的CPU使用率占比，然后把结果输出到控制台，请问这能实现吗？

补充信息：

代码示例 #

import threading
def main(self):
    for i in range(
        int(self.type1_thread_limit)
    ):  # 启动线程推送type1 API数据到缓存
        worker = threading.Thread(target=self.type1_worker, args=())
        worker.start()

    for i in range(
        int(self.type2_thread_limit)
    ):  # 启动线程推送type2 API数据到缓存
        worker = threading.Thread(
            target=self.type2_worker, args=()
        )
        worker.start()

    for i in range(
        int(self.type1_dbthread_limit)
    ):  # 启动线程把缓存里的type1数据推送到DB
        worker2 = threading.Thread(target=self.type1_db_worker, args=())
        worker2.start()

    for i in range(
        int(self.type2__dbthread_limit)
    ):  # 启动线程把缓存里的type2数据推送到DB
        workerDepth = threading.Thread(
            target=self.type2_db_worker, args=()
        )
        workerDepth.start()

    while True:
        # 调度线程任务.....

ps -T -p <pid> 命令输出 #

PID SPID TTY TIME CMD
68085 68085 ? 00:11:52 python3
68085 68146 ? 00:00:00 python3
68085 68147 ? 00:00:00 python3
68085 68148 ? 00:00:00 python3
68085 68149 ? 00:00:00 python3
68085 68150 ? 00:00:00 python3
68085 68151 ? 00:00:00 python3
68085 68152 ? 00:00:00 python3
68085 68154 ? 00:00:00 python3
68085 68698 ? 00:16:00 python3
68085 68699 ? 00:16:22 python3
68085 68700 ? 00:16:15 python3
68085 68701 ? 00:16:38 python3
68085 68702 ? 00:16:01 python3
68085 68703 ? 00:16:44 python3
68085 68704 ? 00:16:04 python3
68085 68705 ? 00:16:27 python3
68085 68706 ? 00:16:39 python3
68085 68707 ? 00:16:42 python3
68085 68708 ? 00:16:40 python3
68085 68709 ? 00:16:33 python3
68085 68710 ? 00:16:25 python3
68085 68711 ? 00:16:52 python3
68085 68712 ? 00:16:23 python3
68085 68713 ? 00:06:50 python3
68085 68714 ? 00:07:08 python3
68085 68715 ? 00:06:42 python3
68085 68716 ? 00:06:41 python3
68085 68717 ? 00:07:12 python3
68085 68718 ? 00:07:06 python3
68085 68719 ? 00:06:59 python3
68085 68720 ? 00:13:37 python3
68085 68721 ? 00:13:30 python3
68085 68722 ? 00:13:33 python3
68085 68723 ? 00:15:08 python3
68085 68724 ? 00:00:01 python3
68085 68725 ? 00:00:01 python3

VS Code调试模式下的线程调用栈 #

• MainThread
• APScheduler
• Thread-7 (type1_worker)
• Thread-8 (type1_worker)
• Thread-9 (type2_worker)
• Thread-10 (type2_worker)
• Thread-11 (type1_db_worker)
• Thread-12 (type1_db_worker)
• Thread-13 (type2_db_worker)
• Thread-14 (type2_db_worker)

完全可以实现，核心思路是用psutil库获取进程内每个线程的CPU使用数据，同时关联线程名称以对应业务线程，下面是具体实现：

实现步骤 #

1. 安装依赖：执行pip install psutil安装进程监控库
2. 关联线程ID与名称：利用Pythonthreading模块获取线程的ident（对应ps命令的SPID）和名称
3. 定时计算使用率：通过两次采集线程CPU时间的差值，计算出单个线程的CPU使用率占比

示例代码 #

在应用中添加一个监控线程，专门输出各线程CPU使用率：

import threading
import psutil
import time
from threading import enumerate as enumerate_threads

def monitor_thread_cpu():
    current_process = psutil.Process()
    # 首次采集线程CPU时间
    prev_thread_times = {t.id: t.cpu_times() for t in current_process.threads()}
    time.sleep(1)

    while True:
        # 二次采集线程CPU时间
        curr_thread_times = {t.id: t.cpu_times() for t in current_process.threads()}
        # 构建线程ID到名称的映射
        thread_name_map = {thread.ident: thread.name for thread in enumerate_threads()}
        
        print("\n=== 线程CPU使用率统计 ===")
        total_usage = 0.0
        for spid, curr_times in curr_thread_times.items():
            prev_times = prev_thread_times.get(spid)
            if not prev_times:
                continue
            # 计算时间差并转换为使用率（单核下，1秒内的时间差即使用率百分比）
            cpu_diff = (curr_times.user + curr_times.system) - (prev_times.user + prev_times.system)
            cpu_usage = cpu_diff / 1 * 100
            total_usage += cpu_usage
            # 获取线程名称，未知则显示SPID
            thread_name = thread_name_map.get(spid, f"UnknownThread-{spid}")
            print(f"线程名称: {thread_name:25} | SPID: {spid:6} | CPU使用率: {cpu_usage:.2f}%")
        
        print(f"总CPU使用率: {total_usage:.2f}%")
        prev_thread_times = curr_thread_times
        time.sleep(2)  # 每2秒刷新一次

# 修改原main函数，添加监控线程
def main(self):
    # 启动守护式监控线程，主进程退出时自动终止
    monitor = threading.Thread(target=monitor_thread_cpu, daemon=True)
    monitor.start()

    # 原有的业务线程启动代码...
    for i in range(int(self.type1_thread_limit)):
        worker = threading.Thread(target=self.type1_worker, args=())
        worker.start()

    for i in range(int(self.type2_thread_limit)):
        worker = threading.Thread(target=self.type2_worker, args=())
        worker.start()

    for i in range(int(self.type1_dbthread_limit)):
        worker2 = threading.Thread(target=self.type1_db_worker, args=())
        worker2.start()

    for i in range(int(self.type2__dbthread_limit)):
        workerDepth = threading.Thread(target=self.type2_db_worker, args=())
        workerDepth.start()

    while True:
        # 调度线程任务.....

关键说明 #

• 线程ID对应关系：Python线程的ident属性就是ps命令中的SPID，可直接关联
• 使用率计算逻辑：单核环境下，1秒内线程的CPU时间消耗占比即为使用率，最大值为100%
• 线程名称优化：创建业务线程时可指定name参数（如threading.Thread(target=self.type1_worker, name="type1_worker-1")），让监控输出更直观
• 守护线程特性：监控线程设为守护线程，避免主进程退出后残留后台线程

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