Python 監控 Windows CPU 使用率：基於 WMI 與 PDH 的實現方案

在 Windows 系統上獲取 CPU 使用率，常用兩種方式：WMI（Windows Management Instrumentation）和 PDH（Performance Data Helper）。本文介紹基於 win32pdh 的高性能實現，並提供持續監控與告警的完整示例。

核心技術選型：PDH vs WMI

依賴安裝

pip install pywin32
# 需以管理員權限運行一次以註冊 DLL
python -m pywin32_postinstall

核心代碼實現

# -*- coding: utf-8 -*-
import win32pdh
import time

# 性能計數器路徑
PROCESSOR_PERCENT = r'Processor(_Total)\% Processor Time'
MEMORY_PERCENT = r'Memory\% Committed Bytes In Use'
MEMORY_COMMITTED = r'Memory\Committed Bytes'
PROCESS_BYTES = lambda x: r'Process(%s)\Private Bytes' % x

class PDHQuery:
    """PDH 查詢封裝類"""
    def __init__(self):
        self.counters = {}
        self.query = win32pdh.OpenQuery(None, 0)
    
    def add_counter(self, path):
        if win32pdh.ValidatePath(path) != 0:
            raise Exception('無效計數器路徑: %s' % path)
        counter = win32pdh.AddCounter(self.query, path, 0)
        self.counters[path] = counter
    
    def remove_counter(self, path):
        win32pdh.RemoveCounter(self.counters[path])
        del self.counters[path]
    
    def get_values(self):
        """收集並返回所有計數器值"""
        values = {}
        win32pdh.CollectQueryData(self.query)
        for path, counter in self.counters.items():
            status, value = win32pdh.GetFormattedCounterValue(
                counter, win32pdh.PDH_FMT_LONG)
            values[path] = value
        return values

# 初始化系統監控查詢器
sys_query = PDHQuery()
sys_query.add_counter(PROCESSOR_PERCENT)
sys_query.add_counter(MEMORY_PERCENT)
sys_query.get_values()  # 首次調用初始化，返回值通常為 0

def get_sysinfo():
    """返回 (內存使用率%, CPU使用率%)"""
    info = sys_query.get_values()
    return info[MEMORY_PERCENT], info[PROCESSOR_PERCENT]

# 持續監控邏輯：統計 1 分鐘內 CPU 超過 4% 的次數
cpu_samples = []
THRESHOLD = 4
WINDOW_SIZE = 10  # 10 次采樣，間隔 2 秒 = 20 秒窗口
ALERT_COUNT = 5   # 超過 5 次觸發告警

print("開始監控 CPU 使用率（每 2 秒采樣一次）...")
print("按 Ctrl+C 停止")

try:
    while True:
        time.sleep(2)
        mem_usage, cpu_usage = get_sysinfo()
        cpu_samples.append(cpu_usage)
        
        # 保持滑動窗口
        if len(cpu_samples) > WINDOW_SIZE:
            cpu_samples.pop(0)
        
        # 統計窗口內超閾值次數
        high_count = sum(1 for v in cpu_samples if v > THRESHOLD)
        
        print(f"CPU: {cpu_usage:3d}% | 內存: {mem_usage:3d}% | 窗口內>4%: {high_count}/{len(cpu_samples)}")
        
        if len(cpu_samples) == WINDOW_SIZE and high_count >= ALERT_COUNT:
            print(f"⚠️ 告警：最近 {WINDOW_SIZE*2} 秒內 CPU 使用率超過 {THRESHOLD}% 達 {high_count} 次！")
            # 這裡可接入郵件、鉤子、短信等告警通道
            cpu_samples.clear()  # 重置窗口避免重複告警

except KeyboardInterrupt:
    print("\n監控已停止")

關鍵點詳解

1. 計數器路徑格式

Windows 性能計數器路徑格式：\\Object(Instance)\Counter

• Processor(_Total)\% Processor Time：所有 CPU 核心總體使用率
• Processor(0)\% Processor Time：第 0 號核心使用率
• Memory\% Committed Bytes In Use：提交內存使用百分比
• Process(chrome)\Private Bytes：Chrome 進程私有內存

2. 雙次采樣原理

PDH 計數器需要 兩次收集 才能計算出有效率值：

1. 首次 CollectQueryData 建立基線
2. 第二次收集時計算差值/速率

因此代碼中初始化時調用一次 get_values() 丟棄結果。

3. 格式化輸出類型

win32pdh.PDH_FMT_LONG      # 長整數
win32pdh.PDH_FMT_DOUBLE    # 雙精度浮點數（更精確）
win32pdh.PDH_FMT_LARGE     # 64位整數

進階擴展：多進程監控

def monitor_process(process_name):
    """監控指定進程的 CPU 和內存"""
    query = PDHQuery()
    query.add_counter(PROCESS_BYTES(process_name))
    query.add_counter(r'Process(%s)\%% Processor Time' % process_name)
    query.get_values()  # 初始化
    
    while True:
        time.sleep(1)
        vals = query.get_values()
        mem_mb = vals[PROCESS_BYTES(process_name)] / 1024 / 1024
        cpu_pct = vals[r'Process(%s)\%% Processor Time' % process_name]
        print(f"{process_name}: CPU={cpu_pct:.1f}% MEM={mem_mb:.1f}MB")

常見問題排查

總結

基於 win32pdh 的 Windows 性能監控具有高性能、低延遲、原生支持等優勢，適合構建生產級監控代理。結合滑動窗口統計邏輯，可實現智能告警，避免抖動誤報。建議在部署前使用 typeperf 驗證計數器路徑可用性。