面试题：Python SocketServer TCP客户端的性能调优与安全加固

性能调优

网络配置
- 调整TCP参数：
  - TCP_NODELAY：禁用Nagle算法，减少小包延迟。在Python中使用setsockopt方法设置，例如：

import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

 - **`SO_REUSEADDR`**：允许重用本地地址，使得程序重启时可以快速绑定到相同端口。

sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

优化缓冲区大小：
- 发送缓冲区：通过setsockopt设置SO_SNDBUF，例如：

sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 65536) # 设置发送缓冲区为64KB

 - **接收缓冲区**：设置`SO_RCVBUF`，例如：

sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 65536) # 设置接收缓冲区为64KB

资源管理
- 多线程/多进程：
  - 多线程：使用threading模块创建多个线程处理不同连接。例如：

import threading
import socket

def handle_connection(sock):
    # 处理连接的逻辑
    pass

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 12345))
server_socket.listen(10)

while True:
    client_sock, addr = server_socket.accept()
    t = threading.Thread(target=handle_connection, args=(client_sock,))
    t.start()

 - **多进程**：使用`multiprocessing`模块创建多个进程处理连接，注意进程间资源管理与通信。

连接池：维护一个连接池，避免频繁创建和销毁连接。可以使用queue模块实现简单连接池，例如：

import socket
import queue

class ConnectionPool:
    def __init__(self, host, port, pool_size):
        self.host = host
        self.port = port
        self.pool = queue.Queue(maxsize = pool_size)
        for _ in range(pool_size):
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.connect((host, port))
            self.pool.put(sock)

    def get_connection(self):
        return self.pool.get()

    def return_connection(self, sock):
        self.pool.put(sock)

高效数据处理
- 使用非阻塞I/O：通过setblocking(0)将套接字设置为非阻塞模式，结合select或epoll（在Linux上）实现高效I/O复用。例如使用select：

import socket
import select

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 12345))
server_socket.listen(10)
server_socket.setblocking(0)

inputs = [server_socket]
while True:
    readable, _, _ = select.select(inputs, [], [])
    for sock in readable:
        if sock is server_socket:
            client_sock, addr = server_socket.accept()
            client_sock.setblocking(0)
            inputs.append(client_sock)
        else:
            data = sock.recv(1024)
            if data:
                # 处理数据
                pass
            else:
                sock.close()
                inputs.remove(sock)

安全加固

加密算法选择
- TLS/SSL：使用ssl模块实现TLS/SSL加密。例如：

import socket
import ssl

context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain(certfile='server.crt', keyfile='server.key')

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(('0.0.0.0', 12345))
server_socket.listen(10)

while True:
    client_sock, addr = server_socket.accept()
    ssl_sock = context.wrap_socket(client_sock, server_side = True)
    try:
        data = ssl_sock.recv(1024)
        # 处理数据
        ssl_sock.send(b'Response')
    finally:
        ssl_sock.close()

认证机制
- 用户名/密码认证：在连接建立后，要求客户端发送用户名和密码进行认证。例如：

def authenticate(sock):
    sock.send(b'Username: ')
    username = sock.recv(1024).decode('utf - 8').strip()
    sock.send(b'Password: ')
    password = sock.recv(1024).decode('utf - 8').strip()
    # 验证用户名和密码逻辑
    if username == 'valid_user' and password == 'valid_pass':
        return True
    return False

防止攻击
- 防止DoS攻击：设置连接超时，限制同一IP的连接数。例如：

server_socket.settimeout(60) # 设置连接超时60秒

ip_connection_count = {}
def handle_connection(sock, addr):
    ip = addr[0]
    if ip not in ip_connection_count:
        ip_connection_count[ip] = 1
    else:
        ip_connection_count[ip] += 1
    if ip_connection_count[ip] > 10: # 限制同一IP连接数为10
        sock.close()
        return
    # 正常处理连接逻辑
    pass

星途面试题库

面试题：Python SocketServer TCP客户端的性能调优与安全加固

知识考点

面试题答案

性能调优

安全加固