很多時候，我們是需要查看服務器的網卡當前跑了多大流量，但對於網卡流量的查詢，在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.代碼實現
1 # -*- coding:utf-8 -*- 2 3 """ 4 @Author: Rainbowhhy 5 @Date: 2020-08-01 18:18:18 6 """ 7 8 import psutil 9 import time 10 from datetime import datetime 11 import curses 12 import argparse 13 14 15 def getNetworkData(): 16 # 獲取網卡流量信息 17 recv = {} 18 sent = {} 19 data = psutil.net_io_counters(pernic=True) 20 interfaces = data.keys() 21 for interface in interfaces: 22 recv.setdefault(interface, data.get(interface).bytes_recv) 23 sent.setdefault(interface, data.get(interface).bytes_sent) 24 return interfaces, recv, sent 25 26 27 def getNetworkRate(num): 28 # 計算網卡流量速率 29 interfaces, oldRecv, oldSent = getNetworkData() 30 time.sleep(num) 31 interfaces, newRecv, newSent = getNetworkData() 32 networkIn = {} 33 networkOut = {} 34 for interface in interfaces: 35 networkIn.setdefault(interface, float("%.3f" % ((newRecv.get(interface) - oldRecv.get(interface)) / num))) 36 networkOut.setdefault(interface, float("%.3f" % ((newSent.get(interface) - oldSent.get(interface)) / num))) 37 return interfaces, networkIn, networkOut 38 39 40 def output(num, unit): 41 # 將監控輸出到終端 42 stdscr = curses.initscr() 43 curses.start_color() 44 # curses.init_pair(1, curses.COLOR_RED, curses.COLOR_WHITE) 45 curses.noecho() 46 curses.cbreak() 47 stdscr.clear() 48 try: 49 # 第一次初始化 50 interfaces, _, _ = getNetworkData() 51 currTime = datetime.now() 52 timeStr = datetime.strftime(currTime, "%Y-%m-%d %H:%M:%S") 53 stdscr.addstr(0, 0, timeStr) 54 i = 1 55 for interface in interfaces: 56 if interface != "lo" and bool(1 - interface.startswith("veth")) and bool( 57 1 - interface.startswith("br")) and bool( 58 1 - interface.startswith("藍牙")) and bool(1 - interface.startswith("VMware")): 59 if unit == "K" or unit == "k": 60 netIn = "%12.2fKB/s" % 0 61 netOut = "%11.2fKB/s" % 0 62 elif unit == "M" or unit == "m": 63 netIn = "%12.2fMB/s" % 0 64 netOut = "%11.2fMB/s" % 0 65 elif unit == "G" or unit == "g": 66 netIn = "%12.3fGB/s" % 0 67 netOut = "%11.3fGB/s" % 0 68 else: 69 netIn = "%12.1fB/s" % 0 70 netOut = "%11.1fB/s" % 0 71 stdscr.addstr(i, 0, interface) 72 stdscr.addstr(i + 1, 0, "Input:%s" % netIn) 73 stdscr.addstr(i + 2, 0, "Output:%s" % netOut) 74 stdscr.move(i + 3, 0) 75 i += 4 76 stdscr.refresh() 77 # 第二次開始循環監控網卡流量 78 while True: 79 _, networkIn, networkOut = getNetworkRate(num) 80 currTime = datetime.now() 81 timeStr = datetime.strftime(currTime, "%Y-%m-%d %H:%M:%S") 82 stdscr.erase() 83 stdscr.addstr(0, 0, timeStr) 84 i = 1 85 for interface in interfaces: 86 if interface != "lo" and bool(1 - interface.startswith("veth")) and bool( 87 1 - interface.startswith("br")) and bool( 88 1 - interface.startswith("藍牙")) and bool(1 - interface.startswith("VMware")): 89 if unit == "K" or unit == "k": 90 netIn = "%12.2fKB/s" % (networkIn.get(interface) / 1024) 91 netOut = "%11.2fKB/s" % (networkOut.get(interface) / 1024) 92 elif unit == "M" or unit == "m": 93 netIn = "%12.2fMB/s" % (networkIn.get(interface) / 1024 / 1024) 94 netOut = "%11.2fMB/s" % (networkOut.get(interface) / 1024 / 1024) 95 elif unit == "G" or unit == "g": 96 netIn = "%12.3fGB/s" % (networkIn.get(interface) / 1024 / 1024 / 1024) 97 netOut = "%11.3fGB/s" % (networkOut.get(interface) / 1024 / 1024 / 1024) 98 else: 99 netIn = "%12.1fB/s" % networkIn.get(interface) 100 netOut = "%11.1fB/s" % networkOut.get(interface) 101 stdscr.addstr(i, 0, interface) 102 stdscr.addstr(i + 1, 0, "Input:%s" % netIn) 103 stdscr.addstr(i + 2, 0, "Output:%s" % netOut) 104 stdscr.move(i + 3, 0) 105 i += 4 106 stdscr.refresh() 107 except KeyboardInterrupt: 108 # 還原終端 109 curses.echo() 110 curses.nocbreak() 111 curses.endwin() 112 except Exception as e: 113 curses.echo() 114 curses.nocbreak() 115 curses.endwin() 116 print("ERROR: %s!" % e) 117 print("Please increase the terminal size!") 118 finally: 119 curses.echo() 120 curses.nocbreak() 121 curses.endwin() 122 123 124 if __name__ == "__main__": 125 parser = argparse.ArgumentParser( 126 description="A command for monitoring the traffic of network interface! Ctrl + C: exit") 127 parser.add_argument("-t", "--time", type=int, help="the interval time for ouput", default=1) 128 parser.add_argument("-u", "--unit", type=str, choices=["b", "B", "k", "K", "m", "M", "g", "G"], 129 help="the unit for ouput", default="B") 130 parser.add_argument("-v", "--version", help="output version information and exit", action="store_true") 131 args = parser.parse_args() 132 if args.version: 133 print("v1.0") 134 exit(0) 135 num = args.time 136 unit = args.unit 137 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下的效果
windows下的效果
號外
可以使用工具將腳本打包成二進制文件，之后可以像執行linux命令那樣直接執行，無需安裝依賴包。
networkstat下載
目前支持Ubuntu14.04以上，CentOS7以上，Debian8以上，Windows10，Windows2008R2以上。