网站建设思维导图模版电商直播app开发公司

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().percentreturn 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().percentreturn cpu, memdef 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, memexcept Exception as e:st.error(f"连接失败: {str(e)}")return 0.0, 0.0def 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()

运行结果：