当前位置：首页 > news >正文

网络编程问题：TCP/UDP 连接异常解决方案

news 2026/5/11 22:42:15

TCP/UDP 连接异常解决方案代码示例

以下是一个基于 Python 的 TCP/UDP 连接异常处理代码示例，涵盖常见的连接异常场景（如超时、连接拒绝、端口占用等），并提供重试机制和日志记录功能。

TCP 连接异常处理

import socket import time import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def tcp_connect_with_retry(host, port, max_retries=3, timeout=5): retry_count = 0 while retry_count < max_retries: try: sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.settimeout(timeout) sock.connect((host, port)) logger.info(f"TCP connection established to {host}:{port}") return sock except socket.timeout: logger.warning(f"TCP connection timeout (attempt {retry_count + 1})") except ConnectionRefusedError: logger.error(f"Connection refused by {host}:{port}") except socket.error as e: logger.error(f"Socket error: {str(e)}") finally: retry_count += 1 time.sleep(1) raise ConnectionError(f"Failed to establish TCP connection after {max_retries} retries")

https://www.zhihu.com/zvideo/1994260861682997179/
https://www.zhihu.com/zvideo/1994260861682997179
https://www.zhihu.com/zvideo/1994260860697341967/
https://www.zhihu.com/zvideo/1994260860697341967
https://www.zhihu.com/zvideo/1994260857056671243/
https://www.zhihu.com/zvideo/1994260857056671243
https://www.zhihu.com/zvideo/1994260851021071994/
https://www.zhihu.com/zvideo/1994260851021071994
https://www.zhihu.com/zvideo/1994260850433886062/
https://www.zhihu.com/zvideo/1994260850433886062
https://www.zhihu.com/zvideo/1994260847934066912/
https://www.zhihu.com/zvideo/1994260847934066912
https://www.zhihu.com/zvideo/1994260843584570191/
https://www.zhihu.com/zvideo/1994260843584570191
https://www.zhihu.com/zvideo/1994260839423832246/
https://www.zhihu.com/zvideo/1994260839423832246
https://www.zhihu.com/zvideo/1994260838081664990/
https://www.zhihu.com/zvideo/1994260838081664990
https://www.zhihu.com/zvideo/1994260835078529378/
https://www.zhihu.com/zvideo/1994260835078529378
https://www.zhihu.com/zvideo/1994260833497273482/
https://www.zhihu.com/zvideo/1994260833497273482
https://www.zhihu.com/zvideo/1994260832259966840/
https://www.zhihu.com/zvideo/1994260832259966840
https://www.zhihu.com/zvideo/1994260832532583874/
https://www.zhihu.com/zvideo/1994260832532583874
https://www.zhihu.com/zvideo/1994260827105146355/
https://www.zhihu.com/zvideo/1994260827105146355
https://www.zhihu.com/zvideo/1994260823569364542/
https://www.zhihu.com/zvideo/1994260823569364542
https://www.zhihu.com/zvideo/1994260822684356958/
https://www.zhihu.com/zvideo/1994260822684356958
https://www.zhihu.com/zvideo/1994260818049655700/
https://www.zhihu.com/zvideo/1994260818049655700
https://www.zhihu.com/zvideo/1994260817747674698/
https://www.zhihu.com/zvideo/1994260817747674698
https://www.zhihu.com/zvideo/1994260815390463543/
https://www.zhihu.com/zvideo/1994260815390463543
https://www.zhihu.com/zvideo/1994260811632358143/
https://www.zhihu.com/zvideo/1994260811632358143

UDP 连接异常处理

def udp_send_with_retry(host, port, data, max_retries=3, timeout=5): retry_count = 0 while retry_count < max_retries: try: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.settimeout(timeout) sock.sendto(data.encode(), (host, port)) logger.info(f"UDP data sent to {host}:{port}") return sock except socket.timeout: logger.warning(f"UDP send timeout (attempt {retry_count + 1})") except socket.error as e: logger.error(f"UDP socket error: {str(e)}") finally: retry_count += 1 time.sleep(1) raise ConnectionError(f"Failed to send UDP data after {max_retries} retries")

https://www.zhihu.com/zvideo/1994266399967815160/
https://www.zhihu.com/zvideo/1994266399967815160
https://www.zhihu.com/zvideo/1994266394804638799/
https://www.zhihu.com/zvideo/1994266394804638799
https://www.zhihu.com/zvideo/1994266393525372736/
https://www.zhihu.com/zvideo/1994266393525372736
https://www.zhihu.com/zvideo/1994266392124466186/
https://www.zhihu.com/zvideo/1994266392124466186
https://www.zhihu.com/zvideo/1994266391612769736/
https://www.zhihu.com/zvideo/1994266391612769736
https://www.zhihu.com/zvideo/1994266391889604666/
https://www.zhihu.com/zvideo/1994266391889604666
https://www.zhihu.com/zvideo/1994266389448504132/
https://www.zhihu.com/zvideo/1994266389448504132
https://www.zhihu.com/zvideo/1994266387447832880/
https://www.zhihu.com/zvideo/1994266387447832880
https://www.zhihu.com/zvideo/1994266387040981461/
https://www.zhihu.com/zvideo/1994266387040981461
https://www.zhihu.com/zvideo/1994266386592199141/
https://www.zhihu.com/zvideo/1994266386592199141
https://www.zhihu.com/zvideo/1994266373937967153/
https://www.zhihu.com/zvideo/1994266373937967153
https://www.zhihu.com/zvideo/1994266375666034264/
https://www.zhihu.com/zvideo/1994266375666034264
https://www.zhihu.com/zvideo/1994266370750301136/
https://www.zhihu.com/zvideo/1994266370750301136
https://www.zhihu.com/zvideo/1994266368053359641/
https://www.zhihu.com/zvideo/1994266368053359641
https://www.zhihu.com/zvideo/1994266368544096292/
https://www.zhihu.com/zvideo/1994266368544096292
https://www.zhihu.com/zvideo/1994266367365490383/
https://www.zhihu.com/zvideo/1994266367365490383
https://www.zhihu.com/zvideo/1994266365352248655/
https://www.zhihu.com/zvideo/1994266365352248655
https://www.zhihu.com/zvideo/1994266364907635656/
https://www.zhihu.com/zvideo/1994266364907635656
https://www.zhihu.com/zvideo/1994266364546942673/
https://www.zhihu.com/zvideo/1994266364546942673
https://www.zhihu.com/zvideo/1994266364588873142/
https://www.zhihu.com/zvideo/1994266364588873142

端口占用检测

def is_port_in_use(port, protocol='tcp'): try: if protocol.lower() == 'tcp': sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) else: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) sock.bind(('0.0.0.0', port)) sock.close() return False except socket.error: return True

使用示例

if __name__ == "__main__": # TCP 示例 try: tcp_sock = tcp_connect_with_retry("example.com", 80) tcp_sock.sendall(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n") response = tcp_sock.recv(4096) print(response.decode()) except Exception as e: print(f"TCP error: {str(e)}") # UDP 示例 try: udp_sock = udp_send_with_retry("example.com", 53, "test data") data, addr = udp_sock.recvfrom(1024) print(f"Received UDP data from {addr}: {data.decode()}") except Exception as e: print(f"UDP error: {str(e)}") # 端口检测 print(f"Port 80 in use (TCP): {is_port_in_use(80)}") print(f"Port 53 in use (UDP): {is_port_in_use(53, 'udp')}")