系统监控模块

psutil模块

psutil模块文档

psutil是一个跨平台库(http://pythonhosted.org/psutil/)能够轻松实现获取系统运行的进程和系统利用率（包括CPU、内存、磁盘、网络等）信息。它主要用来做系统监控，性能分析，进程管理。它实现了同等命令行工具提供的功能，如ps、top、lsof、netstat、ifconfig、who、df、kill、free、nice、ionice、iostat、iotop、uptime、pidof、tty、taskset、pmap等。

一、安装psutil

　　pip install psutil

二、监控cpu信息

import psutil
psutil.cpu_times() #获取cpu（逻辑cpu的平均）占用时间的详细信息
psutil.cpu_times(percpu=True) #获取每个cpu占用时间的详细信息
psutil.cpt_times().user #获取用户进程占用cpu的时间（user+sys+idle+wait=total）

三、监控内存信息

import psutil
psutil.virtual_memory() #获取内存信息
psutil.virtual_memory().total #获取内存总量
psutil.swap_memory() #获取swap信息
psutil.swqp_memory() #获取swap总量

四、监控磁盘信息

import psutil
psutil.disk_partitions() #获取各分区的信息
psutil.disk_usage() #获取各分区的使用情况
psutil.disk_io_counters(perdisk=True) #获取各个分区的io情况
psutil.disk_io_counters(perdisk=True)[‘sda1’].read_count #获取sda1的io读取情况

五、监控网络信息

import psutil
psutil.net_io_counters() #获取所有网络接口io信息
psutil.net_io_counters(pernic=True) #获取每个网络接口的io信息

六、进程信息
import psutil
psutil.Process(pid) #查看对应pid的进程信息
psutil.Process(pid).username() #查看是哪个用户创建的该进程
psutil.Process(pid).cmdline() #查看进程所在的路径

七、登录用户信息

import psutil
psutil.users() #查看目前登录用户信息

例子

import psutil
import time


def get_sys_rc():
    # 获得cpu信息：核心数量、使用率
    cpu_count = psutil.cpu_count()
    print(cpu_count)
    cpu_usage = psutil.cpu_percent(1)  # 间隔1秒钟统计一次使用率
    print(cpu_usage)
    # 获得内存大小和使用率
    print("#" * 50)
    mem = psutil.virtual_memory()
    mem_total = mem.total / 1024 / 1024
    mem_usage = mem.percent
    print(mem_total, mem_usage)

    # 获得磁盘信息和分区信息
    disk_info = psutil.disk_partitions()  # 得到所有的分区信息
    print(disk_info)
    for i in disk_info:  # 遍历所有的分区，得到分区的名字
        print(i.device)  # 输出设备名字
        part_info = psutil.disk_usage(i.device)  # 得到每个分区的使用率
        print(part_info)  # 输出使用率
        # round() 四舍五入的方法
        print(f"{i.device}总大小为{round(part_info.total / 1024 / 1024 / 1024)}G,使用率{part_info.percent}")

    # 获得网卡信息
    net_info = psutil.net_io_counters()
    step1 = net_info.bytes_sent
    time.sleep(3)
    step2 = psutil.net_io_counters().bytes_sent
    avg = (step2 - step1) / 3 / 1000
    print(f"当前平均的网络流量是{round(avg)}KB")

    # 获得所有网卡的ip地址
    # psutil.net_if_stats()
    return {"cpu_count": cpu_count, "cpu_usage": cpu_usage}
    # return  {"cpu_count":cpu_count,"cpu_usage":cpu_usage,mem_total,mem_usage,avg}


if __name__ == "__main__":
    print(get_sys_rc())

output

8
42.8
##################################################
32685.93359375 23.9
[sdiskpart(device='C:\\', mountpoint='C:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='D:\\', mountpoint='D:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='E:\\', mountpoint='E:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='H:\\', mountpoint='H:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='Q:\\', mountpoint='Q:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='S:\\', mountpoint='S:\\', fstype='NTFS', opts='rw,fixed'), sdiskpart(device='U:\\', mountpoint='U:\\', fstype='NTFS', opts='rw,fixed')]
C:\
sdiskusage(total=116955537408, used=76401147904, free=40554389504, percent=65.3)
C:\总大小为109G,使用率65.3
D:\
sdiskusage(total=174902472704, used=128004796416, free=46897676288, percent=73.2)
D:\总大小为163G,使用率73.2
E:\
sdiskusage(total=107373129728, used=24370171904, free=83002957824, percent=22.7)
E:\总大小为100G,使用率22.7
H:\
sdiskusage(total=214747312128, used=140843151360, free=73904160768, percent=65.6)
H:\总大小为200G,使用率65.6
Q:\
sdiskusage(total=214748360704, used=67760369664, free=146987991040, percent=31.6)
Q:\总大小为200G,使用率31.6
S:\
sdiskusage(total=214748360704, used=79423479808, free=135324880896, percent=37.0)
S:\总大小为200G,使用率37.0
U:\
sdiskusage(total=1073741819904, used=507053940736, free=566687879168, percent=47.2)
U:\总大小为1000G,使用率47.2
当前平均的网络流量是1KB
{'cpu_count': 8, 'cpu_usage': 42.8}

时间监控

cProfile介绍

cProfile官方文档

cProfile自python2.5以来就是标准版Python解释器默认的性能分析器。
其他版本的python，比如PyPy里没有cProfile的。
cProfile是一种确定性分析器，只测量CPU时间，并不关心内存消耗和其他与内存相关联的信息。

cProfile模块

时间监控装饰器

#!/usr/bin/env Python
# -- coding: utf-8 --

"""
@version: v1.0
@author: huangyc
@file: profiler_helper.py
@Description: 时间性能监控类，主要监控执行状态、频率和时长
@time: 2021/7/17 15:51
"""

from cProfile import Profile
from functools import wraps
from pstats import Stats


class ProfilerHelper:
    @staticmethod
    def profiler_wrap(sort_by: str = 'cumulative', print_stats: bool = True, print_callers: bool = True):
        """
        行状态、频率和时长装饰器
        :param sort_by: 排序规则
                        可选参数：
                              准则                       含义                       升序/降序排列
                              calls                    调用次数                        降序
                              cumulative               累计时间                        降序
                              cumtime                  累计时间                        降序
                              file                     文件                           升序
                              filename                 文件名                          升序
                              module                   模块名                          升序
                              ncalls                   调用总次数                       降序
                              pcalls                   原始调用书                       降序
                              line                     行号                            升序
                              name                     函数名                          升序
                              nfl                      函数名/文件名/行号组合             降序
                              stdname                  标准名称                         升序
                              time                     函数内部运行时间                   降序
                              tottime                  函数内部运行总时间                 降序
        :param print_stats: Create a Stats object based on the current profile and print the results to stdout.
        :param print_callers: 打印受测函数和调用函数的关系
        :return:
        """

        def wrapper(func):
            @wraps(func)
            def inner_func(*args, **kwargs):
                profiler = Profile()
                res = profiler.runcall(func, *args, **kwargs)
                stats = Stats(profiler)
                # strip_dirs()：删除报告中所有函数文件名的路径信息，这个方法改变stats实例内部的顺序，任何运行该方法的实例都将随机排列项目的顺序。
                # 如果两个项目是相同的，那么这两个项目就可以合并。
                stats.strip_dirs()
                # sort_stats(*keys)：通过一系列条件依次对所有项目进行排序，从而调整stats对象
                stats.sort_stats(sort_by)
                if print_stats:
                    stats.print_stats()
                if print_callers:
                    stats.print_callers()
                return res

            return inner_func

        return wrapper


@ProfilerHelper.profiler_wrap()
def fn(h, a=5, b=8):
    import re
    print(h, a, b)
    return re.compile("aaa|bbb")


if __name__ == '__main__':
    print(fn(464646, 8, b=8))

内存监控

tracemalloc介绍

tracemalloc模块官方文档

tracemalloc模块是跟踪python分配的内存块的调试工具。它提供以下信息：

回溯对象的分配位置
每个文件名和每个行号的已分配内存块的统计信息：已分配内存块的总大小、数量和平均大小
计算两个快照之间的差异以检测内存泄漏

tracemalloc模块

内存监控装饰器

#!/usr/bin/env Python
# -- coding: utf-8 --

"""
@version: v1.0
@author: huangyc
@file: tracemalloc_helper.py
@Description: 内存监控类
@time: 2021/7/17 15:53
"""
import tracemalloc

from functools import wraps


class TracemallocHelper:
    @staticmethod
    def tracemalloc_wrap(key_type: str = 'lineno', monitor_type: str = 'statistics', top_k: int = 10,
                         filter_self: bool = True):
        """
        内存监控装饰器
        :param key_type: 比较的字段，'traceback', 'filename', 'lineno'
        :param monitor_type: 监控类型，'statistics', 'change'
        :param top_k: 关注top_k的内存占用,默认关注前10个
        :param filter_self: 是否过滤掉监控程序本身的消耗
        :return:
        """

        def wrapper(func):
            @wraps(func)
            def inner_func(*args, **kwargs):
                tracemalloc.start()
                # ... start your application ...
                res = func(*args, **kwargs)
                snapshot1 = tracemalloc.take_snapshot()
                # ... call the function leaking memory ...
                snapshot2 = tracemalloc.take_snapshot()

                if monitor_type == 'change':
                    top_stats = snapshot2.compare_to(snapshot1, key_type)
                elif monitor_type == 'statistics':
                    top_stats = snapshot2.statistics(key_type)
                else:
                    top_stats = snapshot2.statistics(key_type)
                print(f"[ Top {top_k} differences ]")
                for idx, stat in enumerate(top_stats[:top_k]):
                    if filter_self and 'tracemalloc.py' in str(stat):
                        continue
                    print('d' % idx, stat)
                return res

            return inner_func

        return wrapper


@TracemallocHelper.tracemalloc_wrap()
def fn(h, a=5, b=8):
    print(h, a, b)

    d = [dict(zip('xy', (5, 6))) for i in range(100000)]
    t = [tuple(zip('xy', (5, 6))) for i in range(100000)]
    return d, t


if __name__ == '__main__':
    fn(68, 7, b=78)