6. Socket 多线程

概念

如果服务器在 accept 之后当场读完整个请求，第二个客户就要等第一个客户结束。并发服务器把 accept 和业务处理分开：一个线程专门接受连接，线程池里的工人各自处理自己的 Socket。池的大小要有上限。每个连接都 new Thread，连接一多就会把进程打满。工人之间如果要修改连接计数、在线用户表，不能各写一个普通变量，要用后面的线程安全类，或者加锁。

主循环只做 accept，然后把 Socket 提交给线程池。
每个工人处理完必须关闭自己的连接，包括发生异常时。
线程数固定，多出来的连接在队列里等待，队列也要有上限。
共享计数使用原子变量。不要让多个工人同时执行 count++。

算法流程

下面的 UML 图说明 Java 例程和 Python 例程共同实现的流程。实线箭头表示调用或发送，虚线箭头表示返回或唤起。

接入与工人分工（UML 时序图）

sequenceDiagram participant Accept as 监听线程 participant Pool as 固定大小线程池 participant Worker as 工人线程 participant Client as 客户端 Accept->>Client: accept Accept->>Pool: 把这条连接提交给线程池 Note over Accept: 立刻回去接受下一个连接 Pool->>Worker: 分配空闲工人 Worker->>Worker: 活跃连接数加一 Worker->>Client: 读取并回写 Worker->>Worker: 活跃连接数减一 Worker->>Client: 关闭这条连接

Java 例程

import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

public class PoolServer {
    static final AtomicInteger ACTIVE = new AtomicInteger(); // 多个工人会同时改它，不能用普通 int

    static void handle(Socket socket) {
        ACTIVE.incrementAndGet();                 // 这个连接开始处理
        try {
            byte[] buf = new byte[256];
            int n = socket.getInputStream().read(buf); // 教学示例只读一块；真实协议要按边界读全
            if (n > 0) socket.getOutputStream().write(buf, 0, n); // 原样回写，n 才是实际读到的长度
        } catch (Exception ignored) {
            // 单个客户异常不能让整个服务器退出
        } finally {
            ACTIVE.decrementAndGet();
            try { socket.close(); } catch (Exception ignored) {} // 无论成功失败都关闭这条连接
        }
    }

    public static void main(String[] args) throws Exception {
        ExecutorService pool = Executors.newFixedThreadPool(8); // 同时最多 8 个连接在处理
        ServerSocket server = new ServerSocket(9000);
        while (true) {
            final Socket client = server.accept(); // 主线程只负责接入，不在这里读数据
            pool.submit(new Runnable() {           // 交给池里的空闲工人
                public void run() { handle(client); }
            });
        }
    }
}

Python 例程

import socket
from concurrent.futures import ThreadPoolExecutor
import threading

active = 0
lock = threading.Lock()               # 保护 active。count += 1 不是原子操作

def handle(conn):
    global active
    with lock:
        active += 1                   # 读改写必须放在同一把锁里
    try:
        data = conn.recv(256)         # 每个工人只处理自己 accept 到的那条连接
        if data:
            conn.sendall(data)        # 回显。空数据表示对端已经关闭
    finally:
        with lock:
            active -= 1               # 和加一使用同一把锁
        conn.close()                  # 异常时也要归还端口和文件描述符

def serve():
    server = socket.socket()
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("127.0.0.1", 9000))
    server.listen(16)                 # 工人都忙时，已握手的连接先排在这里
    pool = ThreadPoolExecutor(max_workers=8)
    while True:
        conn, _ = server.accept()     # 接入线程不读业务数据，避免一个慢客户挡住后面的人
        pool.submit(handle, conn)     # 把这条连接交给池中线程