[

1. XMLRPC接口服务

XMLRPC是python标准库xmlrpc提供的RPC框架,一般我们会根据不同的需求配合使用标准库中的socketserver.ThreadingMixIn或者socketserver.ForkingMixIn来使用多线程或者多进程(注意只能在支持fork的平台使用)处理请求.注意XMLRPC并不安全,需要防止被人利用XML漏洞

下面是一个简单的例子,我们把它放在C0中.这个例子我们计算输入的字符串的md5哈希值

1.1. 最简单的服务端

logger.py用于定义结构化log. xmlrpc默认的会使用方法SimpleXMLRPCRequestHandler.log_error(format, *args)和SimpleXMLRPCRequestHandler.log_request(code='-', size="-")打印请求信息和错误信息到stderr,默认的打印方式是plain text格式的,这种非结构化的数据往往难以收集分析,我们使用structlog将其结构化.

import sys import logging import structlog LOG_LEVEL = logging.INFO SERVER_NAME = "xmlrpc-server" structlog.configure(     processors=[         structlog.stdlib.filter_by_level,  # 判断是否接受某个level的log消息         structlog.stdlib.add_logger_name,  # 增加字段logger         structlog.stdlib.add_log_level,  # 增加字段level         structlog.stdlib.PositionalArgumentsFormatter(),         structlog.processors.TimeStamper(fmt="iso"),  # 增加字段timestamp且使用iso格式输出         structlog.processors.StackInfoRenderer(),         structlog.processors.format_exc_info,  # 捕获异常的栈信息         structlog.processors.StackInfoRenderer(),  # 详细栈信息         structlog.processors.JSONRenderer()  # json格式输出,第一个参数会被放入event字段     ],     context_class=dict,     logger_factory=structlog.stdlib.LoggerFactory(),     wrapper_class=structlog.stdlib.BoundLogger,     cache_logger_on_first_use=True, )  handler = logging.StreamHandler(sys.stdout) root_logger = logging.getLogger() root_logger.addHandler(handler) root_logger.setLevel(LOG_LEVEL)  # 设置最低log等级 log = structlog.get_logger(SERVER_NAME)

server.py用于实现算法逻辑.

import time from hashlib import md5 from http import HTTPStatus from xmlrpc.server import SimpleXMLRPCServer from xmlrpc.server import SimpleXMLRPCRequestHandler from socketserver import ThreadingMixIn from logger import log   HOST = "localhost" PORT = 5000  class RequestHandler(SimpleXMLRPCRequestHandler):     rpc_paths = ('/XMLRPC',)     def log_error(self, format, *args):         """Log an error.          This is called when a request cannot be fulfilled.  By         default it passes the message on to log_message().          Arguments are the same as for log_message().          XXX This should go to the separate error log.          """         log.error("server error",address=self.address_string(),errormsg=format%args)      def log_request(self, code='-', size="-"):         """Log an accepted request.          This is called by send_response().          """         if isinstance(code, HTTPStatus):             code = code.value         log.info("request info",address=self.address_string(),requestline=self.requestline, code=str(code), size=str(size))  class ThreadingXMLRPCServer(ThreadingMixIn, SimpleXMLRPCServer):     pass   def md5_func(text: str)->str:     """md5哈希."""     start = time.time()     result = md5(text.encode("utf-8")).hexdigest()     end = time.time()     log.info("time it",seconds=end-start)     return result   def main():     with ThreadingXMLRPCServer((HOST, PORT), requestHandler=RequestHandler, allow_none=True) as server:         # 注册所有可调用函数的名字到system.listMethods方法         # 注册可调用函数的docstring到system.methodHelp(func_name)方法         # 注册可调用函数的签名到system.methodSignature(func_name)方法         server.register_introspection_functions()          # 这个函数的作用是可以使客户端同时调用服务端的的多个函数。         server.register_multicall_functions()          # 注册一个函数,使它可以被调用,后面的字符串就是被调用的函数名         server.register_function(md5_func, md5_func.__name__)         try:             log.info("server start", msg=f"xmlrpc start @ {HOST}:{PORT}!")             server.serve_forever()         except Exception:             log.info("server error", exc_info=True, stack_info=True)             raise         finally:             log.info("server stoped", msg=f"xmlrpc @ {HOST}:{PORT} stoped!")   if __name__ == "__main__":     main()

xmlrpc只要将函数注册到server对象,然后执行server_forever()方法即可.

1.2. 最简单的客户端

import xmlrpc.client  HOST = "localhost" PORT = 5000  with xmlrpc.client.ServerProxy(f"http://{HOST}:{PORT}/XMLRPC") as proxy:     result = proxy.md5_func("这是一个测试")  print(result)

这个例子中我们使用多线程以提高rpc服务的并发性能,但一般来说rpc是计算密集型任务,并发上去并不解决问题,更关键的是将每个核利用起来,我们当然也可以多起实例通多代理做负载均衡,但更简单的方案是利用多核将重计算的部分放在多进行中执行.

1.3. 利用多核

在多进程部分我们讲过如何使用concurrent.futures进行高层抽象的多进程操作,这边我们还是使用这种方式,代码C1展示了如何改造上面的例子

server.py

... from concurrent.futures import ProcessPoolExecutor,wait  ...  WORKER = 4  ...  # 注意此处修改函数名字 def _md5_func(text: str)->str:     """md5哈希."""     start = time.time()     result = md5(text.encode("utf-8")).hexdigest()     end = time.time()     log.info("time it",seconds=end-start)     return result   def main():     with ProcessPoolExecutor(WORKER) as executor:         # 将要执行的业务包装,实际执行交给执行器         def md5_func(text:str)->str:             fut = executor.submit(_md5_func, text)             wait([fut])             return fut.result()          with ThreadingXMLRPCServer((HOST, PORT), requestHandler=RequestHandler, allow_none=True) as server:             # 注册所有可调用函数的名字到system.listMethods方法             # 注册可调用函数的docstring到system.methodHelp(func_name)方法             # 注册可调用函数的签名到system.methodSignature(func_name)方法             server.register_introspection_functions()              # 这个函数的作用是可以使客户端同时调用服务端的的多个函数。             server.register_multicall_functions()              # 注册一个函数,使它可以被调用,后面的字符串就是被调用的函数名             server.register_function(md5_func, md5_func.__name__)             try:                 log.info("server start", msg=f"xmlrpc start @ {HOST}:{PORT}!")                 server.serve_forever()             except Exception:                 log.info("server error", exc_info=True, stack_info=True)                 raise             finally:                 log.info("server stoped", msg=f"xmlrpc @ {HOST}:{PORT} stoped!")   if __name__ == "__main__":     main()

]

ZhouSa.com