多线程实现socket连接池

发表于 | 分类于

利用线程池实现Socket连接池

先贴上代码，有空再补充

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
# 服务端
import queue
import socket
from threading import Thread
from concurrent.futures import ThreadPoolExecutor


def conn_handler(conn, addr):
    print(f"来自{addr}的连接")
    try:
        while 1:
            try:
                total_data = b""
                while 1:
                    data = conn.recv(1024)
                    total_data += data
                    if len(data) < 1024:
                        break
            except ConnectionResetError as e:
                print("对方已中断连接！")
                break

            print(f"{addr}：{total_data.decode()}")
            conn.send(b"Done")
    finally:
        conn.close()


def worker():
    while 1:
        conn, addr = conn_queue.get()
        conn_handler(conn, addr)
        conn_queue.task_done()


def start_conn_pool(max_num=10):
    with ThreadPoolExecutor(max_num) as executor:
        [executor.submit(worker, ) for _ in range(max_num)]


if __name__ == '__main__':
    conn_queue = queue.Queue(10)
    addr = ('127.0.0.1', 9000)
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind(addr)
    server.listen(1)

    pool = Thread(target=start_conn_pool)
    pool.start()

    while 1:
        conn, add = server.accept()
        conn_queue.put((conn, add))


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 客户端
import socket


def exchange(conn, send_data: str) -> bytes:
    if not isinstance(send_data, str):
        send_data = str(send_data)

    resp_data = b""
    while 1:
        conn.send(send_data.encode())
        recv = conn.recv(1024)
        resp_data += recv
        if len(recv) < 1024:
            break
    return resp_data.decode()


if __name__ == '__main__':
    addr = ('127.0.0.1', 9000)
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

    s.connect(addr)
    local_ip, local_port = s.getsockname()
    print(f"客户端{s.getsockname()}启动")

    while 1:
        send_con = input(f"{local_ip}:{local_port}：")
        if send_con == "close":
            break
        resp = exchange(s, send_con)
        print(f"{addr[0]}:{addr[1]}：{resp}")

    s.close()