Python案例怎么断开无效连接？

本文目录导读：

1. HTTP 请求（requests 库）
2. 数据库连接（MySQL/SQLite）
3. WebSocket 连接
4. Socket 底层连接
5. 使用 try-finally 确保断开
6. 通用断开连接的最佳实践
7. 完整案例：连接池中的无效连接清除

在 Python 中“断开无效连接”通常指处理网络连接（如 HTTP、数据库、WebSocket）或文件描述符时的异常连接清理，以下是常见场景及代码案例：

HTTP 请求（requests 库）

import requests
from requests.exceptions import ConnectionError, Timeout
def safe_request(url, timeout=5):
    try:
        response = requests.get(url, timeout=timeout)
        # 检查状态码
        if response.status_code != 200:
            print(f"无效连接，状态码: {response.status_code}")
            response.close()  # 主动关闭
            return None
        return response
    except (ConnectionError, Timeout) as e:
        print(f"连接失败: {e}")
        return None

数据库连接（MySQL/SQLite）

import pymysql
def safe_db_connection():
    conn = None
    try:
        conn = pymysql.connect(host='localhost', user='root', password='', database='test')
        # 执行操作...
        return conn
    except pymysql.Error as e:
        if conn:
            conn.close()  # 断开无效连接
            print(f"数据库连接断开: {e}")
        return None

WebSocket 连接

import websocket
def safe_websocket():
    ws = None
    try:
        ws = websocket.create_connection("wss://echo.websocket.org", timeout=10)
        ws.send("Hello")
        response = ws.recv()
        return response
    except (websocket.WebSocketTimeoutException, websocket.WebSocketConnectionClosedException) as e:
        if ws:
            ws.close()  # 强制关闭
            print(f"WebSocket无效连接已断开: {e}")
        return None

Socket 底层连接

import socket
def check_socket_connection(host="8.8.8.8", port=53, timeout=3):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((host, port))
        # 发送心跳或测试数据
        sock.send(b'\x00')
        data = sock.recv(1024)
        if not data:  # 无数据返回视为无效
            raise ConnectionError("无数据响应")
        return True
    except (socket.timeout, ConnectionRefusedError, OSError) as e:
        print(f"检测到无效连接: {e}")
        return False
    finally:
        sock.close()  # 确保关闭

使用 try-finally 确保断开

import urllib.request
def safe_url_open(url):
    response = None
    try:
        response = urllib.request.urlopen(url, timeout=5)
        return response.read().decode()
    except urllib.error.URLError as e:
        print(f"URL无效: {e}")
        return ""
    finally:
        if response:
            response.close()  # 确保释放

通用断开连接的最佳实践

完整案例：连接池中的无效连接清除

import time
from threading import Lock
class ConnectionPool:
    def __init__(self):
        self.connections = []
        self.lock = Lock()
    def remove_invalid(self):
        """清理所有无效连接"""
        with self.lock:
            valid_connections = []
            for conn in self.connections:
                try:
                    # 发送心跳检测
                    conn.ping(timeout=1)
                    valid_connections.append(conn)
                except Exception:
                    try:
                        conn.close()  # 断开无效连接
                    except:
                        pass
                    print(f"已移除无效连接: {conn}")
            self.connections = valid_connections

核心原则：

• 使用 try-except 捕获连接异常
• 在 finally 或 except 中调用 .close()
• 设置超时避免无限等待
• 对于长连接池,定期执行心跳检测

根据具体场景选择合适的断开方式,大多数 Python 库都提供了 close() 方法用于显式断开连接。