在 Python psutil 中调用函数时如何监控 CPU 的使用情况?

How to monitor usage of CPU when function is called in Python psutil?

嘿,我正在学习 psutil 包,我想知道如何在函数进行时显示当前 CPU 用法?我想我需要一些线程或类似的东西,但是怎么做呢?谢谢你的回答。

import psutil
import random

def iHateThis():
    tab = []
    for i in range(100000):
        tab.append(random.randint(1, 10000))

    tab.sort()
    return tab;

while(True):
    currentProcess = psutil.Process()
    print(currentProcess.cpu_percent(interval=1))

您可以使用 multiprocessing 库执行此操作。 multiprocessing.Process是一个class,表示一个线程进程,以函数和名称启动,可以随时运行与.start()

import multiprocessing
import psutil
import random

def iHateThis():
    tab = []
    for i in range(100000):
        tab.append(random.randint(1, 10000))
    tab.sort()
    return tab;

hate = multiprocessing.Process(name='hate', target=iHateThis)
hate.start()

while(True):
    currentProcess = psutil.Process()
    print(currentProcess.cpu_percent(interval=1))

您可以使用 threading 到 运行 iHateThis 或 运行 功能与 cpu_percent()。我选择第二个版本。我会运行cpu_percent()发帖

因为它使用 while True 所以线程会永远 运行 并且没有好的方法来停止线程所以我使用全局变量 runningwhile running 来有停止这个循环的方法。

import threading
import psutil

def display_cpu():
    global running

    running = True

    currentProcess = psutil.Process()

    # start loop
    while running:
        print(currentProcess.cpu_percent(interval=1))

def start():
    global t

    # create thread and start it
    t = threading.Thread(target=display_cpu)
    t.start()

def stop():
    global running
    global t

    # use `running` to stop loop in thread so thread will end
    running = False

    # wait for thread's end
    t.join()

现在我可以用它来启动和停止将显示 CPU 的线程。因为我可能不得不使用 Ctrl+C 停止进程所以它会引发错误所以我使用 try/finally 来停止线程即使会有错误。

def i_hate_this():
    tab = []
    for i in range(1000000):
        tab.append(random.randint(1, 10000))
    tab.sort()
    return tab

# ---

start()
try:
    result = i_hate_this()
finally: # stop thread even if I press Ctrl+C
    stop()

完整代码:

import random
import threading
import psutil

def display_cpu():
    global running

    running = True

    currentProcess = psutil.Process()

    # start loop
    while running:
        print(currentProcess.cpu_percent(interval=1))

def start():
    global t

    # create thread and start it
    t = threading.Thread(target=display_cpu)
    t.start()

def stop():
    global running
    global t

    # use `running` to stop loop in thread so thread will end
    running = False

    # wait for thread's end
    t.join()

# ---

def i_hate_this():
    tab = []
    for i in range(1000000):
        tab.append(random.randint(1, 10000))
    tab.sort()
    return tab

# ---

start()
try:
    result = i_hate_this()
finally: # stop thread even if I press Ctrl+C
    stop()

顺便说一句:这可以转换为继承自class Thread 的class,然后它可以在class 中隐藏变量running

import psutil
import random
import threading

class DisplayCPU(threading.Thread):

    def run(self):

        self.running = True

        currentProcess = psutil.Process()

        while self.running:
            print(currentProcess.cpu_percent(interval=1))

    def stop(self):
        self.running = False

# ----

def i_hate_this():
    tab = []
    for i in range(1000000):
        tab.append(random.randint(1, 10000))
    tab.sort()
    return tab

# ---

display_cpu = DisplayCPU()

display_cpu.start()
try:
    result = i_hate_this()
finally: # stop thread even when I press Ctrl+C
    display_cpu.stop()

它也可以转换为上下文管理器 运行 as

with display_cpu():
    i_hate_this()

但是我跳过了这部分。

我认为您不需要使用 psutil Process class,因为我认为它旨在用于监视特定进程。使用@furas 中的代码片段(已接受的答案),您可以使用这样的线程来完成:

def run(self):
    self.run = True 
    while self.run:
        psutil.cpu_percent(interval=1)

在以下情况下,它与接受的答案相同:

    _monitor.start()
    try:
        for i in range(50):
            time.sleep(0.2)
    finally:    
        _monitor.stop()

如果您不想对其进行编码,我会在 public 存储库中进行编码,如果它对某人有任何帮助的话:https://github.com/GTimothee/monitor