如何编写连通性检测脚本

本文目录导读：

1. 基础Ping检测脚本（Bash）
2. 高级连通性检测脚本（Python）
3. 使用示例
4. Docker版连通性检测
5. 企业级连通性检测（带日志和告警）

我来帮你编写一个网络连通性检测脚本,这里提供几个常用版本：

基础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()

这些脚本可以根据你的具体需求进行修改和扩展,选择适合你使用场景的版本即可。