SocketServer 实现并发

【十一】socketserver实现并发#

基于tcp的套接字，关键就是两个循环
- 一个链接循环
- 一个通信循环
socketserver模块中分两大类：server类（解决链接问题）和request类（解决通信问题）
server类：

request类：

以下述代码为例，分析socketserver源码：
- ftpserver=socketserver.ThreadingTCPServer((‘127.0.0.1’,8080),FtpServer) ftpserver.serve_forever()
查找属性的顺序：
- ThreadingTCPServer->ThreadingMixIn->TCPServer->BaseServer
实例化得到ftpserver
- 先找类ThreadingTCPServer的init
- 在TCPServer中找到
- 进而执行server_bind,server_active
找ftpserver下的serve_forever
- 在BaseServer中找到，进而执行self._handle_request_noblock()，该方法同样是在BaseServer中
执行self._handle_request_noblock()进而执行request, client_address = self.get_request()（就是TCPServer中的self.socket.accept()）
- 然后执行self.process_request(request, client_address)
在ThreadingMixIn中找到process_request，开启多线程应对并发
- 进而执行process_request_thread
- 执行self.finish_request(request, client_address)
上述四部分完成了链接循环
- 本部分开始进入处理通讯部分
- 在BaseServer中找到finish_request
- 触发我们自己定义的类的实例化
- 去找init方法，而我们自己定义的类没有该方法
- 则去它的父类也就是BaseRequestHandler中找….
源码分析总结：
- 基于tcp的socketserver我们自己定义的类中的
  - self.server即套接字对象
  - self.request即一个链接
  - self.client_address即客户端地址
- 基于udp的socketserver我们自己定义的类中的
  - self.request是一个元组（第一个元素是客户端发来的数据，第二部分是服务端的udp套接字对象），如(b’adsf’, )
  - self.client_address即客户端地址
FtpServer

1
import socketserver
2
import struct
3
import json
4
import os
5
class FtpServer(socketserver.BaseRequestHandler):
6
    coding='utf-8'
7
    server_dir='file_upload'
8
    max_packet_size=1024
9
    BASE_DIR=os.path.dirname(os.path.abspath(__file__))
10
    def handle(self):
11
        print(self.request)
12
        while True:
13
            data=self.request.recv(4)
14
            data_len=struct.unpack('i',data)[0]
15
            head_json=self.request.recv(data_len).decode(self.coding)
16
            head_dic=json.loads(head_json)
17
            # print(head_dic)
18
            cmd=head_dic['cmd']
19
            if hasattr(self,cmd):
20
                func=getattr(self,cmd)
21
                func(head_dic)
22
    def put(self,args):
23
        file_path = os.path.normpath(os.path.join(
24
            self.BASE_DIR,
25
            self.server_dir,
26
            args['filename']
27
        ))
28

29
        filesize = args['filesize']
30
        recv_size = 0
31
        print('----->', file_path)
32
        with open(file_path, 'wb') as f:
33
            while recv_size < filesize:
34
                recv_data = self.request.recv(self.max_packet_size)
35
                f.write(recv_data)
36
                recv_size += len(recv_data)
37
                print('recvsize:%s filesize:%s' % (recv_size, filesize))
38

39

40
ftpserver=socketserver.ThreadingTCPServer(('127.0.0.1',8080),FtpServer)
41
ftpserver.serve_forever()

FtpClient

1
import socket
2
import struct
3
import json
4
import os
5

6

7

8
class MYTCPClient:
9
    address_family = socket.AF_INET
10

11
    socket_type = socket.SOCK_STREAM
12

13
    allow_reuse_address = False
14

15
    max_packet_size = 8192
16

17
    coding='utf-8'
18

19
    request_queue_size = 5
20

21
    def __init__(self, server_address, connect=True):
22
        self.server_address=server_address
23
        self.socket = socket.socket(self.address_family,
24
                                    self.socket_type)
25
        if connect:
26
            try:
27
                self.client_connect()
28
            except:
29
                self.client_close()
30
                raise
31

32
    def client_connect(self):
33
        self.socket.connect(self.server_address)
34

35
    def client_close(self):
36
        self.socket.close()
37

38
    def run(self):
39
        while True:
40
            inp=input(">>: ").strip()
41
            if not inp:continue
42
            l=inp.split()
43
            cmd=l[0]
44
            if hasattr(self,cmd):
45
                func=getattr(self,cmd)
46
                func(l)
47

48

49
    def put(self,args):
50
        cmd=args[0]
51
        filename=args[1]
52
        if not os.path.isfile(filename):
53
            print('file:%s is not exists' %filename)
54
            return
55
        else:
56
            filesize=os.path.getsize(filename)
57

58
        head_dic={'cmd':cmd,'filename':os.path.basename(filename),'filesize':filesize}
59
        print(head_dic)
60
        head_json=json.dumps(head_dic)
61
        head_json_bytes=bytes(head_json,encoding=self.coding)
62

63
        head_struct=struct.pack('i',len(head_json_bytes))
64
        self.socket.send(head_struct)
65
        self.socket.send(head_json_bytes)
66
        send_size=0
67
        with open(filename,'rb') as f:
68
            for line in f:
69
                self.socket.send(line)
70
                send_size+=len(line)
71
                print(send_size)
72
            else:
73
                print('upload successful')
74

75

76
client=MYTCPClient(('127.0.0.1',8080))
77

78
client.run()

关于并发的概念,我们将会在下一章中进行更详细的讲述

音乐

音乐

【十一】socketserver实现并发#

支持与分享