怎样实现连接超时优化脚本

wen 实用脚本 2026-06-27 121

本文目录导读：

怎样实现连接超时优化脚本

基础超时处理脚本
Shell脚本版本
配置文件示例
使用说明

我来提供一个全面的连接超时优化脚本实现方案,包含多种优化策略：

基础超时处理脚本

Python版本

#!/usr/bin/env python3
"""
连接超时优化脚本
支持HTTP、TCP、数据库连接等多种场景
"""
import socket
import requests
import time
import logging
from functools import wraps
import asyncio
import aiohttp
from typing import Callable, Any
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class TimeoutOptimizer:
    """连接超时优化器"""
    def __init__(self, connect_timeout=5, read_timeout=10, retry_count=3):
        self.connect_timeout = connect_timeout
        self.read_timeout = read_timeout
        self.retry_count = retry_count
    def sync_retry_decorator(self, func: Callable) -> Callable:
        """同步重试装饰器"""
        @wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(self.retry_count):
                try:
                    return func(*args, **kwargs)
                except (socket.timeout, requests.Timeout) as e:
                    last_exception = e
                    wait_time = 2 ** attempt  # 指数退避
                    logger.warning(f"连接超时，第{attempt+1}次重试，等待{wait_time}秒")
                    time.sleep(wait_time)
                except Exception as e:
                    raise e
            raise last_exception
        return wrapper
    def async_retry_decorator(self, func: Callable) -> Callable:
        """异步重试装饰器"""
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(self.retry_count):
                try:
                    return await func(*args, **kwargs)
                except (asyncio.TimeoutError, aiohttp.ClientTimeout) as e:
                    last_exception = e
                    wait_time = 2 ** attempt
                    logger.warning(f"异步连接超时，第{attempt+1}次重试")
                    await asyncio.sleep(wait_time)
                except Exception as e:
                    raise e
            raise last_exception
        return wrapper
# HTTP请求优化
@TimeoutOptimizer(connect_timeout=3, read_timeout=10).sync_retry_decorator
def optimized_http_request(url: str, method: str = 'GET', **kwargs) -> requests.Response:
    """优化的HTTP请求"""
    session = requests.Session()
    # 配置连接池
    adapter = requests.adapters.HTTPAdapter(
        pool_connections=10,
        pool_maxsize=20,
        max_retries=3,
        pool_block=False
    )
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    # 设置超时
    kwargs['timeout'] = kwargs.get('timeout', (3, 10))
    response = session.request(method, url, **kwargs)
    response.raise_for_status()
    return response
# 数据库连接优化（以MySQL为例）
def optimized_mysql_connection(config: dict) -> Any:
    """优化的MySQL连接"""
    try:
        import pymysql
        import dns.resolver
        # DNS解析优化
        start_time = time.time()
        try:
            answers = dns.resolver.resolve(config['host'], 'A')
            config['host'] = str(answers[0])
        except:
            pass  # 使用原始host
        logger.info(f"DNS解析耗时: {time.time() - start_time:.2f}s")
        # 连接配置
        connection = pymysql.connect(
            host=config['host'],
            user=config['user'],
            password=config['password'],
            database=config['database'],
            connect_timeout=config.get('connect_timeout', 5),
            read_timeout=config.get('read_timeout', 30),
            max_allowed_packet=16*1024*1024,  # 16MB
            autocommit=True
        )
        # 设置会话超时
        with connection.cursor() as cursor:
            cursor.execute("SET SESSION wait_timeout=28800")
            cursor.execute("SET SESSION interactive_timeout=28800")
        return connection
    except Exception as e:
        logger.error(f"数据库连接失败: {e}")
        raise
# TCP连接优化
def optimized_tcp_connection(host: str, port: int, timeout: int = 5) -> socket.socket:
    """优化的TCP连接"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    # TCP优化参数
    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
    if hasattr(socket, 'TCP_KEEPIDLE'):
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 60)
    if hasattr(socket, 'TCP_KEEPINTVL'):
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 10)
    if hasattr(socket, 'TCP_KEEPCNT'):
        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 3)
    # 连接超时优化
    sock.settimeout(timeout)
    sock.connect((host, port))
    # 设置非阻塞模式
    sock.setblocking(0)
    return sock
# 异步HTTP请求优化
async def optimized_async_http_request(url: str, session: aiohttp.ClientSession = None) -> dict:
    """优化的异步HTTP请求"""
    if session is None:
        timeout = aiohttp.ClientTimeout(total=30, connect=5)
        connector = aiohttp.TCPConnector(
            limit=100,
            limit_per_host=20,
            ttl_dns_cache=300,
            enable_cleanup_closed=True
        )
        async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
            return await _make_request(session, url)
    else:
        return await _make_request(session, url)
async def _make_request(session: aiohttp.ClientSession, url: str) -> dict:
    """内部请求方法"""
    try:
        async with session.get(url) as response:
            data = await response.json()
            return {
                'status': response.status,
                'data': data,
                'headers': dict(response.headers)
            }
    except asyncio.TimeoutError:
        logger.error(f"异步请求超时: {url}")
        raise
    except Exception as e:
        logger.error(f"异步请求失败: {e}")
        raise
# 连接池管理器
class ConnectionPoolManager:
    """连接池管理器"""
    def __init__(self, max_connections: int = 20):
        self.max_connections = max_connections
        self.connections = []
        self.lock = asyncio.Lock()
    async def get_connection(self) -> socket.socket:
        """获取连接"""
        async with self.lock:
            # 清理过期连接
            self._cleanup_connections()
            if self.connections:
                return self.connections.pop()
            # 创建新连接
            return await self._create_connection()
    async def release_connection(self, conn: socket.socket):
        """释放连接"""
        async with self.lock:
            if len(self.connections) < self.max_connections:
                self.connections.append(conn)
            else:
                conn.close()
    def _cleanup_connections(self):
        """清理过期连接"""
        self.connections = [c for c in self.connections if self._is_connected(c)]
    def _is_connected(self, sock: socket.socket) -> bool:
        """检查连接是否有效"""
        try:
            sock.settimeout(0)
            sock.send(b'')
            return True
        except:
            return False
    async def _create_connection(self) -> socket.socket:
        """创建新连接"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(5)
        return sock
# 主函数示例
async def main():
    """主函数"""
    optimizer = TimeoutOptimizer()
    # 测试同步HTTP请求
    try:
        response = optimized_http_request('https://api.example.com/data')
        print(f"同步请求成功: {response.status_code}")
    except Exception as e:
        print(f"同步请求失败: {e}")
    # 测试异步请求
    try:
        result = await optimized_async_http_request('https://api.example.com/data')
        print(f"异步请求成功: {result['status']}")
    except Exception as e:
        print(f"异步请求失败: {e}")
    # 测试TCP连接
    try:
        tcp_conn = optimized_tcp_connection('example.com', 443)
        print("TCP连接成功")
        tcp_conn.close()
    except Exception as e:
        print(f"TCP连接失败: {e}")
if __name__ == "__main__":
    asyncio.run(main())

Shell脚本版本

#!/bin/bash
# 连接超时优化脚本
# 配置
CONNECT_TIMEOUT=5
MAX_RETRIES=3
RETRY_DELAY=2
# 颜色输出
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m' # No Color
# 日志函数
log_info() {
    echo -e "${GREEN}[INFO]${NC} $1"
}
log_warn() {
    echo -e "${YELLOW}[WARN]${NC} $1"
}
log_error() {
    echo -e "${RED}[ERROR]${NC} $1"
}
# HTTP连接检测
check_http_connection() {
    local url=$1
    local retry=$2
    if [ -z "$retry" ]; then
        retry=0
    fi
    log_info "检测HTTP连接: $url"
    # 使用curl检测连接
    response=$(curl -s -o /dev/null -w "%{http_code}" \
        --connect-timeout $CONNECT_TIMEOUT \
        --max-time 30 \
        --retry $MAX_RETRIES \
        --retry-delay $RETRY_DELAY \
        "$url" 2>/dev/null)
    if [ $? -eq 0 ] && [ "$response" -ge 200 ] && [ "$response" -lt 400 ]; then
        log_info "连接成功，HTTP状态码: $response"
        return 0
    else
        log_warn "连接失败，HTTP状态码: $response"
        return 1
    fi
}
# TCP端口检测
check_tcp_connection() {
    local host=$1
    local port=$2
    log_info "检测TCP连接: $host:$port"
    # 使用nc或timeout检测
    if command -v nc &> /dev/null; then
        timeout $CONNECT_TIMEOUT nc -zv "$host" "$port" 2>/dev/null
    else
        timeout $CONNECT_TIMEOUT bash -c "echo >/dev/tcp/$host/$port" 2>/dev/null
    fi
    if [ $? -eq 0 ]; then
        log_info "TCP连接成功"
        return 0
    else
        log_error "TCP连接失败"
        return 1
    fi
}
# DNS解析优化
optimize_dns() {
    local host=$1
    log_info "优化DNS解析: $host"
    # 配置DNS服务器
    echo "nameserver 8.8.8.8" > /etc/resolv.conf
    echo "nameserver 8.8.4.4" >> /etc/resolv.conf
    # 测试DNS解析
    dig_output=$(dig +short "$host" 2>/dev/null)
    if [ -n "$dig_output" ]; then
        log_info "DNS解析成功: $dig_output"
    else
        log_warn "DNS解析失败，尝试备用DNS"
        echo "nameserver 114.114.114.114" > /etc/resolv.conf
    fi
    # 缓存DNS结果
    if command -v nscd &> /dev/null; then
        nscd -i hosts 2>/dev/null
    fi
}
# 系统参数优化
optimize_system_params() {
    log_info "优化系统网络参数..."
    # TCP参数优化
    sysctl -w net.ipv4.tcp_fin_timeout=30 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_tw_reuse=1 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_tw_recycle=1 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_keepalive_time=120 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_keepalive_intvl=30 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_keepalive_probes=3 >/dev/null 2>&1
    # 连接队列优化
    sysctl -w net.core.somaxconn=65535 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_max_syn_backlog=8192 >/dev/null 2>&1
    # 超时参数优化
    sysctl -w net.ipv4.tcp_syn_retries=2 >/dev/null 2>&1
    sysctl -w net.ipv4.tcp_synack_retries=2 >/dev/null 2>&1
    log_info "系统参数优化完成"
}
# 连接测试函数
test_connections() {
    local target=$1
    local port=$2
    log_info "开始连接测试..."
    # 测试TCP连接
    check_tcp_connection "$target" "$port"
    tcp_result=$?
    # 如果是HTTP/HTTPS端口，测试HTTP连接
    if [ "$port" = "80" ] || [ "$port" = "443" ]; then
        check_http_connection "http://$target:$port"
        http_result=$?
    fi
    # 重试逻辑
    if [ $tcp_result -ne 0 ]; then
        log_warn "连接失败，开始重试..."
        for i in $(seq 1 $MAX_RETRIES); do
            sleep $((RETRY_DELAY * i))
            check_tcp_connection "$target" "$port"
            if [ $? -eq 0 ]; then
                log_info "重试成功"
                return 0
            fi
        done
        log_error "所有重试均失败"
        return 1
    fi
    return 0
}
# 主函数
main() {
    echo "=========================================="
    echo "    连接超时优化脚本 v1.0"
    echo "=========================================="
    echo ""
    # 检查参数
    if [ $# -lt 2 ]; then
        echo "用法: $0 <目标地址> <端口>"
        echo "示例: $0 example.com 443"
        exit 1
    fi
    TARGET=$1
    PORT=$2
    # 检查权限
    if [ "$EUID" -ne 0 ]; then 
        log_warn "建议使用root权限运行以获得最佳优化效果"
    fi
    # 执行优化
    optimize_system_params
    optimize_dns "$TARGET"
    test_connections "$TARGET" "$PORT"
    # 显示结果
    if [ $? -eq 0 ]; then
        echo ""
        log_info "连接优化完成！！"
    else
        echo ""
        log_error "连接优化失败"
        exit 1
    fi
}
# 执行主函数
main "$@"

配置文件示例

# timeout_config.yaml
# 连接超时配置
general:
  connect_timeout: 5
  read_timeout: 30
  write_timeout: 10
  max_retries: 3
http:
  timeout: [3, 10]  # (connect, read)
  pool_connections: 10
  pool_maxsize: 20
  max_retries: 3
  backoff_factor: 0.5
database:
  mysql:
    connect_timeout: 5
    read_timeout: 30
    pool_size: 10
    pool_recycle: 3600
  redis:
    socket_timeout: 5
    socket_connect_timeout: 3
    retry_on_timeout: true
tcp:
  keepalive_time: 120
  keepalive_intvl: 30
  keepalive_probes: 3
  tcp_nodelay: true
dns:
  cache_ttl: 300
  servers:
    - 8.8.8.8
    - 8.8.4.4
    - 114.114.114.114

使用说明

安装依赖

# Python依赖
pip install requests aiohttp pymysql
# 系统依赖
sudo apt-get install netcat dnsutils  # Ubuntu/Debian
sudo yum install netcat bind-utils     # CentOS/RHEL

运行脚本

# Python版本
python3 timeout_optimizer.py
# Shell版本
chmod +x timeout_optimizer.sh
sudo ./timeout_optimizer.sh example.com 443

这个脚本提供了全面的连接超时优化方案，包括重试机制、连接池管理、系统参数优化等功能,可以根据实际需求进行调整。

本文地址： https://www.bmjup.com/post/9591.html