python多任务 面向对象 命令行启动动态绑定端口号静态web服务器代码实现

一、静态web服务器-多任务

多任务web服务器：使用多线程，比进程更节省资源，支持多用户同时访问，可以同时处理多个客户端请求

实现步骤若客户端与服务端建立连接，则创建子线程，使用子线程处理客户端请求，防止主线程阻塞将创建的子线程设置为守护主线程，防止主线程无法退出

实现代码

import socketimport threadingdef handle_client_request(comm_socket): # 处理客户端请求recv_data = comm_socket.recv(4096) # 接收请求的二进制数据if len(recv_data) == 0:print('未获取到请求数据！')comm_socket.close()returnrecv_content = recv_data.decode('utf-8') # 对二进制数据解码print('获取到的数据内容为：', recv_content)request_list = recv_content.split(" ", maxsplit=2) # 根据指定字符串空格进行分割，最大分割次数为2request_path = request_list[1] # 获取请求资源路径print('请求路径为：', request_path)if request_path == "/": # 判断请求的是否是根目录，若是则返回首页指定数据request_path = "/index.html"try:with open('C:/Users/username/Desktop/ubuntu' + request_path, 'rb') as file: # 动态打开指定文件file_data = file.read() # 读取指定文件数据except Exception as e: # 请求异常，资源不存在，返回指定404错误数据with open('C:/Users/username/Desktop/ubuntu/error.html', 'rb') as file: # 打开指定错误文件error_data = file.read() # 读取指定错误数据response_line = "HTTP/1.1 404 Not Found！！\r\n" # 响应行response_header = "Server: PWS1.0 服务器名称及版本……\r\n" # 响应头response_body = error_data # 响应体response_data = (response_line + response_header + '\r\n').encode('utf-8') + response_body # 拼接响应报文comm_socket.send(response_data) # 发送数据给浏览器else:response_line = "HTTP/1.1 200 OK # 成功！！\r\n"response_header = "Server: PWS1.0 # 服务器名称版本！\r\n"response_body = file_dataresponse_data = (response_line + response_header + '\r\n').encode('utf-8') + response_bodycomm_socket.send(response_data)finally:comm_socket.close() # 关闭服务于客户端的套接字def main():server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建TCP服务端套接字server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) # 设置端口号复用，程序退出端口号立即释放server_socket.bind(("", 9000)) # 绑定端口号server_socket.listen(128) # 设置监听while True:comm_socket, ip_port = server_socket.accept() # 等待接受客户端连接请求thread = threading.Thread(target=handle_client_request, args=(comm_socket,)) # 若客户端和服务器连接成功，创建子线程thread.setDaemon(True) # 设置守护主线程thread.start() # 启动子线程执行对应任务if __name__ == '__main__':main()

输出如下

获取到的数据内容为： GET /index.html HTTP/1.1Host: 127.0.0.1:9000Connection: keep-aliveCache-Control: max-age=0sec-ch-ua: "Google Chrome";v="105", "Not)A;Brand";v="8", "Chromium";v="105"sec-ch-ua-mobile: ?0sec-ch-ua-platform: "Windows"Upgrade-Insecure-Requests: 1User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,;q=0.8,application/signed-exchange;v=b3;q=0.9Sec-Fetch-Site: noneSec-Fetch-Mode: navigateSec-Fetch-User: ?1Sec-Fetch-Dest: documentAccept-Encoding: gzip, deflate, brAccept-Language: zh-CN,zh;q=0.9Cookie: csrftoken=1wb7v0r0BQuokJCxRS4JAO2XApvrHXFP90t2PiYb0mz7AwWS0NoKTi0zNaIjOfTl; Hm_lvt_18f619820855042bca58b32408f44de7=1658219884请求路径为： /index.html

浏览器响应如下

二、静态web服务器-面向对象

实现步骤把提供服务的web服务器抽象为一个类提供web服务器初始化方法，在方法中创建socket对象提供一个开启web服务器的方法，让web服务器处理客户端请求操作

实现代码

import socketimport threadingclass HttpWebServer(object): # 定义web服务器类，将提供服务的web服务器抽象为一个类def __init__(self): # 初始化服务端套接字server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建TCP套接字server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) # 设置端口复用，程序退出端口号立即释放server_socket.bind(("", 9000))# 绑定端口号server_socket.listen(128) # 设置监听self.server_socket = server_socket@staticmethoddef handle_client_request(comm_socket):recv_data = comm_socket.recv(4096) # 接收请求的二进制数据if len(recv_data) == 0:print('未获取到请求数据！')comm_socket.close()returnrecv_content = recv_data.decode('utf-8') # 对二进制数据解码print('获取到的数据内容为：', recv_content)request_list = recv_content.split(" ", maxsplit=2) # 根据指定字符串空格进行分割，最大分割次数为2request_path = request_list[1] # 获取请求资源路径print('请求路径为：', request_path)if request_path == "/": # 判断请求的是否是根目录，若是则返回首页指定数据request_path = "/index.html"try:with open('C:/Users/熊世强/Desktop/ubuntu' + request_path, 'rb') as file: # 动态打开指定文件file_data = file.read() # 读取指定文件数据except Exception as e: # 请求异常，资源不存在，返回指定404错误数据with open('C:/Users/熊世强/Desktop/ubuntu/error.html', 'rb') as file: # 打开指定错误文件error_data = file.read() # 读取指定错误数据response_line = "HTTP/1.1 404 Not Found！！\r\n" # 响应行response_header = "Server: PWS1.0 服务器名称及版本……\r\n" # 响应头response_body = error_data # 响应体response_data = (response_line + response_header + '\r\n').encode('utf-8') + response_body # 拼接响应报文comm_socket.send(response_data) # 发送数据给浏览器else:response_line = "HTTP/1.1 200 OK # 成功！！\r\n"response_header = "Server: PWS1.0 # 服务器名称版本！\r\n"response_body = file_dataresponse_data = (response_line + response_header + '\r\n').encode('utf-8') + response_bodycomm_socket.send(response_data)finally:comm_socket.close() # 关闭服务于客户端的套接字def start(self): # 启动web服务器while True:comm_socket, ip_port = self.server_socket.accept() # 等待接收客户端连接请求thread = threading.Thread(target=self.handle_client_request, args=(comm_socket,)) # 若客户端和服务器建立连接，则创建子线程thread.setDaemon(True) # 设置守护子线程thread.start() # 启动子线程执行对应任务def main(): # 程序入口函数webServer = HttpWebServer() # 创建web服务器对象webServer.start() # 启动web服务器进行工作if __name__ == '__main__':main()

输出如下

获取到的数据内容为： GET /index.html HTTP/1.1Host: 127.0.0.1:9000Connection: keep-aliveCache-Control: max-age=0sec-ch-ua: "Google Chrome";v="105", "Not)A;Brand";v="8", "Chromium";v="105"sec-ch-ua-mobile: ?0sec-ch-ua-platform: "Windows"Upgrade-Insecure-Requests: 1User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/105.0.0.0 Safari/537.36Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,;q=0.8,application/signed-exchange;v=b3;q=0.9Sec-Fetch-Site: noneSec-Fetch-Mode: navigateSec-Fetch-User: ?1Sec-Fetch-Dest: documentAccept-Encoding: gzip, deflate, brAccept-Language: zh-CN,zh;q=0.9Cookie: csrftoken=1wb7v0r0BQuokJCxRS4JAO2XApvrHXFP90t2PiYb0mz7AwWS0NoKTi0zNaIjOfTl; Hm_lvt_18f619820855042bca58b32408f44de7=1658219884请求路径为： /index.html

浏览器响应如下

三、静态web服务器-命令行启动动态绑定端口号

实现步骤获取执行python程序的终端命令行参数判断参数类型，设置端口号必须是整型为web服务器类的初始化方法添加一个端口号参数，用于绑定端口号

实现代码

import socketimport threadingimport sysclass HttpWebServer(object): # 定义web服务器类，将提供服务的web服务器抽象为一个类def __init__(self, port): # 初始化服务端套接字，添加端口号参数server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 创建TCP套接字server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True) # 设置端口复用，程序退出端口号立即释放server_socket.bind(("", port))# 绑定端口号server_socket.listen(128) # 设置监听self.server_socket = server_socket@staticmethoddef handle_client_request(comm_socket): # 处理客户端请求recv_data = comm_socket.recv(4096) # 接收请求的二进制数据if len(recv_data) == 0:print('未获取到请求数据！')comm_socket.close()returnrecv_content = recv_data.decode('utf-8') # 对二进制数据解码print('获取到的数据内容为：', recv_content)request_list = recv_content.split(" ", maxsplit=2) # 根据指定字符串空格进行分割，最大分割次数为2request_path = request_list[1] # 获取请求资源路径print('请求路径为：', request_path)if request_path == "/": # 判断请求的是否是根目录，若是则返回首页指定数据request_path = "/index.html"try:with open('C:/Users/username/Desktop/ubuntu' + request_path, 'rb') as file: # 动态打开指定文件file_data = file.read() # 读取指定文件数据except Exception as e: # 请求异常，资源不存在，返回指定404错误数据with open('C:/Users/username/Desktop/ubuntu/error.html', 'rb') as file: # 打开指定错误文件error_data = file.read() # 读取指定错误数据response_line = "HTTP/1.1 404 Not Found！！\r\n" # 响应行response_header = "Server: PWS1.0 服务器名称及版本……\r\n" # 响应头response_body = error_data # 响应体response_data = (response_line + response_header + '\r\n').encode('utf-8') + response_body # 拼接响应报文comm_socket.send(response_data) # 发送数据给浏览器else:response_line = "HTTP/1.1 200 OK # 成功！！\r\n"response_header = "Server: PWS1.0 # 服务器名称版本！\r\n"response_body = file_dataresponse_data = (response_line + response_header + '\r\n').encode('utf-8') + response_bodycomm_socket.send(response_data)finally:comm_socket.close() # 关闭服务于客户端的套接字def start(self): # 启动web服务器while True:comm_socket, ip_port = self.server_socket.accept() # 等待接收客户端连接请求thread = threading.Thread(target=self.handle_client_request, args=(comm_socket,)) # 若客户端和服务器建立连接，则创建子线程thread.setDaemon(True) # 设置守护子线程thread.start() # 启动子线程执行对应任务def main(): # 程序入口函数print('命令行输入的参数为：', sys.argv)if len(sys.argv) != 2: # 判断命令行参数个数是否为2print("请执行格式为[ python3 xx.py 9000 ] 的命令")returnif not sys.argv[1].isdigit(): # 判断字符串是否为数字组成，必须为整型print("请执行格式为[ python3 xx.py 9000 ]的命令")returnport = int(sys.argv[1]) # 获取终端命令行参数web_server = HttpWebServer(port)web_server.start()if __name__ == '__main__':main()

命令行执行

浏览器响应如下

学习导航：http://xqnav.top/