网站建设 >

查看其它板块

46网络编程_socketserver-创新互联

成都创新互联公司是一家企业级云计算解决方案提供商,超15年IDC数据中心运营经验。主营GPU显卡服务器，站群服务器，成都移动机房，海外高防服务器，机柜大带宽、租用·托管，动态拨号VPS，海外云手机，海外云服务器，海外服务器租用托管等。

socketserver模块：...1

编程接口：...2

总结，创建服务器步骤：...4

例，实现EchoServer：...4

例，改写ChatServer：...5

socketserver模块：

socket过于底层，编程虽有套路，但想要写出健壮的代码比较困难，所以很多语言都对socket底层API进行封装，py的封装就是socketserver模块，网络服务编程框架，全球企业级快速开发；

socketserver简化了网络服务器的编写；

+------------+

| BaseServer |

+------------+

+-----------+ +------------------+

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

+-----------+ +------------------+

+-----------+ +--------------------+

| 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()

输出：

('127.0.0.1', 7576)

handler

{'request': , 'client_address': ('127.0.0.1', 7576), 'server': }

{'__doc__': None, '__module__': '__main__', 'handle': }

(,)

{'setup': , '__init__': , '__dict__': , '__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': , '__weakref__': , 'finish': }

[<_MainThread(MainThread, started 4136)>, ]

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()

输出：

>>> [, , <_MainThread(MainThread, started 9820)>]

('127.0.0.1', 8000) {}

[, , , <_MainThread(MainThread, started 9820)>]

('127.0.0.1', 8003) {}