Python3网络编程-socket编程之Tcp编程

1：Python3中socket编程介绍

这里就不介绍网络编程的基础知识了，比如TCP/IP协议，OSI模型，TCP的三次握手等。下面直接介绍python中socket编程；

2：简单的点对点

只接受单个连接的服务端：

01_TcpServer.py：

# -*- coding: utf-8 -*- import socketimport time server = socket.socket()ip_port = ('127.0.0.1',8000)server.bind(ip_port) # 绑定ip+端口server.listen(10) # 监听 # print(help(server.listen)) print('启动服务：等待客户端的连接......')conn,addr = server.accept()print('客户端已连接:')print(conn)print(addr) while True: print('等待客户端的数据：') client_data = conn.recv(1024) # 接收客户端的数据，接收的数据是当前时间的秒数，下面处理是把秒转换为字符串 client_data = client_data.decode('utf-8') stru_t = time.localtime(float(client_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S',stru_t) print(f'接收来自{addr}的数据：{strTime}' ) conn.send(str(time.time()).encode('utf-8')) # 给客户端发送当前时间的秒数 conn.close()server.close()

01_TcpClient.py

# -*- coding: utf-8 -*- import socketimport timeclient = socket.socket()ip_port = ('127.0.0.1',8000)client.connect(ip_port) while True: client.send(str(time.time()).encode('utf-8')) print('等待服务端的数据：') server_data = client.recv(1024) server_data = server_data.decode('utf-8') stru_t = time.localtime(float(server_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'来自服务端的数据:{strTime}') time.sleep(10) client.close()

3：并发服务端

3.1：创建线程处理

对于每个客户端连接都创建一个线程来处理：

02_TcpServer.py：

# -*- coding: utf-8 -*- import socketfrom socket import SOL_SOCKET,SO_REUSEADDRimport timeimport threading # 线程模块import traceback # 打印异常信息 # 处理每个链接def handle_client(conn, addr): print(conn, addr) while True: try: print(f'等待客户端{addr}的数据：') client_data = conn.recv(1024) # 接收客户端的数据，接收的数据是当前时间的秒数，下面处理是把秒转换为字符串 client_data = client_data.decode('utf-8') stru_t = time.localtime(float(client_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'接收来自{addr}的数据：{strTime}') conn.send(str(time.time()).encode('utf-8')) # 给客户端发送当前时间的秒数 except ConnectionResetError: val = traceback.format_exc() print(val) break except Exception: val = traceback.format_exc() print(val) print(f'关闭链接{addr}') conn.close() def StartTcpServer(ip,port): server = socket.socket() server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # 在bind前加，允许地址重用 server.bind((ip,port)) # 绑定ip+端口 server.listen(10) # 监听 # print(help(server.accept)) while True: try: print('等待客户端的连接......') conn, addr = server.accept() # 一直会阻塞在这里，直到有新的连接进来 # 创建一个线程来处理每个链接 threading.Thread(target=handle_client, args=(conn, addr)).start() except Exception: val = traceback.format_exc() print(val) break # 有异常退出循环 server.close() if __name__ == '__main__': StartTcpServer('127.0.0.1', 8000)

02_TcpClient.py：

# -*- coding: utf-8 -*- import socketimport timeimport traceback def StartTcpClient(ip_port): # 开3个客户端 client_list = [socket.socket() for i in range(3)] # ip_port = ('127.0.0.1',8000) for client in client_list: print(client) client.connect(ip_port) while True: for index,client in enumerate(client_list): try: client.send(str(time.time()).encode('utf-8')) print(f'clirnt[{index}],等待服务端的数据：') server_data = client.recv(1024) server_data = server_data.decode('utf-8') stru_t = time.localtime(float(server_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'来自服务端的数据:{strTime}') time.sleep(2) except Exception: val = traceback.format_exc() print(val) client.close() client_list.remove(client) if __name__ == '__main__': StartTcpClient(('127.0.0.1',8000))

3.2：socketserver模块处理

在上面使用每来一个连接，就创建一个线程的方式来处理，如果连接的数量过多，创建线程就会出现问题。

在Python中提供了socketserver模块，socketserver在内部使用IO多路复用以及多线程/进程机制，实现了并发处理多个客户端请求的socket服务端。

03_TcpServer.py

# -*- coding: utf-8 -*- import socketserverimport timeimport traceback # 打印异常信息 def handle_client_data(data,addr): try: data = data.decode('utf-8') # print(type(data),data) stru_t = time.localtime(int(data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f"来自{addr}的客户端向你发来信息:{data},转换之后:{strTime}" ) except Exception: pass # 每个链接都使用一个 TcpServer 实例对象来处理，并且会自动调用handle方法，退出handle方法链接就会自动断开。class TcpServer(socketserver.BaseRequestHandler): """ 必须继承socketserver.BaseRequestHandler类 """ def handle(self): """ 必须实现这个方法！ :return: """ conn = self.request # request里封装了所有请求的数据 conn.sendall(str(time.time()).encode('utf-8')) print(f"1111：{self.client_address}") # 打日志验证每个链接是不是都会进来一次 while True: try: data = conn.recv(1024) handle_client_data(data,self.client_address) conn.sendall(str(int(time.time())).encode('utf-8')) except Exception: val = traceback.format_exc() print(val) break print(f'退出客户端{self.client_address}的处理。') def StartTcpServer(ip,port): # 创建一个多线程TCP服务器 server = socketserver.ThreadingTCPServer((ip,port), TcpServer) print("启动socketserver服务器！") # 启动服务器，服务器将一直保持运行状态 server.serve_forever() if __name__ == '__main__': StartTcpServer('127.0.0.1', 8000)

03_TcpClient.py：

# -*- coding: utf-8 -*- import socketimport timeimport traceback def StartTcpClient(ip_port,nums): client_list = [socket.socket() for i in range(nums)] for client in client_list: print(client) client.connect(ip_port) while True: try: for index,client in enumerate(client_list): try: # client.send(str( int(time.time()) ).encode('utf-8')) client.sendall(str( int(time.time()) ).encode('utf-8')) print(f'clirnt[{index}],等待服务端的数据：') server_data = client.recv(1024) # print(f'来自服务端的数据:{server_data}') try: server_data = server_data.decode('utf-8') stru_t = time.localtime(float(server_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'来自服务端的数据:{server_data} 转换之后：{strTime}') except Exception: pass time.sleep(1) except (ConnectionResetError,ConnectionAbortedError): val = traceback.format_exc() print(val) client.close() client_list.remove(client) if len(client_list) < 1: raise except Exception: val = traceback.format_exc() print(val) except Exception: val = traceback.format_exc() print(val) print('退出') break def test(): print(str(time.time() ),time.time(),int(time.time())) if __name__ == '__main__': StartTcpClient(('127.0.0.1',8000),1) test()

3.3：使用select模块

Python中的select模块专注于I/O多路复用，提供了select  poll  epoll三个方法(其中后两个在Linux中可用，windows仅支持select)，另外也提供了kqueue方法(freeBSD系统)

select方法：

进程指定内核监听哪些文件描述符(最多监听1024个fd)的哪些事件，当没有文件描述符事件发生时，进程被阻塞；当一个或者多个文件描述符事件发生时，进程被唤醒。

04_TcpServer.py：

# -*- coding: utf-8 -*- import socketimport selectfrom socket import SOL_SOCKET,SO_REUSEADDRimport timeimport traceback # 打印异常信息 def handle_client_data(data,addr): try: data = data.decode('utf-8') # print(type(data),data) stru_t = time.localtime(int(data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f"来自{addr.getpeername()}的客户端向你发来信息:{data},转换之后:{strTime}") except Exception: pass def StartTcpServer(ip,port): server = socket.socket() server.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) # 在bind前加，允许地址重用 server.bind((ip,port)) # 绑定ip+端口 server.listen(10) # 监听 read_fd_list = [server,] while True: try: r_list, w_list, error_list = select.select(read_fd_list, [], [], 1) # 超时设置为1秒 stru_t = time.localtime(time.time()) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'select之后：len(r_list)={len(r_list)},时间:{strTime}') for fd in r_list: # r_list 为可读的文件描述符列表 if fd == server: # 可读的文件描述符 为 server（上面起服务创建的），说明有链接请求过来。 conn, addr = fd.accept() # 接收客户端链接 print(addr) read_fd_list.append(conn) # 把客户端，加入读轮询列表 data = conn.recv(1024) print(f'接收数据：msg={data}') handle_client_data(data, conn) conn.sendall(str(time.time()).encode('utf-8')) # sendall 会循环调用send，把所有数据都发出去。 # conn.send(str(time.time()).encode('utf-8')) # send 发送出去的数据可能是部分数据 else: # 客户端可读，就是有客户端发送数据过来。 try: data = fd.recv(1024) # print(help(fd)) handle_client_data(data,fd) fd.sendall(str(int(time.time())).encode('utf-8')) except (ConnectionResetError,ConnectionAbortedError,ConnectionRefusedError): val = traceback.format_exc() print(val) fd.close() read_fd_list.remove(fd) except Exception: val = traceback.format_exc() print(val) break # 有异常退出循环 server.close() if __name__ == '__main__': StartTcpServer('127.0.0.1', 8000)

04_TcpClient.py：

# -*- coding: utf-8 -*- import socketimport timeimport traceback def StartTcpClient(ip_port,nums): client_list = [socket.socket() for i in range(nums)] for client in client_list: print(client) client.connect(ip_port) while True: try: for index,client in enumerate(client_list): try: client.sendall(str( int(time.time()) ).encode('utf-8')) # client.send(str( int(time.time()) ).encode('utf-8')) print(f'clirnt[{index}],等待服务端的数据：') server_data = client.recv(1024) # print(f'来自服务端的数据:{server_data}') server_data = server_data.decode('utf-8') stru_t = time.localtime(float(server_data)) strTime = time.strftime('%Y-%m-%d %H:%M:%S', stru_t) print(f'来自服务端的数据:{server_data} 转换之后：{strTime}') # time.sleep(1) except (ConnectionResetError, ConnectionAbortedError, ConnectionRefusedError): val = traceback.format_exc() print(val) client.close() client_list.remove(client) if len(client_list) < 1: raise except Exception: val = traceback.format_exc() print(val) except Exception: val = traceback.format_exc() print(val) print('退出') break if __name__ == '__main__': StartTcpClient(('127.0.0.1',8000),1) # StartTcpClient(('127.0.0.1',8000),2000)

4：socket实现web服务器

4.1：简单web服务器

在浏览器中访问  http://127.0.0.1:8000/home 等

# -*- coding: utf-8 -*-import socket def StartWebServer(ip,port): # 创建socket对象 sk = socket.socket() # 绑定IP和端口 sk.bind((ip,port)) # 监听 sk.listen() while True: # 等待连接 conn, addr = sk.accept() # 接收数据 data = conn.recv(8096) print(data) # 返回状态行 conn.send(b'HTTP/1.1 200 OK\r\n\r\n') # 返回数据 try: data = data.decode('utf-8') url = data.split()[1] ret_data = url if url == "/test": ret_data = "test" elif url == "/home": ret_data = "home" elif url == "/index": ret_data = "index" else: ret_data = "404" conn.send(f'{ret_data}'.encode('utf-8')) except Exception: pass # 关闭连接 conn.close() if __name__ == '__main__': StartWebServer('127.0.0.1', 8000)

4.2：简单web服务器：函数版本

# -*- coding: utf-8 -*-import socketimport time def test(url): ret = f'test: {url}' return ret.encode('utf-8') def index(url): ret = f'index: {url}' return ret.encode('utf-8') def home(url): ret = f'home: {url}' return ret.encode('utf-8') def gettime(url): now = time.strftime('%Y-%m-%d %H:%M:%S') ret = f'time: {url} {now}' return ret.encode('utf-8') url_map = {'/test':test,'/index':index,'/home':home,'/time':gettime} def handle_client_data(data): data = data.decode('utf-8') url = data.split()[1] ret_data = url if url in url_map: func = url_map[url] ret_data = func(url) else: ret_data = b"404" return ret_data def StartWebServer(ip,port): # 创建socket对象 sk = socket.socket() # 绑定IP和端口 sk.bind((ip,port)) # 监听 sk.listen() while True: # 等待连接 conn, addr = sk.accept() # 接收数据 data = conn.recv(8096) print(data) # 返回状态行 conn.send(b'HTTP/1.1 200 OK\r\n\r\n') # 返回数据 try: ret_data = handle_client_data(data) conn.send(ret_data) except Exception: pass # 关闭连接 conn.close() if __name__ == '__main__': StartWebServer('127.0.0.1', 8000)

在浏览器中访问：http://127.0.0.1:8000/time

4.3：简单web服务器：返回动态页面

# -*- coding: utf-8 -*-import socketimport time def test(url): ret = f'test: {url}' return ret.encode('utf-8') def index(url): with open('index.html', 'rb') as f: ret = f.read() return ret def home(url): ret = f'home: {url}' return ret.encode('utf-8') def gettime(url): now = time.strftime('%Y-%m-%d %H:%M:%S') with open('time.html', 'r', encoding='utf-8') as f: data = f.read() data = data.replace('##time##', now) return data.encode('utf-8') url_map = {'/test':test,'/index':index,'/home':home,'/time':gettime} def handle_client_data(data): data = data.decode('utf-8') url = data.split()[1] ret_data = url if url in url_map: func = url_map[url] ret_data = func(url) else: ret_data = b"404" return ret_data def StartWebServer(ip,port): # 创建socket对象 sk = socket.socket() # 绑定IP和端口 sk.bind((ip,port)) # 监听 sk.listen() while True: # 等待连接 conn, addr = sk.accept() # 接收数据 data = conn.recv(8096) print(data) # 返回状态行 conn.send(b'HTTP/1.1 200 OK\r\n\r\n') # 返回数据 try: ret_data = handle_client_data(data) conn.send(ret_data) except Exception: pass # 关闭连接 conn.close() if __name__ == '__main__': StartWebServer('127.0.0.1', 8000)

index.html

index index index index index

time.html

time

4.4：并发web服务器

使用socketserver实现：

# -*- coding: utf-8 -*- import socketserverimport timeimport traceback # 打印异常信息 def test(url): ret = f'test: {url}' return ret.encode('utf-8') def index(url): with open('index.html', 'rb') as f: ret = f.read() return ret def home(url): ret = f'home: {url}' return ret.encode('utf-8') def gettime(url): now = time.strftime('%Y-%m-%d %H:%M:%S') with open('time.html', 'r', encoding='utf-8') as f: data = f.read() data = data.replace('##time##', now) return data.encode('utf-8') url_map = {'/test':test,'/index':index,'/home':home,'/time':gettime} def handle_client_data(data): data = data.decode('utf-8') url = data.split()[1] ret_data = url if url in url_map: func = url_map[url] ret_data = func(url) else: ret_data = b"404" return ret_data # 每个链接都使用一个 TcpServer 实例对象来处理，并且会自动调用handle方法，退出handle方法链接就会自动断开。class TcpServer(socketserver.BaseRequestHandler): """ 必须继承socketserver.BaseRequestHandler类 """ def handle(self): """ 必须实现这个方法！ :return: """ conn = self.request # request里封装了所有请求的数据 # conn.sendall(str(time.time()).encode('utf-8')) # print(f"1111：{self.client_address}") # 打日志验证每个链接是不是都会进来一次 while True: try: data = conn.recv(8096) print(data) conn.send(b'HTTP/1.1 200 OK\r\n\r\n') # 返回数据 try: ret_data = handle_client_data(data) conn.sendall(ret_data) except Exception: pass break except Exception: val = traceback.format_exc() print(val) break print(f'退出客户端{self.client_address}的处理。') def StartTcpServer(ip,port): # 创建一个多线程TCP服务器 server = socketserver.ThreadingTCPServer((ip,port), TcpServer) print("启动socketserver服务器！") # 启动服务器，服务器将一直保持运行状态 server.serve_forever() if __name__ == '__main__': StartTcpServer('127.0.0.1', 8000)