一个简单的并发服务器

Posted 2021-01-08 昀溪

最简单的Socket的并发服务器

概述

那么最简单的服务端并发处理客户端请求就是，父进程用监听套接字监听，当有连接过来时那么监听套接字就变成了已连接套接字（源和目的的IP和端口都包含了），这时候就可以和客户端通信，但此时其他客户端无法连接进来，因为这个套接字被占用，所以就会产生一个子进程来处理和客户端的通信，也就是这个连接套接字由子进程处理，而父进程继续用监听套接字的形式来等待下一个连接请求。

代码段

服务器程序

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 # Author: rex.cheny
 4 # E-mail: [e-mail]
 5 
 6 import socket
 7 import os, time, sys
 8 
 9 
10 def echoStr(connFd):
11     print("新连接：", connFd.getpeername())
12     while True:
13         bytesData = connFd.recv(1024)
14         data = bytesData.decode(encoding="utf-8")
15         print("收到客户端消息：", data)
16         if data == "Bye":
17             return
18         else:
19             time.sleep(1)
20             connFd.send(data.encode(encoding="utf-8"))
21 
22 
23 def main():
24     sockFd = socket.socket()
25     sockFd.bind(("", 5555))
26     sockFd.listen(5)
27 
28     print("等待客户端连接......")
29     while True:
30         connFd, remAddr = sockFd.accept()
31 
32         try:
33             pid = os.fork()
34             if pid == 0:
35                 # 说明当前运行在子进程中
36                 sockFd.close()  # 关闭监听套接字
37                 echoStr(connFd)  # 执行回显函数
38                 # connFd.close()  # 关闭连接套接字，这里是否要显示的关闭和客户端的连接套接字与个人编程风格有关，因为客户端发送完数据后就主动调用了close()
39                 exit(0)  # 子进程退出
40             else:
41                 """
42                 关闭连接套接字，这时候并不会关闭服务器与客户端的TCP连接，因为connFd这个套接字的会被子进程所使用，所以该套接字的引用
43                 计数器为1，只有为0时才会被关闭。
44                 """
45                 connFd.close()
46         except Exception as err:
47             print(err)
48 
49 
50 if __name__ == \'__main__\':
51     main()

客户端程序

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 # Author: rex.cheny
 4 # E-mail: [e-mail]
 5 
 6 import socket
 7 
 8 
 9 def echoStr(sockFd, data):
10     sockFd.send(data)
11     bytesData = sockFd.recv(1024)
12     data = bytesData.decode(encoding="utf-8")
13     print(data)
14 
15 
16 def main():
17     sockFd = socket.socket()
18     sockFd.connect(("127.0.0.1", 5555))
19 
20     for i in range(1, 11):
21         data = "第：" + str(i) + " 条消息。"
22         echoStr(sockFd, data.encode(encoding="utf-8"))
23 
24     echoStr(sockFd, "Bye".encode(encoding="utf-8"))
25     sockFd.close()
26 
27 
28 if __name__ == \'__main__\':
29     main()

结果演示

 

这时候就实现了多个客户端并发连接服务器端。这里只是演示了一种最简单也是最原始的一种方式，因为fork一个进程系统开销很大所以虽然是并发但是不适用大规模并发的情况下。无论是多进程或者是进程池或者是多线程并发其实效率都不高当然这是相对来讲，你想一想1000个请求难道要启动1000个线程或者进程吗？显然不现实。在面对大量并发请求的时候就要用到多路复用或者是异步，这个后面章节会讲。不过需要先明白多路复用和并发处理其实是两个概念，多路复用主要解决不阻塞问题而并发是同时处理问题，我上面这种FORK的形式虽然是并发的，但是单一进程内其实还是阻塞的，无论是对服务器进程还是客户端进程它们内部都是阻塞的。

我们这次通过后台运行查看一下网络和进程状态。

# 这样来运行服务器程序
python3 ./server.py >> ./log.txt &

当有客户端连接进来是这样的，子进程PID是65686而它的PID也就是父进程ID是65623；网络监控显示有一个TCP连接成功建立

当客户端执行完毕之后是这样的，服务器显示的连接套接字时TIME_WAIT状态，而之前fork的子进程变成了僵尸进程。TIME_WAIT过一会儿就消失掉，但是这个僵尸进程会一直存在，直到父进程退出。

僵尸进程

上面最后一个图中的 “Z” 就表示僵尸进程。

接下来我们改进一下服务器程序来解决一下僵尸进程问题，客户端不变。

 1 #!/usr/bin/env python
 2 # -*- coding: utf-8 -*-
 3 # Author: rex.cheny
 4 # E-mail: [e-mail]
 5 
 6 import socket
 7 import os, time, sys
 8 import signal
 9 
10 
11 def echoStr(connFd):
12     print("新连接：", connFd.getpeername())
13     while True:
14         bytesData = connFd.recv(1024)
15         data = bytesData.decode(encoding="utf-8")
16         print("收到客户端消息：", data)
17         if data == "Bye":
18             return
19         else:
20             time.sleep(1)
21             connFd.send(data.encode(encoding="utf-8"))
22 
23 
24 def sigChld(signum, frame):
25     """
26     僵尸进程处理函数这两个参数是必须的
27     :param signum: 发生的信号
28     :param frame: 发生信号的时候的函数调用栈
29     :return:
30     """
31 
32     """
33     waitpid() 用于清理僵尸进程，返回值为已终止的子进程ID和进程终止状态
34     pid 进程号，如果是-1表示等待第一个终止的子进程
35     os.WNOHANG 参数，默认为0也就是os.WNOHANG，表示在内核没有通知有已终止的子进程时不阻塞
36     """
37     pid, status = os.waitpid(-1, os.WNOHANG)
38     while pid > 0:
39         print("child ", pid, " is terminated")
40         return
41 
42 
43 def main():
44     sockFd = socket.socket()
45     sockFd.bind(("", 5555))
46     sockFd.listen(5)
47 
48     """
49     调用signal函数，第一个参数是信号名，第二个是信号处理函数。这个signal函数就是对系统调用signal函数的简单封装。
50     这个函数必须在fork第一个子进程之前做且只能做一次。
51     """
52     signal.signal(signal.SIGCHLD, sigChld)
53     print("等待客户端连接......")
54     while True:
55         connFd, remAddr = sockFd.accept()
56 
57         try:
58             pid = os.fork()
59             if pid == 0:
60                 # 说明当前运行在子进程中
61                 sockFd.close()  # 关闭监听套接字
62                 echoStr(connFd)  # 执行回显函数
63                 # connFd.close()  # 关闭连接套接字，这里是否要显示的关闭和客户端的连接套接字与个人编程风格有关，因为客户端发送完数据后就主动调用了close()
64                 exit(0)  # 子进程退出
65             else:
66                 """
67                 关闭连接套接字，这时候并不会关闭服务器与客户端的TCP连接，因为connFd这个套接字的会被子进程所使用，所以该套接字的引用
68                 计数器为1，只有为0时才会被关闭。
69                 """
70                 connFd.close()
71         except Exception as err:
72             print(err)
73 
74 
75 if __name__ == \'__main__\':
76     main()

SIGCHLD信号的含义是一个进程终止或者停止，将该信号发送给父进程，父进程的wait或者waitpid可以捕捉这个信号。

这时候可以看到通信完毕后已经不存在僵尸进程了。

僵尸进程会被init或者systemd进程接管，而他们会使用wait来清理这些僵尸进程，所以当我们使用fork子进程的时候都要wait他们防止他们变成僵尸进程。wait和waitpid都可以清理僵尸进程，但是略有不同：

• 调用wait的时候如果没有已经终止的子进程那么调用wait的进程将会被阻塞在这里直到出现一个被终止的子进程
• 调用waitpid我们可以通过参数指定pid号也可以指定如果没有已经终止的子进程是否要阻塞，这样体验就会好很多。

如果你这里调用wait将会发生什么呢？已上面的程序为例你同时启动5个客户端，在服务器就会fork5个子进程处理，当通信完毕需要清理僵尸进程的时候有很大可能会清理小于5个。因为5个中断信号几乎同时到达，第一个到达的时候阻塞在wait，而后面相继到达，Unix的信号是不排队的，也就是被阻塞期间产生一次或多次最终只提交一次。

Python自带的一个叫做socketserver模块里面有一个ForkingMixIn的其核心实现就是fork加信号处理。

需要知道的事情

服务器进程崩溃会怎么样

与服务器之间的网络中断会怎么样