这个案例能让你学会Python中Socket的setsockopt选项设置吗

访客网络编程 2026-06-04 19:32:47 1

本文目录导读：

案例：构建一个可复用地址的TCP回显服务器
关键知识点详解
进阶使用场景
实验验证

这是一个很好的学习切入点！我通过一个完整的实战案例，带你掌握 Python Socket 中 setsockopt 的核心用法。

案例：构建一个可复用地址的TCP回显服务器

问题场景

当服务器程序崩溃后立即重启时,经常遇到 Address already in use 错误，这是因为操作系统还在TIME_WAIT状态，我们用 setsockopt 解决这个问题。

完整代码

import socket
import sys
import time
class RobustEchoServer:
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
        self.server_socket = None
    def create_server_socket(self):
        """创建并配置服务器套接字"""
        try:
            # 1. 创建 TCP socket
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            # 2. 关键配置：设置 socket 选项
            self._configure_socket_options()
            # 3. 绑定地址
            self.server_socket.bind((self.host, self.port))
            # 4. 开始监听
            self.server_socket.listen(5)
            print(f"[+] 服务器启动在 {self.host}:{self.port}")
        except socket.error as e:
            print(f"[-] 创建socket失败: {e}")
            sys.exit(1)
    def _configure_socket_options(self):
        """配置各种socket选项"""
        options = [
            # 选项1: 允许地址重用（解决Address already in use）
            (socket.SOL_SOCKET, socket.SO_REUSEADDR, 1),
            # 选项2: 设置接收缓冲区大小（32KB）
            (socket.SOL_SOCKET, socket.SO_RCVBUF, 32768),
            # 选项3: 设置发送缓冲区大小（32KB）
            (socket.SOL_SOCKET, socket.SO_SNDBUF, 32768),
            # 选项4: 设置TCP keep-alive
            (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
            # 选项5: 设置TCP no-delay（禁用Nagle算法，减少延迟）
            (socket.IPPROTO_TCP, socket.TCP_NODELAY, 1),
        ]
        for level, optname, value in options:
            try:
                self.server_socket.setsockopt(level, optname, value)
                print(f"  [+] 设置选项成功: level={level}, optname={optname}, value={value}")
            except socket.error as e:
                print(f"  [-] 设置选项失败: {e}")
    def get_socket_info(self):
        """显示当前socket的配置信息"""
        info = {
            'SO_REUSEADDR': self.server_socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR),
            'SO_KEEPALIVE': self.server_socket.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE),
            'SO_RCVBUF': self.server_socket.getsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF),
            'SO_SNDBUF': self.server_socket.getsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF),
            'TCP_NODELAY': self.server_socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY),
        }
        return info
    def start(self):
        """启动服务器"""
        self.create_server_socket()
        # 显示socket配置信息
        print("\n[+] Socket配置信息:")
        for key, value in self.get_socket_info().items():
            print(f"  {key}: {value}")
        print("\n[+] 等待客户端连接...")
        while True:
            try:
                # 接受客户端连接
                client_socket, client_address = self.server_socket.accept()
                print(f"\n[+] 新连接: {client_address}")
                # 处理客户端请求
                self.handle_client(client_socket, client_address)
            except KeyboardInterrupt:
                print("\n[-] 服务器关闭...")
                break
            except socket.error as e:
                print(f"[-] 连接错误: {e}")
        self.server_socket.close()
    def handle_client(self, client_socket, client_address):
        """处理客户端请求（回显功能）"""
        with client_socket:
            while True:
                try:
                    data = client_socket.recv(1024)
                    if not data:
                        break
                    # 回显数据
                    client_socket.sendall(data)
                    print(f"  [*] 收到并回显: {data.decode('utf-8', errors='ignore')}")
                except socket.error as e:
                    print(f"  [-] 客户端异常: {e}")
                    break
        print(f"  [-] 连接关闭: {client_address}")
class EchoClient:
    """简单的客户端用于测试"""
    def __init__(self, host='127.0.0.1', port=8888):
        self.host = host
        self.port = port
    def test(self, messages):
        """发送消息并接收回显"""
        client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            client_socket.connect((self.host, self.port))
            print(f"[+] 连接到服务器 {self.host}:{self.port}")
            for msg in messages:
                # 发送消息
                client_socket.sendall(msg.encode('utf-8'))
                print(f"[>] 发送: {msg}")
                # 接收回显
                response = client_socket.recv(1024)
                print(f"[<] 回显: {response.decode('utf-8')}")
                time.sleep(0.5)
        finally:
            client_socket.close()
            print("[-] 客户端关闭")
# 运行服务器
if __name__ == "__main__":
    # 创建并启动服务器
    server = RobustEchoServer()
    try:
        server.start()
    except KeyboardInterrupt:
        print("\n服务器已停止")

测试代码（单独运行客户端）

# 在另一个终端运行
if __name__ == "__client__":
    client = EchoClient()
    test_messages = ["Hello", "World", "Socket", "setsockopt"]
    client.test(test_messages)

关键知识点详解

setsockopt 的三个参数

socket.setsockopt(level, optname, value)

参数	说明	示例值
`level`	选项级别	`SOL_SOCKET`(通用), `IPPROTO_TCP`(TCP专用)
`optname`	选项名称	`SO_REUSEADDR`, `SO_KEEPALIVE`
`value`	选项值	整型(0/1), 字节串, 结构体

本案例使用的选项

选项	作用	使用场景
`SO_REUSEADDR`	允许重用本地地址	服务器崩溃后快速重启
`SO_KEEPALIVE`	TCP心跳检测	检测死连接
`SO_RCVBUF`	接收缓冲区大小	优化大文件传输
`SO_SNDBUF`	发送缓冲区大小	优化大文件传输
`TCP_NODELAY`	禁用Nagle算法	实时性要求高的应用

常见选项查询

# 获取已设置的选项值
reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
keepalive = sock.getsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE)

进阶使用场景

设置组播选项

# 加入多播组
import struct
group = socket.inet_aton("224.0.0.1")
mreq = struct.pack("4sL", group, socket.INADDR_ANY)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)

设置超时选项

# 设置socket超时
sock.settimeout(5.0)  # 5秒超时
# 等价于
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, struct.pack('LL', 5, 0))

实验验证

启动服务器：运行上面的服务器代码
测试地址重用：按 Ctrl+C 中断服务器，立即重新启动
用客户端测试：运行客户端发送几条消息
验证配置：查看服务器打印的配置信息

通过这个案例,你应该可以掌握 setsockopt 的：

✅ 基本使用方法
✅ 常见选项设置
✅ 参数含义
✅ 实际应用场景
✅ 查看已设置选项

还有什么关于 setsockopt 的问题吗？比如你想了解某个特定选项的详细用法？

标签： SO_KEEPALIVE

本文地址： https://dfhcn.com/post/44.html

文章来源：访客