Python 网络编程基础

2025-11-04 约 2530 字预计阅读 6 分钟

学习目标：掌握 Python 网络编程的核心概念，理解 TCP/UDP 协议差异，学会使用 socket 模块创建客户端和服务器。

1. 网络编程概述

网络编程是指编写能够在网络中不同计算机之间进行通信的程序。Python 提供了强大的 socket 模块，让我们能够轻松实现网络通信。

1.1 什么是 Socket

Socket（套接字）是网络通信的端点，可以看作是不同主机上的进程之间进行双向通信的端点。它封装了 IP 地址和端口号，为应用程序提供了发送和接收数据的机制。

        
import socket

# 查看 socket 模块版本
print(socket.__version__ if hasattr(socket, '__version__') else "内置模块")

1.2 TCP vs UDP

特性	TCP	UDP
连接方式	面向连接	无连接
可靠性	可靠传输	不可靠传输
数据顺序	保证顺序	不保证顺序
开销	较大	较小
适用场景	文件传输、网页浏览	视频流、游戏、DNS

2. TCP 客户端/服务器

2.1 TCP 服务器

        
        
        
    
import socket
import threading

def handle_client(conn, addr):
    """处理客户端连接"""
    print(f"新连接: {addr}")
    
    try:
        while True:
            # 接收数据
            data = conn.recv(1024)
            if not data:
                break
            
            message = data.decode('utf-8')
            print(f"收到来自 {addr}: {message}")
            
            # 发送响应
            response = f"服务器已收到: {message}".encode('utf-8')
            conn.send(response)
    except Exception as e:
        print(f"处理客户端 {addr} 时出错: {e}")
    finally:
        conn.close()
        print(f"连接关闭: {addr}")

def start_server(host='127.0.0.1', port=8888):
    """启动 TCP 服务器"""
    # 创建 socket 对象
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    # 绑定地址和端口
    server.bind((host, port))
    
    # 开始监听
    server.listen(5)
    print(f"服务器启动，监听 {host}:{port}")
    
    try:
        while True:
            # 接受连接
            conn, addr = server.accept()
            
            # 创建线程处理客户端
            thread = threading.Thread(
                target=handle_client, 
                args=(conn, addr)
            )
            thread.start()
            
            print(f"活动连接数: {threading.active_count() - 1}")
    except KeyboardInterrupt:
        print("\n服务器正在关闭...")
    finally:
        server.close()

# 运行服务器（在独立线程中）
# start_server()

2.2 TCP 客户端

        
        
        
    
import socket

def tcp_client(host='127.0.0.1', port=8888):
    """TCP 客户端"""
    # 创建 socket 对象
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # 连接服务器
        client.connect((host, port))
        print(f"已连接到服务器 {host}:{port}")
        
        # 发送数据
        message = "你好，服务器！"
        client.send(message.encode('utf-8'))
        print(f"发送: {message}")
        
        # 接收响应
        response = client.recv(1024).decode('utf-8')
        print(f"收到: {response}")
        
    except Exception as e:
        print(f"客户端错误: {e}")
    finally:
        client.close()
        print("连接已关闭")

# 运行客户端
# tcp_client()

2.3 使用上下文管理器

        
        
        
    
from contextlib import contextmanager

@contextmanager
def tcp_connection(host, port):
    """TCP 连接上下文管理器"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.connect((host, port))
        yield sock
    finally:
        sock.close()

# 使用示例
# with tcp_connection('127.0.0.1', 8888) as conn:
#     conn.send(b'Hello')
#     data = conn.recv(1024)

3. UDP 编程

3.1 UDP 服务器

        
        
        
    
import socket

def udp_server(host='127.0.0.1', port=9999):
    """UDP 服务器"""
    # 创建 UDP socket
    server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    # 绑定地址
    server.bind((host, port))
    print(f"UDP 服务器启动，监听 {host}:{port}")
    
    try:
        while True:
            # 接收数据和客户端地址
            data, addr = server.recvfrom(1024)
            message = data.decode('utf-8')
            print(f"收到来自 {addr}: {message}")
            
            # 发送响应
            response = f"UDP 回复: {message}".encode('utf-8')
            server.sendto(response, addr)
    except KeyboardInterrupt:
        print("\n服务器关闭")
    finally:
        server.close()

# udp_server()

3.2 UDP 客户端

        
        
        
    
import socket

def udp_client(host='127.0.0.1', port=9999):
    """UDP 客户端"""
    client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    
    try:
        # 发送数据（不需要连接）
        message = "Hello UDP!"
        client.sendto(message.encode('utf-8'), (host, port))
        print(f"发送: {message}")
        
        # 接收响应
        data, addr = client.recvfrom(1024)
        print(f"收到来自 {addr}: {data.decode('utf-8')}")
        
    except Exception as e:
        print(f"错误: {e}")
    finally:
        client.close()

# udp_client()

4. 实用网络工具

4.1 获取主机信息

        
        
        
    
import socket

# 获取本机主机名
hostname = socket.gethostname()
print(f"主机名: {hostname}")

# 获取本机 IP
ip = socket.gethostbyname(hostname)
print(f"IP 地址: {ip}")

# 通过域名获取 IP
try:
    baidu_ip = socket.gethostbyname('www.baidu.com')
    print(f"百度 IP: {baidu_ip}")
except socket.gaierror:
    print("无法解析域名")

# 获取地址信息
addr_info = socket.getaddrinfo('www.baidu.com', None)
for info in addr_info[:2]:
    print(f"地址信息: {info}")

4.2 简单的端口扫描器

        
        
        
    
import socket
from concurrent.futures import ThreadPoolExecutor

def scan_port(host, port):
    """扫描单个端口"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(1)
    
    try:
        result = sock.connect_ex((host, port))
        if result == 0:
            return port, True
        return port, False
    except Exception:
        return port, False
    finally:
        sock.close()

def scan_ports(host, start_port=1, end_port=100, max_workers=50):
    """扫描端口范围"""
    print(f"扫描 {host} 的端口 {start_port}-{end_port}...")
    
    open_ports = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(scan_port, host, port)
            for port in range(start_port, end_port + 1)
        ]
        
        for future in futures:
            port, is_open = future.result()
            if is_open:
                print(f"端口 {port} 开放")
                open_ports.append(port)
    
    print(f"扫描完成，发现 {len(open_ports)} 个开放端口")
    return open_ports

# 扫描本地常见端口（仅用于学习目的）
# scan_ports('127.0.0.1', 20, 100)

5. 网络协议解析

5.1 简单的 HTTP 请求

        
        
        
    
import socket

def simple_http_get(host, path='/', port=80):
    """使用 socket 发送 HTTP GET 请求"""
    # 创建 socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)
    
    try:
        # 连接服务器
        sock.connect((host, port))
        
        # 构造 HTTP 请求
        request = f"GET {path} HTTP/1.1\r\n"
        request += f"Host: {host}\r\n"
        request += "Connection: close\r\n"
        request += "\r\n"
        
        # 发送请求
        sock.send(request.encode('utf-8'))
        
        # 接收响应
        response = b''
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
        
        # 解析响应
        header, _, body = response.partition(b'\r\n\r\n')
        print("响应头:")
        print(header.decode('utf-8', errors='ignore'))
        print(f"\n响应体长度: {len(body)} 字节")
        
        return response.decode('utf-8', errors='ignore')
        
    except Exception as e:
        print(f"请求失败: {e}")
    finally:
        sock.close()

# 示例：获取示例网站（仅用于学习）
# response = simple_http_get('example.com')

5.2 URL 解析

        
        
        
    
from urllib.parse import urlparse, urljoin, parse_qs

# 解析 URL
url = "https://www.example.com:8080/path/to/page?name=john&age=30#section"
parsed = urlparse(url)

print(f"协议: {parsed.scheme}")
print(f"域名: {parsed.netloc}")
print(f"主机: {parsed.hostname}")
print(f"端口: {parsed.port}")
print(f"路径: {parsed.path}")
print(f"查询参数: {parsed.query}")
print(f"锚点: {parsed.fragment}")

# 解析查询参数
params = parse_qs(parsed.query)
print(f"参数字典: {params}")

# 拼接 URL
base = "https://example.com/api/"
full = urljoin(base, "users/123")
print(f"完整 URL: {full}")

6. 高级 Socket 选项

6.1 设置 Socket 选项

        
import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 地址复用（服务器快速重启）
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

# 发送/接收缓冲区大小
sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 8192)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8192)

# 获取选项值
bufsize = sock.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF)
print(f"发送缓冲区: {bufsize}")

# 设置超时
sock.settimeout(5.0)
print(f"超时时间: {sock.gettimeout()} 秒")

sock.close()

6.2 非阻塞 Socket

        
        
        
    
import socket
import select

def non_blocking_server(host='127.0.0.1', port=8888):
    """使用 select 的非阻塞服务器"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    server.setblocking(False)
    
    # 监控的 socket 列表
    inputs = [server]
    outputs = []
    
    print(f"非阻塞服务器启动 {host}:{port}")
    
    while inputs:
        # 使用 select 监控可读、可写、异常的 socket
        readable, writable, exceptional = select.select(
            inputs, outputs, inputs
        )
        
        for s in readable:
            if s is server:
                # 新连接
                conn, addr = s.accept()
                conn.setblocking(False)
                inputs.append(conn)
                print(f"新连接: {addr}")
            else:
                # 客户端数据
                data = s.recv(1024)
                if data:
                    print(f"收到: {data.decode('utf-8')}")
                    s.send(b"Received")
                else:
                    # 连接关闭
                    inputs.remove(s)
                    s.close()

# non_blocking_server()

7. 小结

概念	要点
Socket	网络通信端点，封装 IP + 端口
TCP	面向连接、可靠、适合数据传输
UDP	无连接、快速、适合实时应用
`socket` 模块	Python 网络编程基础模块
多线程/多进程	处理多个客户端连接

8. 练习题

修改 TCP 服务器，使其能够广播消息给所有连接的客户端。
使用 UDP 实现一个简单的多人聊天室。
编写一个函数，检测本机哪些端口正在监听。
实现一个简单的文件传输工具，使用 TCP 传输文件。

提示：网络编程时务必处理异常，注意资源释放，生产环境建议使用成熟的网络库如 asyncio、aiohttp 或 Web 框架。

目录