您現在的位置是：首頁 > 單機遊戲首頁單機遊戲
Psutil + Flask + Pyecharts + Bootstrap 開發動態視覺化系統監控
- 2021-12-22
監控軟體mtu是什麼
文 | 某某白米飯
psutil 是一個跨平臺庫（http：//pythonhosted。org/psutil）能夠獲取到系統執行的程序和系統利用率（包括CPU、記憶體、磁碟、網路等）資訊。主要用來做系統監控，效能分析，程序管理。支援 Linux、Mac OS、Windows 系統。
本文以 psutil 模組獲取系統資訊開發一個監控 Mac OS 系統的平臺。
準備工作
技術選擇
監控的系統是 Mac OS 系統
監控系統模組選擇 psutil 模組
Web 框架選擇的是 Flask 框架
前端 UI 選擇的是 Bootstrap UI
動態視覺化圖表選擇 Pyecharts 模組
安裝 psutil
pip3 install psutil
安裝 Flask、pyecharts、Bootstrap
Flask 的教程是在公眾號文章：Web 開發 Flask 介紹
Pyecharts 的教程在公眾號文章：Python 圖表利器 pyecharts，按照官網 （http：//pyecharts。org/#/zh-cn/web_flask） 文件整合 Flask 框架，並使用定時全量更新圖表。
Bootstrap 是一個 前端的 Web UI，官網地址是 （https：//v4。bootcss。com）
獲取系統資訊
CPU資訊
透過 psutil 獲取 CPU 資訊
>>> import psutil
# 獲取當前 CPU 的利用率
>>> psutil。cpu_percent
53。8
# 獲取當前 CPU 的使用者/系統/空閒時間
>>> psutil。cpu_times
scputimes（user=197483。49， nice=0。0， system=114213。01， idle=1942295。68）
# 1/5/15 分鐘之內的 CPU 負載
>>> psutil。getloadavg
（7。865234375， 5。1826171875， 4。37353515625）
# CPU 邏輯個數
>>> psutil。cpu_count
4
# CPU 物理個數
>>> psutil。cpu_count（logical=False）
2
在監控平臺上每 2 秒請求 url 獲取 CPU 負載，並動態顯示圖表
cpu_percent_dict = {}
def cpu：
# 當前時間
now = time。strftime（‘%H：%M：%S’， time。localtime（time。time））
# CPU 負載
cpu_percent = psutil。cpu_percent
cpu_percent_dict［now］ = cpu_percent
# 保持在圖表中 10 個數據
if len（cpu_percent_dict。keys） == 11：
cpu_percent_dict。pop（list（cpu_percent_dict。keys）［0］）
def cpu_line -> Line：
cpu
# 全量更新 pyecharts 圖表
c = （
Line
。add_xaxis（list（cpu_percent_dict。keys））
。add_yaxis（‘’， list（cpu_percent_dict。values）， areastyle_opts=opts。AreaStyleOpts（opacity=0。5））
。set_global_opts（title_opts=opts。TitleOpts（title = now + “CPU負載”，pos_left = “center”），
yaxis_opts=opts。AxisOpts（min_=0，max_=100，split_number=10，type_=“value”， name=‘%’））
）
return c
@app。route（“/cpu”）
def get_cpu_chart：
c = cpu_line
return c。dump_options_with_quotes
示例結果
記憶體
透過 psutil 獲取記憶體和交換區資訊
# 系統記憶體資訊 總記憶體/立刻可用給程序使用的記憶體/記憶體負載/已使用記憶體/空閒記憶體/當前正在使用或者最近使用的記憶體/未使用的記憶體/永久在記憶體
>>> psutil。virtual_memory
svmem（total=8589934592， available=2610610176， percent=69。6， used=4251074560， free=387874816， active=2219110400， inactive=2069094400， wired=2031964160）
# 交換區記憶體 總記憶體/使用的記憶體/空閒的記憶體/負載/系統從磁碟交換進來的位元組數（累計）/系統從磁碟中交換的位元組數（累積）
>>> psutil。swap_memory
sswap（total=2147483648， used=834404352， free=1313079296， percent=38。9， sin=328911147008， sout=3249750016）
在監控平臺上每 2 秒請求 url 獲取記憶體負載，並動態顯示圖表
def memory：
memory = psutil。virtual_memory
swap = psutil。swap_memory
# 在 Mac OS 上 未使用記憶體 = 總記憶體 - （空閒記憶體 + 未使用記憶體）
return memory。total， memory。total - （memory。free + memory。inactive）， memory。free + memory。inactive， swap。total， swap。used， swap。free， memory。percent
def memory_liquid -> Gauge：
mtotal， mused， mfree， stotal， sused， sfree， mpercent = memory
c = （
Gauge
。add（“”， ［（“”， mpercent）］）
。set_global_opts（title_opts=opts。TitleOpts（title=“記憶體負載”， pos_left = “center”））
）
return mtotal， mused， mfree， stotal， sused， sfree， c
@app。route（“/memory”）
def get_memory_chart：
mtotal， mused， mfree， stotal， sused， sfree， c = memory_liquid
return jsonify（{‘mtotal’： mtotal， ‘mused’： mused， ‘mfree’： mfree， ‘stotal’： stotal， ‘sused’： sused， ‘sfree’： sfree， ‘liquid’： c。dump_options_with_quotes}）
示例結果
磁碟
透過 psutil 獲取磁碟大小、分割槽、使用率和磁碟IO
# 磁碟分割槽情況
>>> psutil。disk_partitions
［sdiskpart（device=‘/dev/disk1s5’， mountpoint=‘/’， fstype=‘apfs’， opts=‘ro，local，rootfs，dovolfs，journaled，multilabel’）， sdiskpart（device=‘/dev/disk1s1’， mountpoint=‘/System/Volumes/Data’， fstype=‘apfs’， opts=‘rw，local，dovolfs，dontbrowse，journaled，multilabel’）， sdiskpart（device=‘/dev/disk1s4’， mountpoint=‘/private/var/vm’， fstype=‘apfs’， opts=‘rw，local，dovolfs，dontbrowse，journaled，multilabel’）， sdiskpart（device=‘/dev/disk1s3’， mountpoint=‘/Volumes/Recovery’， fstype=‘apfs’， opts=‘rw，local，dovolfs，dontbrowse，journaled，multilabel’）］
# 磁碟的使用情況 磁碟總大小/已使用大小/空閒大小/負載
>>> psutil。disk_usage（‘/’）
sdiskusage（total=250790436864， used=10872418304， free=39636717568， percent=21。5）
# 磁碟IO 讀取次數/寫入次數/讀取資料/寫入資料/磁碟讀取所花費的時間/寫入磁碟所花費的時間
>>> psutil。disk_io_counters
sdiskio（read_count=26404943， write_count=11097500， read_bytes=609467826688， write_bytes=464322912256， read_time=7030486， write_time=2681553）
在監控平臺上每 2 秒請求 url 獲取磁碟資訊，並動態顯示圖表
disk_dict = {‘disk_time’：， ‘write_bytes’： ， ‘read_bytes’： ， ‘pre_write_bytes’： 0， ‘pre_read_bytes’： 0， ‘len’： -1}
def disk：
disk_usage = psutil。disk_usage（‘/’）
disk_used = 0
# 磁碟已使用大小 = 每個分割槽的總和
partitions = psutil。disk_partitions
for partition in partitions：
partition_disk_usage = psutil。disk_usage（partition［1］）
disk_used = partition_disk_usage。used + disk_used
now = time。strftime（‘%H：%M：%S’， time。localtime（time。time））
count = psutil。disk_io_counters
read_bytes = count。read_bytes
write_bytes = count。write_bytes
# 第一次請求
if disk_dict［‘len’］ == -1：
disk_dict［‘pre_write_bytes’］ = write_bytes
disk_dict［‘pre_read_bytes’］ = read_bytes
disk_dict［‘len’］ = 0
return disk_usage。total， disk_used， disk_usage。free
# 當前速率=現在寫入/讀取的總位元組-前一次請求寫入/讀取的總位元組
disk_dict［‘write_bytes’］。append（（write_bytes - disk_dict［‘pre_write_bytes’］）/1024）
disk_dict［‘read_bytes’］。append（（read_bytes - disk_dict［‘pre_read_bytes’］）/ 1024）
disk_dict［‘disk_time’］。append（now）
disk_dict［‘len’］ = disk_dict［‘len’］ + 1
# 把現在寫入/讀取的總位元組放入前一個請求的變數中
disk_dict［‘pre_write_bytes’］ = write_bytes
disk_dict［‘pre_read_bytes’］ = read_bytes
# 保持在圖表中 50 個數據
if disk_dict［‘len’］ == 51：
disk_dict［‘write_bytes’］。pop（0）
disk_dict［‘read_bytes’］。pop（0）
disk_dict［‘disk_time’］。pop（0）
disk_dict［‘len’］ = disk_dict［‘len’］ - 1
return disk_usage。total， disk_used， disk_usage。free
def disk_line -> Line：
total， used， free = disk
c = （
Line（init_opts=opts。InitOpts（width=“1680px”， height=“800px”））
。add_xaxis（xaxis_data=disk_dict［‘disk_time’］）
。add_yaxis（
series_name=“寫入資料”，
y_axis=disk_dict［‘write_bytes’］，
areastyle_opts=opts。AreaStyleOpts（opacity=0。5），
linestyle_opts=opts。LineStyleOpts，
label_opts=opts。LabelOpts（is_show=False），
）
。add_yaxis（
series_name=“讀取資料”，
y_axis=disk_dict［‘read_bytes’］，
yaxis_index=1，
areastyle_opts=opts。AreaStyleOpts（opacity=0。5），
linestyle_opts=opts。LineStyleOpts，
label_opts=opts。LabelOpts（is_show=False），
）
。extend_axis（
yaxis=opts。AxisOpts（
name_location=“start”，
type_=“value”，
is_inverse=True，
axistick_opts=opts。AxisTickOpts（is_show=True），
splitline_opts=opts。SplitLineOpts（is_show=True），
name=‘KB/2S’
）
）
。set_global_opts（
title_opts=opts。TitleOpts（
title=“磁碟IO”，
pos_left=“center”，
pos_top=“top”，
），
tooltip_opts=opts。TooltipOpts（trigger=“axis”， axis_pointer_type=“cross”），
legend_opts=opts。LegendOpts（pos_left=“left”），
xaxis_opts=opts。AxisOpts（type_=“category”， boundary_gap=False），
yaxis_opts=opts。AxisOpts（ type_=“value”， name=‘KB/2S’），
）
。set_series_opts（
axisline_opts=opts。AxisLineOpts，
）
）
return total， used， free， c
@app。route（“/disk”）
def get_disk_chart：
total， used， free， c = disk_line
return jsonify（{‘total’： total， ‘used’： used， ‘free’： free， ‘line’： c。dump_options_with_quotes}）
示例結果
網絡卡
透過 psutil 獲取網路介面和網路連線的資訊
# 獲取網路位元組數和包的個數 傳送的位元組數/收到的位元組數/傳送的包數/收到的包數
>>> psutil。net_io_counters
snetio（bytes_sent=9257984， bytes_recv=231398400， packets_sent=93319， packets_recv=189501， errin=0， errout=0， dropin=0， dropout=0）
# 獲取當前的網路連線 注意：net_connections 需要用管理員許可權執行 Python 檔案
>>> psutil。net_connections
［sconn（fd=6， family=
sconn（fd=12， family=
sconn（fd=6， family=
sconn（fd=10， family=
sconn（fd=11， family=
。。。
sconn（fd=30， family=
sconn（fd=31， family=
# 獲取網路介面資訊
>>> psutil。net_if_addrs
{‘lo0’： ［snicaddr（family=
snicaddr（family=
。。。，
‘utun1’： ［snicaddr（family=
# 獲取網路介面的狀態
>>> psutil。net_if_stats
{‘lo0’： snicstats（isup=True， duplex=
。。。
‘utun1’： snicstats（isup=True， duplex=
在監控平臺上每 2 秒請求 url 獲取網絡卡IO，並動態顯示圖表
net_io_dict = {‘net_io_time’：， ‘net_io_sent’： ， ‘net_io_recv’： ， ‘pre_sent’： 0， ‘pre_recv’： 0， ‘len’： -1}
def net_io：
now = time。strftime（‘%H：%M：%S’， time。localtime（time。time））
# 獲取網路資訊
count = psutil。net_io_counters
g_sent = count。bytes_sent
g_recv = count。bytes_recv
# 第一次請求
if net_io_dict［‘len’］ == -1：
net_io_dict［‘pre_sent’］ = g_sent
net_io_dict［‘pre_recv’］ = g_recv
net_io_dict［‘len’］ = 0
return
# 當前網路傳送/接收的位元組速率 = 現在網路傳送/接收的總位元組 - 前一次請求網路傳送/接收的總位元組
net_io_dict［‘net_io_sent’］。append（g_sent - net_io_dict［‘pre_sent’］）
net_io_dict［‘net_io_recv’］。append（g_recv - net_io_dict［‘pre_recv’］）
net_io_dict［‘net_io_time’］。append（now）
net_io_dict［‘len’］ = net_io_dict［‘len’］ + 1
net_io_dict［‘pre_sent’］ = g_sent
net_io_dict［‘pre_recv’］ = g_recv
# 保持在圖表中 10 個數據
if net_io_dict［‘len’］ == 11：
net_io_dict［‘net_io_sent’］。pop（0）
net_io_dict［‘net_io_recv’］。pop（0）
net_io_dict［‘net_io_time’］。pop（0）
net_io_dict［‘len’］ = net_io_dict［‘len’］ - 1
def net_io_line -> Line：
net_io
c = （
Line
。add_xaxis（net_io_dict［‘net_io_time’］）
。add_yaxis（“傳送位元組數”， net_io_dict［‘net_io_sent’］， is_smooth=True）
。add_yaxis（“接收位元組數”， net_io_dict［‘net_io_recv’］， is_smooth=True）
。set_series_opts（
areastyle_opts=opts。AreaStyleOpts（opacity=0。5），
label_opts=opts。LabelOpts（is_show=False），
）
。set_global_opts（
title_opts=opts。TitleOpts（title=“網絡卡IO/2秒”），
xaxis_opts=opts。AxisOpts（
axistick_opts=opts。AxisTickOpts（is_align_with_label=True），
is_scale=False，
boundary_gap=False，
），
））
return c
@app。route（“/netio”）
def get_net_io_chart：
c = net_io_line
return c。dump_options_with_quotes
示例結果
程序
透過 psutil 可以獲取所有程序的資訊
# 所有程序的 pid
>>> psutil。pids
［0， 1， 134， 135， 138， 139， 140， 141， 144， 145， 147， 152， 。。。， 30400， 97792］
# 單個程序
>>> p = psutil。Process（30400）
# 名稱
>>> p。name
‘pycharm’
# 使用記憶體負載
>>> p。memory_percent
12。838459014892578
# 啟動時間
>>> p。create_time
1587029962。493182
# 路徑
>>> p。exe
‘/Applications/PyCharm。app/Contents/MacOS/pycharm’
# 狀態
>>> p。status
‘running’
# 使用者名稱
>>> p。username
‘imeng’
# 記憶體資訊
>>> p。memory_info
pmem（rss=1093005312， vms=9914318848， pfaults=7813313， pageins=8448）
列出所有不需要許可權的程序
def process：
result =
process_list =
pid = psutil。pids
for k， i in enumerate（pid）：
try：
proc = psutil。Process（i）
ctime = time。strftime（“%Y-%m-%d %H：%M：%S”， time。localtime（proc。create_time））
process_list。append（（str（i）， proc。name， proc。cpu_percent， proc。memory_percent， ctime））
except psutil。AccessDenied：
# 需要管理員許可權
pass
except psutil。NoSuchProcess：
pass
except SystemError：
pass
# 按負載排序
process_list。sort（key=process_sort， reverse=True）
for i in process_list：
result。append（{‘PID’： i［0］， ‘name’： i［1］， ‘cpu’： i［2］， ‘mem’： “%。2f%%”%i［3］， ‘ctime’： i［4］}）
return jsonify（{‘list’， result}）
def process_sort（elem）：
return elem［3］
@app。route（“/process”）
def get_process_tab：
c = process
return c
@app。route（“/delprocess”）
def del_process：
pid = request。args。get（“pid”）
os。kill（int（pid）， signal。SIGKILL）
return jsonify（{‘status’： ‘OK’}）
示例結果
process。gif
總結
本文以 Psutil + Flask + Pyecharts + Bootstrap 開發一個簡單的系統監控平臺，可以算做是本公眾號內容的一個學以致用。在 Psutil 還有許多方法文章沒有列舉感興趣的小夥伴可以去嘗試並使用。
PS：
公號內回覆 ：Python，即可進入Python 新手學習交流群，一起100天計劃！
老規矩
，兄弟們還記得麼，，如果感覺文章內容不錯的話，記得分享朋友圈讓更多的人知道！
【程式碼
獲取方式
】
