在Python的psutil中调用函数时如何监控CPU使用率？

Question

嘿，我正在学习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))

- ajirebardyn

---

你可以在“while”循环之前创建“currentProcess”。 - furas

Answer 1

您可以使用 threading 来运行iHateThis或者运行带有 cpu_percent()的函数。我选择第二个版本，会在线程中运行 cpu_percent()。

由于它使用while True，所以线程将永远运行，并且没有很好的停止线程的方法，因此我使用全局变量running并使用while 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()

顺便说一句：这可以转换为继承自类Thread的类，然后它可以在类中隐藏变量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()

它也可以转换为上下文管理器，以以下方式运行：

with display_cpu():
    i_hate_this()

but I skip this part.

- furas

---

嗨，不知道为什么控制台一直显示0.0 :/ 我正在尝试制作一个非常基本的资源监视器（类似于Windows功能）。也许cpu_percent函数不正确？我不知道... - ajirebardyn

也许你的电脑速度非常快，几乎不需要 CPU 占用率。而我的古老电脑需要 98.0 的占用率。你可以尝试在 range() 中使用更大的值。或者它可能需要比 1 更小的 interval。 - furas

我使用Linux，但在Windows上可能会有所不同。您可以在线程和主循环中使用print(psutil.Process().pid)来检查线程是否计算完整进程的CPU或仅计算线程的CPU。 - furas

所以我按照你说的做了，但结果并不理想... 我设置了范围为1000000000000000和间隔为0.0000000015 - 什么也没发生 另外，我打印pid，它总是显示相同的数字，例如：996、1300、6060、10096等。 - ajirebardyn

请检查我的代码，也许我做错了什么。 https://pastebin.com/FHZh2gMCPsutil是可靠的包来完成这样的事情吗？ - ajirebardyn

Answer 2

你可以使用 multiprocessing 库来实现。 multiprocessing.Process 是一个代表线程进程的类，它通过函数和名称进行初始化，并且可以随时使用 .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))

- Nat Weiland

Answer 3

让我们以不同的方式来解决问题，提出一个装饰器，可以在运行时测量CPU利用率。

from functools import partial, wraps


def log_cpu_usage(func=None, msg_prefix: str = None):
    """
    This function is a decorator that measures the execution time of a function and logs it.
    """
    debug = True
    if not debug:
        return func
    if func is None:
        return partial(log_cpu_usage, msg_prefix=msg_prefix)
    
    def new_func(data: mp.Queue, *args, **kwargs):
        result = func(*args, **kwargs)
        data.put(result)

    @wraps(func)
    def trace_execution(*args, **kwargs):
        manager = mp.Queue() # to save return val between multi process
        worker_process = mp.Process(target=new_func, args=(manager, *args), kwargs=kwargs)
        worker_process.start()
        p = psutil.Process(worker_process.pid)
        cpu_percents = []
        while worker_process.is_alive(): # while the subprocess is running
            cpu_percents.append(p.cpu_percent() / psutil.cpu_count())
            time.sleep(0.01)
        worker_process.join()
        ret_values = manager.get()
        return sum(cpu_percents) / len(cpu_percents), ret_values

@log_cpu_usage
def iHateThis():
    pass

- Alex Nea Kameni

Answer 4

我认为你不需要使用 psutil 的 Process 类，因为我认为它旨在用于监视特定的进程。使用来自 @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()

如果您不想编写代码，我已经在公共存储库中完成了它，如果对某人有所帮助：https://github.com/GTimothee/monitor

- ava_punksmash