cpu、mem监控

1、安装依赖

/usr/bin/python3 -m pip install  psutil streamlit

2、psutil库

psutil是一个跨平台的Python库，用于获取系统使用情况的信息，包括CPU、内存、磁盘和进程等。主要用途是进行系统监控、分析和限制流程资源以及管理正在运行的流程。

3、streamlit库

streamlit官方文档 开源框架，支持用python实现前端页面。

4、实战

4.1 单机CPU和内存使用率监控

import psutil
import time
import streamlit as st
from datetime import datetime

# 初始化 Streamlit 页面
st.set_page_config(page_title="资源监控看板", layout="wide")
st.title("实时 CPU 和内存使用率监控")

# 初始化数据存储（使用 Streamlit 的 session_state 持久化数据）
if "data" not in st.session_state:
    st.session_state.data = {
        "timestamps": [],
        "cpu_usages": [],
        "mem_usages": []
    }


# 定义获取 CPU 和内存使用率的函数
def get_system_usage():
    """获取当前 CPU 和内存使用率"""
    cpu_percent = psutil.cpu_percent(interval=1)
    mem_percent = psutil.virtual_memory().percent
    return cpu_percent, mem_percent


# 创建页面布局
col1, col2 = st.columns(2)

# 实时数据展示区域
with col1:
    st.subheader("实时数值")
    cpu_metric = st.metric(label="CPU 使用率", value="0%")
    mem_metric = st.metric(label="内存使用率", value="0%")

# 图表展示区域
with col2:
    st.subheader("历史趋势")
    chart_placeholder = st.empty()

# 持续更新数据
while True:
    # 获取最新数据
    cpu, mem = get_system_usage()

    if cpu > 90:
        st.error("CPU 使用率超过 90%!")
    if mem > 90:
        st.error("内存使用率超过 90%!")

    current_time = datetime.now().strftime("%H:%M:%S")

    # 更新数据存储（保留最近 60 个点）
    st.session_state.data["timestamps"].append(current_time)
    st.session_state.data["cpu_usages"].append(cpu)
    st.session_state.data["mem_usages"].append(mem)

    if len(st.session_state.data["timestamps"]) > 60:
        st.session_state.data["timestamps"] = st.session_state.data["timestamps"][-60:]
        st.session_state.data["cpu_usages"] = st.session_state.data["cpu_usages"][-60:]
        st.session_state.data["mem_usages"] = st.session_state.data["mem_usages"][-60:]

    # 更新实时指标
    cpu_metric.metric(label="CPU 使用率", value=f"{cpu}%")
    mem_metric.metric(label="内存使用率", value=f"{mem}%")

    # 更新折线图
    chart_placeholder.line_chart(
        {
            "CPU (%)": st.session_state.data["cpu_usages"],
            "Memory (%)": st.session_state.data["mem_usages"]
        },
        height=400,
        use_container_width=True
    )

    # 间隔 1 秒
    time.sleep(1)

运行应用：

/usr/bin/python3  -m streamlit  run hhh.py

结果结果：

4.2 多机CPU和内存使用率监控

import time
import psutil
import pandas as pd
import streamlit as st
from datetime import datetime
from paramiko import SSHClient, AutoAddPolicy

# ======================
# 配置区（需修改以下信息）
# ======================
MACHINES = {
    "服务器A": {
        "hostname": "192.168.3.255",
        "port": 22,
        "username": "user",
        "password": "password",
        "use_local": False  # 是否本地机器
    },
    "服务器B": {
        "hostname": "10.179.89.143",
        "username": "ubuntu",
        "key_filename": "/path/to/key.pem",
        "use_local": False
    }
}


# ======================
# 核心功能实现
# ======================
class MachineMonitor:
    def __init__(self):
        self.init_session_data()

    def init_session_data(self):
        """初始化数据存储"""
        if "monitor_data" not in st.session_state:
            st.session_state.monitor_data = pd.DataFrame(
                columns=["machine", "timestamp", "cpu", "mem"]
            )

    def get_local_usage(self):
        """获取本地机器资源使用率"""
        cpu = psutil.cpu_percent(interval=1)
        mem = psutil.virtual_memory().percent
        return cpu, mem

    def get_remote_usage(self, machine_config):
        """通过SSH获取远程机器资源"""
        try:
            with SSHClient() as client:
                client.set_missing_host_key_policy(AutoAddPolicy())
                client.connect(**{k: v for k, v in machine_config.items()
                                  if k not in ['use_local']})

                # 执行命令获取数据（兼容不同Linux发行版）
                stdin, stdout, stderr = client.exec_command(
                    "echo 'CPU: $(top -bn1 | grep Cpu | awk '{print $2}')%; " +
                    "MEM: $(free -m | awk '/Mem:/ {printf \"%.1f\", $3/$2*100}')%'"
                )
                output = stdout.read().decode().strip()

                # 解析数据
                cpu = float(output.split("CPU: ")[1].split("%")[0])
                mem = float(output.split("MEM: ")[1].split("%")[0])
                return cpu, mem

        except Exception as e:
            st.error(f"连接失败: {str(e)}")
            return 0.0, 0.0

    def collect_data(self):
        """收集所有机器数据"""
        new_data = []

        for machine_name, config in MACHINES.items():
            if config.get("use_local"):
                cpu, mem = self.get_local_usage()
            else:
                cpu, mem = self.get_remote_usage(config)

            new_data.append({
                "machine": machine_name,
                "timestamp": datetime.now().strftime("%H:%M:%S"),
                "cpu": cpu,
                "mem": mem
            })

        # 更新数据
        new_df = pd.DataFrame(new_data)
        st.session_state.monitor_data = pd.concat(
            [st.session_state.monitor_data, new_df]
        ).drop_duplicates(keep="last")

        # 保留最近100条记录
        if len(st.session_state.monitor_data) > 100:
            st.session_state.monitor_data = st.session_state.monitor_data.iloc[-100:]


# ======================
# Streamlit 界面
# ======================
def main():
    st.set_page_config(page_title="多机器监控看板", layout="wide")
    st.title("多机器资源监控系统")

    monitor = MachineMonitor()
    refresh_rate = st.sidebar.slider("刷新频率（秒）", 1, 10, 2)

    # 实时数据展示
    monitor.collect_data()

    # 为每个机器创建图表
    for machine in MACHINES.keys():
        with st.expander(f"🔧 {machine} 监控详情", expanded=True):
            col1, col2 = st.columns(2)

            # 实时数值
            with col1:
                latest = st.session_state.monitor_data[
                    st.session_state.monitor_data.machine == machine].iloc[-1]
                st.metric("CPU使用率", f"{latest.cpu:.1f}%")
                st.metric("内存使用率", f"{latest.mem:.1f}%")

            # 历史图表
            with col2:
                chart_data = st.session_state.monitor_data[
                    st.session_state.monitor_data.machine == machine]
                st.line_chart(
                    chart_data,
                    x="timestamp",
                    y=["cpu", "mem"],
                    height=200,
                    use_container_width=True
                )

    # 自动刷新
    time.sleep(refresh_rate)
    st.rerun()


if __name__ == "__main__":
    main()

运行结果：