46網絡編程_socketserver

發布時間：2020-07-22 17:39:49 來源：網絡閱讀：338 作者：chaijowin 欄目：編程語言

socketserver模塊：... 1

編程接口：... 2

總結，創建服務器步驟：... 4

例，實現EchoServer：... 4

例，改寫ChatServer：... 5

socketserver模塊：

socket過于底層，編程雖有套路，但想要寫出健壯的代碼比較困難，所以很多語言都對socket底層API進行封裝，py的封裝就是socketserver模塊，網絡服務編程框架，全球企業級快速開發；

socketserver簡化了網絡服務器的編寫；

??????? +------------+

??????? | BaseServer |

??????? +------------+

????????????? |

????????????? v

??????? +-----------+??????? +------------------+

??????? | TCPServer |------->| UnixStreamServer |

??????? +-----------+??????? +------------------+

???? ?????????|

????????????? v

??????? +-----------+??????? +--------------------+

??????? | UDPServer |------->| UnixDatagramServer |

??????? +-----------+??????? +--------------------+

4個sync同步類：

TCPServer、UDPServer、UnixStreamServer、UnixDatagramServer；

很少用；

2個mixin類：

ForkingMixIn、ThreadingMixIn；

4個async異步類，生產中常用：

ForkingTCPServer(ForkingMixIn,TCPServer)、ForkingUDPServer(ForkingMixIn,UDPServer)?? #創建多進程

ThreadingTCPServer(ThreadingMixIn,TCPServer)、ThreadingUDPServer(ThreadingMixIn,UDPServer) ??#創建多線程

注：

一般ThreadingTCPServer夠用；

如果并發很高可考慮用ForkingTCPServer；

ThreadingUDPServer甚至也很少用，盡管在LAN中，如果忙起來時接收到的包的順序是亂的；

編程接口：

class BaseServer:

??? def __init__(self, server_address, RequestHandlerClass):?? #服務器綁定的地址信息；用于處理請求，該類必須是BaseRequestHandler類的子類

??? def finish_request(self, request, client_address):?? #處理請求的方法

??????? """Finish one request by instantiating RequestHandlerClass."""

??????? self.RequestHandlerClass(request, client_address, self)?? #實例化，RequesthandlerClass的構造

查看源碼，寫框架的思想：

class BaseRequestHandler:?? #和用戶連接的用戶請求處理類，server實例接收用戶請求后，最后會實例化這個類；它會一次調用三個函數setup()（每一個連接初始化）、handler()（每一次請求處理，必須覆蓋）、finish()（每一個連接清理），子類可覆蓋

??? def __init__(self, request, client_address, server):?? #初始化時送入3個構造參數，request、client_address、server（TCPServer），以后可在BaseRequestHandler類的實例上使用self.request（和client連接的socket對象）、self.cleint_address（是客戶端地址）、self.server（是TCPServer本身）

??????? self.request = request

??????? self.client_address = client_address

??????? self.server = server

??????? self.setup()

??????? try:

??????????? self.handle()

??????? finally:

??????????? self.finish()

??? def setup(self):?? #每一個連接初始化，初始化工作，如ChatServer中維護的數據結構放到此段；實現了這三個方法，只不過是空操作，而raise NotImplementedError稱為抽象，不實現

??????? pass

??? def handle(self):?? #每一次請求處理，必須覆蓋；handle()和sock.accept()對應，用戶連接請求過來后，建立連接并生成一個socket對象（保存在self.request中）和客戶端地址（保存在self.client_address中），之后的操作就和socket編程一樣了

??????? pass

??? def finish(self):?? #每一個連接清理，清理工作

??????? pass

注：

setup()和finish()只執行一次；

handler()在不加鎖情況下，也是執行一次；

例：

class MyHandler(socketserver.BaseRequestHandler):?? #右鍵MyHandler，Generate-->Overwrite Methods，可快速生成要覆蓋的方法

??? def handle(self):

??????? super().handle()?? #此句可不寫，因為父類中的handler()為空操作；但如果是StreamRequestHandler則必須要寫，該類中實現了handler()方法

??????? print(self.request, self.client_address, self.server)

??????? print('{} handler'.format(self.__class__))

??????? print(self.__dict__)

??????? print(type(self).__dict__)

??????? print(self.__class__.__bases__[0].__dict__)

??????? print(threading.enumerate(), threading.current_thread())

??????? # pass?? #TODO? ?#提醒自己還沒寫完

??????? print('come')

??????? for i in range(3):?? #client和server端長時間連接，在handler里循環；分布式服務之間需傳遞心跳包（傳遞事務、節點信息等），服務之間要長連接，不能斷；數據庫連接池不應用長連接，傳完數據就可斷開，有很多連接等著連DB

??????????? data = self.request.recv(1024)

??????????? print(data)

addr = ('127.0.0.1', 9998)

server = socketserver.ThreadingTCPServer(addr, MyHandler)?? #用多client連接測

# server = socketserver.TCPServer(addr, MyHandler)?? #同步，等前一個連接斷開后，才能接收并處理下一個連接的請求

server.serve_forever()?? #啟動大循環，類似while

server.shutdown()

server.server_close()?? #建議關閉連接前先server.shutdown()

輸出：

<socket.socket fd=232, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9998), raddr=('127.0.0.1', 7576)> ('127.0.0.1', 7576) <socketserver.ThreadingTCPServer object at 0x0000000000B656A0>

<class '__main__.MyHandler'> handler

{'request': <socket.socket fd=232, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=0, laddr=('127.0.0.1', 9998), raddr=('127.0.0.1', 7576)>, 'client_address': ('127.0.0.1', 7576), 'server': <socketserver.ThreadingTCPServer object at 0x0000000000B656A0>}

{'__doc__': None, '__module__': '__main__', 'handle': <function MyHandler.handle at 0x0000000001231C80>}

(<class 'socketserver.BaseRequestHandler'>,)

{'setup': <function BaseRequestHandler.setup at 0x0000000001477A60>, '__init__': <function BaseRequestHandler.__init__ at 0x00000000014779D8>, '__dict__': <attribute '__dict__' of 'BaseRequestHandler' objects>, '__module__': 'socketserver', '__doc__': 'Base class for request handler classes.\n\n??? This class is instantiated for each request to be handled.? The\n??? constructor sets the instance variables request, client_address\n??? and server, and then calls the handle() method.? To implement a\n??? specific service, all you need to do is to derive a class which\n? ??defines a handle() method.\n\n??? The handle() method can find the request as self.request, the\n??? client address as self.client_address, and the server (in case it\n??? needs access to per-server information) as self.server.? Since a\n??? separate instance is created for each request, the handle() method\n??? can define other arbitrary instance variables.\n\n??? ', 'handle': <function BaseRequestHandler.handle at 0x0000000001477AE8>, '__weakref__': <attribute '__weakref__' of 'BaseRequestHandler' objects>, 'finish': <function BaseRequestHandler.finish at 0x0000000001477B70>}

[<_MainThread(MainThread, started 4136)>, <Thread(Thread-1, started 4372)>] <Thread(Thread-1, started 4372)>

come

總結，創建服務器步驟：

1、class MyHandler(socketserver.BaseRequestHandler):，通過對BaseRequestHandler類進行子類化并覆蓋其handle()方法，來創建請求處理程序類，此方法處理傳入請求；

2、server=socketserver.ThreadingTCPServer(addr,MyHandler)，必須實例化一個服務器類，并向其傳入服務器的地址和請求處理程序類；

3、server.serve_forever()或server.handle_request()，調用服務器對象的serve_forever()（一直啟動）或server.handle_request()（一次性的）方法；

4、server.shutdown()、server.close()，調用server.close()（關閉套接字）前先server.shutdown()等待停止server.serve_forever()；

為每一個連接提供RequestHandlerClass類實例，一次調用setup()、handler()、finish()方法，且使用了try...finally結構（查看BaseRequestHandler源碼）保證finish()方法一定能被調用，這些方法一次執行完成；

如果想維持這個連接與客戶端通信，需要在handler()中使用循環；

socketserver模塊提供不同的類，但編程接口是一樣的，即使是多進程、多線程的類也是一樣，大大減少了編程的難度；

例，實現EchoServer：

client發來什么，就返回什么消息；

class EchoHandler(socketserver.BaseRequestHandler):

??? def setup(self):

??????? super().setup()

??????? self.event = threading.Event()

??? def handle(self):

??????? super().handle()

??????? while not self.event.is_set():

??????????? data = self.request.recv(1024)

??????????? data = data.decode()

??????????? msg = 'ack: {} {}'.format(self.client_address, data)

??????????? msg = msg.encode()

??????????? self.request.send(msg)

??????? print('end')

??? def finish(self):

??????? super().finish()

??????? self.event.set()

addr = ('127.0.0.1', 9998)

server = socketserver.ThreadingTCPServer(addr, EchoHandler)

# server.serve_forever()

server_thread = threading.Thread(target=server.serve_forever, daemon=True)

server_thread.start()

# server.shutdown()

# server.server_close()

try:

??? while True:

??????? cmd = input('>>> ')

??????? if cmd.strip() == 'quit':?? #只有在client都斷開，與server端沒有連接時才正常退出

??????????? break

except Exception as e:

??? print(e)

except KeyboardInterrupt:

??? print('exit')

finally:

??? server.shutdown()

??? server.server_close()

例，改寫ChatServer：

如果使用文件處理，使用StreamRequestHandler；

可用心跳機制；

class ChatHandler(socketserver.BaseRequestHandler):

??? clients = {}

??? def setup(self):

??????? super().setup()

??????? self.event = threading.Event()

??????? print(self.client_address, threading.current_thread(), self.clients)

??? def handle(self):

??????? super().handle()

??????? while not self.event.is_set():

??????????? try:?? #緩沖區異常、連接異常最好自己捕獲到，雖然父類中有try，但最好自己捕獲

??????????????? data = self.request.recv(1024).decode().strip()

???????????????????????????????????? if len(data) == 0:?? #同if not data，解決client主動斷開后產生的異常，20180901追加尚未測試

?????????????????????????????????????????????? raise BrokenPipeError('client broken')

??????????? except Exception as e:

??????????????? logging.info(e)

??????????????? data = 'quit'?? #技巧，某個連接一旦有問題，會有各種異常，此處直接斷開

??????????? logging.info(data)

??????????? if data == 'quit':

??????????????? break

??? ????????self.clients[self.client_address] = self.request

??????????? msg = 'ack: {}'.format(data)

??????????? for c in self.clients.values():

??????????????? c.send(msg.encode())

??? def finish(self):

??????? super().finish()

??????? self.clients.pop(self.client_address)

??????? self.event.set()

addr = ('127.0.0.1', 9998)

server = socketserver.ThreadingTCPServer(addr, ChatHandler)

server_thread = threading.Thread(target=server.serve_forever, daemon=True)

server_thread.start()

myutils.show_threads()?? #在主線程中就可，沒必要放到工作線程中

try:

??? while True:

??????? cmd = input('>>> ').strip()

??????? if cmd == 'quit':

??????????? break

except Exception as e:

??? print(e)

except KeyboardInterrupt:

??? print('exit')

finally:

??? server.shutdown()

??? server.server_close()

輸出：

>>> [<Thread(Thread-1, started daemon 9552)>, <Thread(show_threads, started daemon 9644)>, <_MainThread(MainThread, started 9820)>]

('127.0.0.1', 8000) <Thread(Thread-2, started 4008)> {}

[<Thread(Thread-1, started daemon 9552)>, <Thread(show_threads, started daemon 9644)>, <Thread(Thread-2, started 4008)>, <_MainThread(MainThread, started 9820)>]

('127.0.0.1', 8003) <Thread(Thread-3, started 9456)> {}