本文目录导读：
我来帮你编写一个网络连通性检测脚本,这里提供几个常用版本：
基础Ping检测脚本（Bash）
#!/bin/bash
# 基础连通性检测脚本
# 用法: ./connectivity_check.sh
# 要检测的主机列表
HOSTS=(
"8.8.8.8"
"www.baidu.com"
"192.168.1.1"
)
echo "开始检测网络连通性..."
echo "========================="
for host in "${HOSTS[@]}"; do
if ping -c 3 -W 2 "$host" &> /dev/null; then
echo "✓ $host 可达 (延迟: $(ping -c 1 "$host" | grep 'time=' | awk -F'time=' '{print $2}' | awk '{print $1}'))"
else
echo "✗ $host 不可达"
fi
done
echo "========================="
echo "检测完成"
高级连通性检测脚本（Python）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import subprocess
import platform
import time
import socket
import argparse
from concurrent.futures import ThreadPoolExecutor
class ConnectivityChecker:
def __init__(self, timeout=3, count=3):
self.timeout = timeout
self.count = count
self.system = platform.system().lower()
def ping_host(self, host):
"""Ping检测主机连通性"""
param = '-n' if self.system == 'windows' else '-c'
timeout_param = '-w' if self.system == 'windows' else '-W'
try:
cmd = ['ping', param, str(self.count),
timeout_param, str(self.timeout), host]
result = subprocess.run(cmd, capture_output=True, text=True,
timeout=self.timeout * 2)
# 解析延迟
if result.returncode == 0:
if 'time=' in result.stdout:
latency = result.stdout.split('time=')[1].split(' ')[0]
return {'host': host, 'status': 'reachable', 'latency': latency}
return {'host': host, 'status': 'reachable', 'latency': 'unknown'}
else:
return {'host': host, 'status': 'unreachable', 'latency': None}
except subprocess.TimeoutExpired:
return {'host': host, 'status': 'timeout', 'latency': None}
except Exception as e:
return {'host': host, 'status': f'error: {str(e)}', 'latency': None}
def tcp_check(self, host, port):
"""TCP端口连通性检测"""
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(self.timeout)
result = sock.connect_ex((host, port))
sock.close()
if result == 0:
return {'host': host, 'port': port, 'status': 'open'}
else:
return {'host': host, 'port': port, 'status': 'closed'}
except Exception as e:
return {'host': host, 'port': port, 'status': f'error: {str(e)}'}
def check_multiple_hosts(self, hosts, method='ping'):
"""并发检测多个主机"""
with ThreadPoolExecutor(max_workers=5) as executor:
if method == 'ping':
results = list(executor.map(self.ping_host, hosts))
else:
results = list(executor.map(
lambda x: self.tcp_check(x[0], x[1]), hosts))
return results
def main():
parser = argparse.ArgumentParser(description='网络连通性检测工具')
parser.add_argument('-t', '--targets', nargs='+',
help='要检测的目标主机或IP列表')
parser.add_argument('-p', '--ports', nargs='+', type=int,
help='要检测的端口号')
parser.add_argument('--timeout', type=int, default=3,
help='超时时间(秒)')
parser.add_argument('--count', type=int, default=3,
help='Ping次数')
args = parser.parse_args()
checker = ConnectivityChecker(timeout=args.timeout, count=args.count)
# 默认检测列表
default_hosts = ['8.8.8.8', '1.1.1.1', 'www.baidu.com', '192.168.1.1']
targets = args.targets if args.targets else default_hosts
print(f"\n{'='*50}")
print(f"网络连通性检测报告")
print(f"检测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print(f"{'='*50}\n")
if args.ports:
print("TCP端口检测:")
targets = [(host, port) for host in targets for port in args.ports]
results = checker.check_multiple_hosts(targets, method='tcp')
for result in results:
icon = '✓' if result['status'] == 'open' else '✗'
print(f"{icon} {result['host']}:{result['port']} - {result['status']}")
else:
print("Ping检测:")
results = checker.check_multiple_hosts(targets, method='ping')
for result in results:
if result['status'] == 'reachable':
print(f"✓ {result['host']} - 可达 (延迟: {result['latency']})")
else:
print(f"✗ {result['host']} - {result['status']}")
print(f"\n{'='*50}")
print(f"检测完成")
if __name__ == "__main__":
main()
使用示例
Bash脚本使用：
# 保存脚本 chmod +x connectivity_check.sh ./connectivity_check.sh
Python脚本使用：
# 基本使用 python3 connectivity_check.py # 检测指定主机 python3 connectivity_check.py -t google.com github.com # 检测端口 python3 connectivity_check.py -t 192.168.1.1 -p 80 443 # 设置超时 python3 connectivity_check.py -t example.com --timeout 5
Docker版连通性检测
FROM alpine:latest RUN apk add --no-cache bash iputils net-tools curl COPY connectivity_check.sh /check.sh RUN chmod +x /check.sh CMD ["/check.sh"]
企业级连通性检测（带日志和告警）
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import json
import logging
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
import sys
sys.path.append('/path/to/your/script')
from connectivity_check import ConnectivityChecker
class EnterpriseConnectivityChecker:
def __init__(self, config_file='config.json'):
self.checker = ConnectivityChecker()
self.setup_logging()
self.load_config(config_file)
def setup_logging(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('connectivity.log'),
logging.StreamHandler(sys.stdout)
]
)
self.logger = logging.getLogger(__name__)
def load_config(self, config_file):
with open(config_file, 'r') as f:
self.config = json.load(f)
def send_alert(self, failures):
"""发送告警邮件"""
if not self.config.get('alert_email'):
return
msg = MIMEText(f"""
网络连通性故障报告
时间: {datetime.now()}
故障详情:
{json.dumps(failures, indent=2)}
""")
msg['Subject'] = f'网络连通性告警 - {datetime.now().strftime("%Y-%m-%d %H:%M")}'
msg['From'] = self.config['alert_email']['from']
msg['To'] = self.config['alert_email']['to']
with smtplib.SMTP(self.config['alert_email']['smtp_server']) as server:
server.send_message(msg)
def run_check(self):
"""执行检测任务"""
hosts = self.config.get('hosts', ['8.8.8.8', 'www.baidu.com'])
self.logger.info(f"开始连通性检测: {', '.join(hosts)}")
failures = []
for host in hosts:
result = self.checker.ping_host(host)
if result['status'] != 'reachable':
failures.append(result)
self.logger.warning(f"主机不可达: {host}")
else:
self.logger.info(f"主机可达: {host} (延迟: {result['latency']})")
if failures:
self.send_alert(failures)
self.logger.error(f"检测完成，发现 {len(failures)} 个故障")
else:
self.logger.info("所有主机均可达")
if __name__ == "__main__":
checker = EnterpriseConnectivityChecker()
checker.run_check()
这些脚本可以根据你的具体需求进行修改和扩展,选择适合你使用场景的版本即可。
