Python Socket中常用的代码模板汇总(最新整理)

import socket

import socket

def tcp_server():
    # 创建socket对象 TCP
    server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # 解决服务端重启端口立即复用，避免 WinError 10048 端口占用
    server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    # 绑定地址 0.0.0.0 监听本机所有网卡
    addr = ("0.0.0.0", 8888)
    server_sock.bind(addr)
    # 开启监听，参数为等待连接队列大小
    server_sock.listen(5)
    print(f"TCP服务端启动成功，监听 {addr}")

    while True:
        # 阻塞等待客户端接入
        client_conn, client_addr = server_sock.accept()
        print(f"新客户端连接：{client_addr}")
        try:
            while True:
                # 最多接收1024字节
                data = client_conn.recv(1024)
                if not data:
                    # 空数据 = 客户端主动断开连接
                    break
                msg = data.decode("utf-8")
                print(f"收到[{client_addr}]消息：{msg}")
                # 回复客户端
                resp = f"服务端已收到：{msg}"
                client_conn.sendall(resp.encode("utf-8"))
        except Exception as e:
            print("连接异常", e)
        finally:
            client_conn.close()
            print(f"客户端 {client_addr} 已断开")

if __name__ == "__main__":
    tcp_server()

import socket

def tcp_client():
    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_addr = ("127.0.0.1", 8888)
    # 连接服务端
    client_sock.connect(server_addr)
    print("成功连接服务端")

    send_msg = "Hello Socket TCP"
    client_sock.sendall(send_msg.encode("utf-8"))
    # 接收响应
    recv_data = client_sock.recv(1024)
    print("服务端返回：", recv_data.decode("utf-8"))

    client_sock.close()

if __name__ == "__main__":
    tcp_client()

import socket

def tcp_client_timeout():
    client_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_sock.settimeout(5)  # 设置超时5秒
    try:
        client_sock.connect(("127.0.0.1", 8888))
        client_sock.sendall(b"test timeout")
        data = client_sock.recv(1024)
        print(data.decode())
    except socket.timeout:
        print("连接/接收超时！")
    except ConnectionRefusedError:
        print("无法连接，服务端未启动")
    finally:
        client_sock.close()

if __name__ == "__main__":
    tcp_client_timeout()

import socket

def udp_server():
    udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    udp_sock.bind(("0.0.0.0", 9999))
    print("UDP服务端启动，端口9999")

    while True:
        data, addr = udp_sock.recvfrom(1024)
        msg = data.decode("utf-8")
        print(f"来自{addr}: {msg}")
        # 回包
        reply = f"UDP收到消息：{msg}"
        udp_sock.sendto(reply.encode("utf-8"), addr)

if __name__ == "__main__":
    udp_server()

import socket

def udp_client():
    udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_addr = ("127.0.0.1", 9999)
    msg = "Hello UDP Socket"
    udp_sock.sendto(msg.encode("utf-8"), server_addr)
    # 等待接收返回
    data, addr = udp_sock.recvfrom(1024)
    print("服务端响应：", data.decode("utf-8"))
    udp_sock.close()

if __name__ == "__main__":
    udp_client()

def get_local_ip():
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        # 随便连接一个公网地址，不真实发包，用来获取本机网卡IP
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except Exception:
        return "127.0.0.1"

print(get_local_ip())

import socket

def is_port_used(host="127.0.0.1", port=8888):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(1)
    result = False
    try:
        sock.connect((host, port))
        result = True
    except ConnectionRefusedError:
        result = False
    finally:
        sock.close()
    return result

print(is_port_used(port=8888))

# 字符串转字节发送
raw_data = bytes.fromhex("01 02 03 0A FF")
# sock.sendall(raw_data)

# 收到字节打印十六进制
# data = sock.recv(1024)
# print(data.hex())