很多時候，我們是需要檢視伺服器的網卡目前跑了多大流量，但對于網卡流量的查詢，在linux下似乎沒有像top那樣的原生指令。雖然top功能很強大，可以實時檢視cpu、記憶體、程序的動态，但是卻沒有對網卡流量的監控。既然沒有，那那就自己寫一個吧，哈哈。本文代碼綜合使用了psutil和curses子產品，腳本工具同時支援在linux和windows下使用，支援動态檢視網卡流量。當然現在也有些第三方的工具可以使用，比如iftop。
1.psutil子產品
psutil可以擷取系統的很多資訊，包括CPU，記憶體，磁盤，網絡，并且還可以檢視系統程序等資訊。psutil是一個跨平台的庫，支援linux、windows、freebsd、OSX等多個作業系統平台。
2.curses子產品
curses庫提供一個獨立于終端的螢幕顯示，支援各種控制代碼來執行常見的操作，如移動光标、滾動螢幕和擦除區域。curses庫将終端螢幕看成是由字元單元組成的坐标系，每一個單元由行坐标和列坐标來标示。坐标原點是螢幕的左上角，行坐标自上而下遞增，列坐标自左而右遞增。
3.實作思路
利用psutil擷取到網卡的上傳下載下傳流量，計算出目前網卡的流量速率。然後利用curses将流量資料動态的輸出到終端，實作top指令類似的動态重新整理效果。
4.代碼實作
# -*- coding:utf-8 -*-
"""
@Author: Rainbowhhy
@Date: 2020-08-01 18:18:18
"""
import psutil
import time
from datetime import datetime
import curses
import argparse
def getNetworkData():
# 擷取網卡流量資訊
recv = {}
sent = {}
data = psutil.net_io_counters(pernic=True)
interfaces = data.keys()
for interface in interfaces:
recv.setdefault(interface, data.get(interface).bytes_recv)
sent.setdefault(interface, data.get(interface).bytes_sent)
return interfaces, recv, sent
def getNetworkRate(num):
# 計算網卡流量速率
interfaces, oldRecv, oldSent = getNetworkData()
time.sleep(num)
interfaces, newRecv, newSent = getNetworkData()
networkIn = {}
networkOut = {}
for interface in interfaces:
networkIn.setdefault(interface, float("%.3f" % ((newRecv.get(interface) - oldRecv.get(interface)) / num)))
networkOut.setdefault(interface, float("%.3f" % ((newSent.get(interface) - oldSent.get(interface)) / num)))
return interfaces, networkIn, networkOut
def output(num, unit):
# 将監控輸出到終端
stdscr = curses.initscr()
curses.start_color()
# curses.init_pair(1, curses.COLOR_RED, curses.COLOR_WHITE)
curses.noecho()
curses.cbreak()
stdscr.clear()
try:
# 第一次初始化
interfaces, _, _ = getNetworkData()
currTime = datetime.now()
timeStr = datetime.strftime(currTime, "%Y-%m-%d %H:%M:%S")
stdscr.addstr(0, 0, timeStr)
i = 1
for interface in interfaces:
if interface != "lo" and bool(1 - interface.startswith("veth")) and bool(
1 - interface.startswith("藍牙")) and bool(1 - interface.startswith("VMware")):
if unit == "K" or unit == "k":
netIn = "%12.2fKB/s" % 0
netOut = "%11.2fKB/s" % 0
elif unit == "M" or unit == "m":
netIn = "%12.2fMB/s" % 0
netOut = "%11.2fMB/s" % 0
elif unit == "G" or unit == "g":
netIn = "%12.3fGB/s" % 0
netOut = "%11.3fGB/s" % 0
else:
netIn = "%12.1fB/s" % 0
netOut = "%11.1fB/s" % 0
stdscr.addstr(i, 0, interface)
stdscr.addstr(i + 1, 0, "Input:%s" % netIn)
stdscr.addstr(i + 2, 0, "Output:%s" % netOut)
stdscr.move(i + 3, 0)
i += 4
stdscr.refresh()
# 第二次開始循環監控網卡流量
while True:
_, networkIn, networkOut = getNetworkRate(num)
currTime = datetime.now()
timeStr = datetime.strftime(currTime, "%Y-%m-%d %H:%M:%S")
stdscr.erase()
stdscr.addstr(0, 0, timeStr)
i = 1
for interface in interfaces:
if interface != "lo" and bool(1 - interface.startswith("veth")) and bool(
1 - interface.startswith("藍牙")) and bool(1 - interface.startswith("VMware")):
if unit == "K" or unit == "k":
netIn = "%12.2fKB/s" % (networkIn.get(interface) / 1024)
netOut = "%11.2fKB/s" % (networkOut.get(interface) / 1024)
elif unit == "M" or unit == "m":
netIn = "%12.2fMB/s" % (networkIn.get(interface) / 1024 / 1024)
netOut = "%11.2fMB/s" % (networkOut.get(interface) / 1024 / 1024)
elif unit == "G" or unit == "g":
netIn = "%12.3fGB/s" % (networkIn.get(interface) / 1024 / 1024 / 1024)
netOut = "%11.3fGB/s" % (networkOut.get(interface) / 1024 / 1024 / 1024)
else:
netIn = "%12.1fB/s" % networkIn.get(interface)
netOut = "%11.1fB/s" % networkOut.get(interface)
stdscr.addstr(i, 0, interface)
stdscr.addstr(i + 1, 0, "Input:%s" % netIn)
stdscr.addstr(i + 2, 0, "Output:%s" % netOut)
stdscr.move(i + 3, 0)
i += 4
stdscr.refresh()
except KeyboardInterrupt:
# 還原終端
curses.echo()
curses.nocbreak()
curses.endwin()
except Exception as e:
curses.echo()
curses.nocbreak()
curses.endwin()
print("ERROR: %s!" % e)
print("Please increase the terminal size!")
finally:
curses.echo()
curses.nocbreak()
curses.endwin()
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="A command for monitoring the traffic of network interface! Ctrl + C: exit")
parser.add_argument("-t", "--time", type=int, help="the interval time for ouput", default=1)
parser.add_argument("-u", "--unit", type=str, choices=["b", "B", "k", "K", "m", "M", "g", "G"],
help="the unit for ouput", default="B")
parser.add_argument("-v", "--version", help="output version information and exit", action="store_true")
args = parser.parse_args()
if args.version:
print("v1.0")
exit(0)
num = args.time
unit = args.unit
output(num, unit)
5.用法說明
1.僅支援python3。
2.需安裝psutil庫，windows需安裝curses庫，linux預設自帶，如果沒有也需安裝。
3.支援指定重新整理時間間隔，-t或者--time。
4.支援指定流量輸出機關，-u或者--unit。
5.使用示例：
(1)每5秒重新整理一次資料，流量機關為M/s
python3 network.py -t 5 -u M或者python3 network.py -t 5 -u m
(2)預設不指定參數，表示每一秒重新整理一次資料，流量機關為B/s
python3 network.py
6.實作效果
linux下的效果
![作業系統（python）多程序學習[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)