溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
這篇文章主要介紹了Python怎么實現監控遠程主機實時數據的相關知識，內容詳細易懂，操作簡單快捷，具有一定借鑒價值，相信大家閱讀完這篇Python怎么實現監控遠程主機實時數據文章都會有所收獲，下面我們一起來看看吧。
本程序為一個基于TCP協議的服務端程序，可以接收客戶端發送的指令并執行相應的操作，最終將操作結果返回給客戶端。程序運行在localhost（即本機）的8888端口。
主要功能及指令：
• 獲取CPU使用率：指令“cpu”
• 獲取內存使用率：指令“memory”
• 獲取網絡帶寬信息：指令“network”
• 獲取當前登錄用戶：指令“user”
• 獲取系統負載情況：指令“loadavg”
• 獲取當前時間：指令“time”
• 獲取進程列表：指令“process”
• 獲取系統信息：指令“system”
• 獲取網絡連接列表：指令“connection”
• 獲取GPU使用情況：指令“gpu”
• 獲取磁盤使用情況：指令“disk”
對于不同的指令，程序采用不同的庫函數進行數據獲取和處理，具體如下：
• 對于指令“cpu”，使用psutil庫獲取CPU使用率。
• 對于指令“memory”，使用psutil庫獲取內存使用率。
• 對于指令“network”，使用speedtest-cli庫獲取網絡帶寬信息。
• 對于指令“user”，使用psutil庫獲取當前登錄用戶。
• 對于指令“loadavg”，使用os庫獲取系統負載情況。
• 對于指令“time”，使用datetime庫獲取當前時間。
• 對于指令“process”，使用psutil庫獲取進程列表。程序將進程按照內存使用量排序，只返回前10個進程。
• 對于指令“system”，使用platform庫獲取系統信息。
• 對于指令“connection”，使用psutil庫獲取網絡連接列表。程序將連接按照進程ID排序，只返回前10個連接。
• 對于指令“gpu”，使用nvidia-smi工具獲取GPU使用情況。
• 對于指令“disk”，使用psutil庫獲取磁盤使用情況。
每個指令的處理結果均以字符串形式返回給客戶端。在處理指令時，程序會判斷接收到的數據是否為空。如果為空，則斷開連接并關閉客戶端套接字。程序中的handle_client函數是負責與單個客戶端通信的線程函數，每個客戶端連接都會啟動一個線程。
該程序是一個簡單的TCP客戶端，可以連接到一個服務器并向其發送請求，然后等待服務器的響應。程序使用Python的socket模塊來創建TCP套接字，并使用connect方法連接到一個服務器地址。程序通過循環不斷地等待用戶輸入請求數據類型，然后將請求數據類型編碼成UTF-8格式并使用sendall方法將其發送給服務器。接著程序等待服務器響應，并將響應數據解碼成UTF-8格式并打印出來。最后，程序關閉客戶端套接字。
程序的主要流程如下：
1. 導入socket模塊。
2. 創建一個TCP套接字對象。
3. 連接到指定的服務器地址。
4. 循環等待用戶輸入請求數據類型。
5. 將請求數據類型編碼成UTF-8格式并發送給服務器。
6. 等待服務器響應并接收響應數據。
7. 將響應數據解碼成UTF-8格式并打印出來。
8. 關閉客戶端套接字。
該程序可以用于與提供特定數據類型的服務器進行通信。用戶可以輸入不同的請求數據類型來獲取不同類型的數據。服務器將根據請求類型返回相應的數據。程序的運行依賴于服務器的可用性和響應速度。如果服務器無法響應，程序將一直等待，直到服務器響應或者程序被中斷。
服務端
import os
import socket
import subprocess
import threading
from datetime import datetime
from sys import platform
import psutil
from speedtest import Speedtest
# 創建TCP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 綁定IP和端口
server_address = ('localhost', 8888)
server_socket.bind(server_address)
# 監聽連接請求
server_socket.listen(5)
def handle_client(client_socket, client_address):
while True:
# 接收客戶端發送的數據
data = client_socket.recv(1024)
# 如果接收到空數據，則斷開連接
if not data:
client_socket.close()
print(f"Connection with {client_address} closed")
break
# 處理接收到的數據
request = data.decode('utf-8')
if request == 'cpu':
# 使用psutil庫獲取CPU使用率
cpu_percent = psutil.cpu_percent(interval=1)
response_data = f'CPU使用率：{cpu_percent}%'.encode('utf-8')
elif request == 'memory':
# 使用psutil庫獲取內存使用率
memory_percent = psutil.virtual_memory().percent
response_data = f'內存使用率：{memory_percent}%'.encode('utf-8')
elif request == 'network':
# 使用speedtest-cli庫獲取網絡帶寬信息
st = Speedtest()
download_speed = st.download()
upload_speed = st.upload()
response_data = f'下載速度：{download_speed / 1000000}Mbps，上傳速度：{upload_speed / 1000000}Mbps'.encode('utf-8')
elif request == 'user':
# 使用psutil庫獲取當前登錄用戶
username = psutil.users()[0].name
response_data = f'當前登錄用戶：{username}'.encode('utf-8')
elif request == 'loadavg':
# 使用os庫獲取系統負載情況
load_avg = os.getloadavg()
response_data = f'系統負載情況：{load_avg}'.encode('utf-8')
elif request == 'time':
# 使用datetime庫獲取當前時間
current_time = datetime.datetime.now()
response_data = f'當前時間：{current_time}'.encode('utf-8')
elif request == 'process':
# 使用psutil庫獲取進程列表
process_list = []
for process in psutil.process_iter(['pid', 'name', 'memory_info']):
try:
process_list.append((process.info['pid'], process.info['name'], process.info['memory_info'].rss))
except (psutil.AccessDenied, psutil.NoSuchProcess):
pass
process_list.sort(key=lambda x: x[2], reverse=True)
response_data = ''
for i, (pid, name, memory) in enumerate(process_list[:10]):
response_data += f'{i + 1}. 進程名稱：{name}，進程ID：{pid}，占用內存：{memory / 1024 / 1024:.2f}MB\n'
response_data = response_data.encode('utf-8')
elif request == 'system':
# 使用platform庫獲取系統信息
system_info = f'操作系統：{platform.system()} {platform.release()}\n處理器：{platform.processor()}\nPython版本：{platform.python_version()}'
response_data = system_info.encode('utf-8')
elif request == 'connection':
# 使用psutil庫獲取網絡連接列表
conn_list = []
for conn in psutil.net_connections():
if conn.status == psutil.CONN_ESTABLISHED:
conn_list.append((conn.laddr.ip, conn.laddr.port, conn.raddr.ip, conn.raddr.port, conn.pid))
conn_list.sort(key=lambda x: x[4])
response_data = ''
for i, (laddr_ip, laddr_port, raddr_ip, raddr_port, pid) in enumerate(conn_list[:10]):
response_data += f'{i + 1}. 本地地址：{laddr_ip}:{laddr_port}，遠程地址：{raddr_ip}:{raddr_port}，進程ID：{pid}\n'
response_data = response_data.encode('utf-8')
elif request == 'disk':
# 使用psutil庫獲取磁盤使用情況
disk_usage = psutil.disk_usage('/')
disk_info = f'磁盤總容量：{disk_usage.total / 1024 / 1024 / 1024:.2f}GB，已用容量：{disk_usage.used / 1024 / 1024 / 1024:.2f}GB，可用容量：{disk_usage.free / 1024 / 1024 / 1024:.2f}GB'
response_data = disk_info.encode('utf-8')
elif request == 'load':
# 使用psutil庫獲取系統負載
load_avg = psutil.getloadavg()
load_info = f'1分鐘內平均負載：{load_avg[0]:.2f}，5分鐘內平均負載：{load_avg[1]:.2f}，15分鐘內平均負載：{load_avg[2]:.2f}'
response_data = load_info.encode('utf-8')
elif request == 'thread':
# 使用psutil庫獲取進程線程數
thread_info = f'當前進程線程數：{psutil.Process().num_threads()}'
response_data = thread_info.encode('utf-8')
else:
response_data = b'Invalid request'
# 發送響應數據
client_socket.sendall(response_data)
# 接收多個客戶端連接
while True:
client_socket, client_address = server_socket.accept()
print(f"New connection from {client_address}")
# 創建新線程處理客戶端連接
client_thread = threading.Thread(target=handle_client, args=(client_socket, client_address))
client_thread.start()客戶端
import socket
# 創建TCP套接字
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 連接服務器
server_address = ('localhost', 8888)
client_socket.connect(server_address)
while True:
# 發送請求數據給服務器
request = input("請輸入要請求的數據類型（cpu/memory/network/user/loadavg/time/process/system/connection/disk/load/thread）：")
client_socket.sendall(request.encode('utf-8'))
# 接收服務器響應數據
response_data = client_socket.recv(1024)
# 處理接收到的數據
response = response_data.decode('utf-8')
print(response)
# 關閉客戶端套接字
client_socket.close()關于“Python怎么實現監控遠程主機實時數據”這篇文章的內容就介紹到這里，感謝各位的閱讀！相信大家對“Python怎么實現監控遠程主機實時數據”知識都有一定的了解，大家如果還想學習更多知識，歡迎關注億速云行業資訊頻道。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
