美文网首页
巧用策略模式-逃离苦力模式

巧用策略模式-逃离苦力模式

作者: markfork | 来源:发表于2020-04-06 19:18 被阅读0次
    • 背景
    • 解决方案
      • 解决方案-1
        • 出现的问题
      • 解决方案-2
        • 如何解决上述问题
        • 策略模式伪代码快速理解(只需2分钟)
        • 优化后的代码结构
    • 解决效果

    背景

    最近的主要工作是包装AI 算法,使之成为算法服务集群。说白了就是包装若干算法能力,提供远程调用接口,供各个调用方来调用。算法主要是媒体资源的处理,包括打标签、媒体资源质量提升(分辨率提升、画面质量提升)
    算法模块比较多,大约20个左右。

    解决方案:

    解决方案-1

    因项目比较紧急,所以起初通过搬砖模式完成了3个模块的落地,落地也没什么架构可言,就像下面这样:


    image.png

    (PS:因项目比较紧急,所以前期并没有做细致的容量性能评估,只能后续优化了,当前情况下先上能力比什么都重要,如果是比较大的项目,或者要求服务稳定性、性能高的服务,必须要做容量性能评估,才好确认什么是好的架构,而且这个细致的过程需要经过技术团队的评审,并且需要预留充足的评审时间,否则加班是难免的)

    详细解释:

    • 绿色框代表客户端请求,其实就是内部主调方,调用模式分为存量调用 & 在线调用,具体调用量暂时不详。(优化阶段需要详细了解一下)
    • 蓝色虚线框中是部署的若干服务模块,m_srv_x代表不同的服务能力。
      工作模式大致是:客户端请求-> 请求参数处理->下载待分析资源->m_srv处理->吐出结果->回调通知调用方处理结果

    带来的问题:

    在包装了3个模块之后,个人发现:
    不考虑服务架构上面的问题

    1. 个人陷入苦力模式、落地模块重复操作太多,带来的问题就是,重复代码量迅速扩张;
    2. 每落地一个模块就需要包装1个接口,接口量迅速提升,而且需要同步维护对接的api 文档,多的是copy,大大影响工作效率。

    在紧急上线了3个模块之后,我对上述无脑式的垒代码的工作方式产生了反思。如何解决上面2个问题,是我接下来最需要优先解决的。否则,大概率陷入到苦力模式,陷入泥潭。

    解决方案-2

    如何解决上述问题

    总结出了上述2个问题,接下来的事情就是一一分析、找解决方案了:

    1. 苦力模式- 对于每个模块重复的代码,暂且称之为前置处理(抽象共性)吧,每个模块的前置处理大致都相同,可以将相同代码提取、封装为common_util,即复用
    2. 接口量过多问题-只写一个接口,当前接口充当阉割式的API 网管,做简单参数校验、请求路由(PS:负载均衡、服务降级、熔断、限流、监控是API网关需要着重考虑的点,当前场景不需要考虑上述问题,撸一个API网关,工作量很大,不值当),这样,主调方接入也非常方便,API接口文档 ,只需要按需做少量更新,进而解放生产力。
    3. 2 中提到了请求路由,那么如何实现-解决办法是客户端请求时携带 http://xxx.com/api?module_name=$module_name ,根据module_name 将请求路由到具体模块。
    4. 2 中请求路由问题解决后,又有新问题,请求路由的具体实现方式是什么呢? 难道要写若干if... else来实现,实现起来也没问题,但是会带来如下问题:
      1. 如果某个模块代码写的有问题,那么其他模块也不能成功运行。
      1. 如果要下线、上线新模块,那么代码就需要重新更新,添加新的if...else代码块
        4中的2个问题有没有解决方案呢?有的,那就是通过策略设计模式来实现。

    策略模式伪代码快速理解(只需2分钟)

    1. 定义策略模式契约 -AbstractHandler

    class AbstractHandler:
         def __init__(self, next_handler=None):
              self.next_handler = next_handler
        
        def _judge(self, request):
             pass
        
       def _handler(self):
            # 处理请求,吐出结果
            pass
    
         def handler(self, request):
              if judgement(request):
                self._handler(request):
              else:
                if self.next:
                   return self.next.handler(request)
             return None
    

    2. 定义具体策略执行类-MSrv1Handler

    class Msr1Handler(AbstractHandler):
        def __init__(self):
             super().__init__()
      
       def _judge(self, request):
            return True if request > 10 else False
               
       def _handler(self):
           print('Msr1Handler handled this request')
    

    3. 定义具体策略执行类-MSrv2Handler

    class Msr2Handler(AbstractHandler):
        def __init__(self):
             super().__init__()
      
       def _judge(self, request):
            return True if request < 10 else False
               
       def _handler(self):
           print('Msr2Handler handled this request')
    

    ...
    简介下上述3个文件充当的角色:

    • AbstractHandler-策略契约定义者,其中定义了处理过程,如果当前遵守契约的处理单元不能处理此请求,则交由下一个处理单元去处理。
    • Msr1Handler & Msr2Handler- 遵守(extends)策略契约的处理单元,其中只定义了_judge() 判定自己可处理请求、和 _handler() 处理请求的核心逻辑,但是其具有 handel() 方法,因为通过 class Msr2Handler(AbstractHandler):继承而来。
    • AbstractHandler next 属性 - next把遵守契约的处理单元串成链表,这样请求就可以在这条链上传递,直至被某个处理单元处理。

    优化后的代码结构

    分层结构

    代码分层

    简单解释下:

    common: 封装了常用操作,如下载资源、生成输出文件存储目录、下载文件重命名等功能‘
    exception: 自定义的业务逻辑异常
    handler_engine: Ai 模块处理引擎仓库,可实现平铺式新增能力。
    module_handler_list: 模块引擎初始化入口
    workflow: 一个统一化接口,接收请求,并路由请求到处理引擎。
    

    抽象引擎-策略契约定义者 AbstractModuleHandler

    #!/usr/bin/env python
    # -*- coding: utf-8 -*-
    """
    # @Time    : 2020/4/4 下午8:32
    # @Author  : 
    # @Site    : 
    # @File    : handler.py
    # @Software: PyCharm
    # 抽象处理逻辑为职责链,通过外部请求参数控制链节点进行请求处理
      eg: xx?module_name=A -> moduleA.handler()
    """
    import json
    import functools
    
    print = functools.partial(print, flush=True)
    from concurrent.futures import ThreadPoolExecutor
    
    from config.ai_module_config import all_modules
    from common.common_handler_util import *
    from exception.async_handler_exception import AsyncHandlerException
    
    EXECUTOR = ThreadPoolExecutor(8)
    
    
    class AbstractRequestEntity():
        """请求实例化为entity"""
    
    
    class AbstractModuleHandler:
        """
        抽象模型处理器
        """
    
        def __init__(self):
            self.fill_attr()
            self.next_module_handler = None
    
        def _fill_attr(self, attr_tuple):
            """
            动态填充属性 辅助函数
            :param attr_tuple: (key,val) = self.$key = $val
            :return:
            """
    
            if not isinstance(attr_tuple, tuple):
                return False, 'dynamic attr not match format,please give attr like (key,val)'
    
            attr_name, attr_val = attr_tuple
            if getattr(self, attr_name, None):
                return
            setattr(self, attr_name, attr_val)
    
        def fill_attr(self):
            current_module = all_modules.get(self._module_name, {})
            for module in current_module.items():
                self._fill_attr(module)
            return
    
        def _check_request_param(self, request):
            if self._module_name != request.get('module_name', ''):
                ''' judge request module name '''
                return False, 'this module not the correct module handler'
            for param in request.items():
                param_name = param[0]
                request_params = getattr(self, 'request_params', None)
                if not request_params:
                    return False, 'this module don\'t need request params!'
                if param_name not in request_params:
                    return False, 'this request is not legitimate'
            return True, 'check request params success'
    
        def check_request_params(self, request):
            if not request:
                return False, 'request params not be none!', ''
            if not isinstance(request, dict):
                return False, 'request params\'s type must be map', ''
            return self._check_request_param(request), ''
    
        def _async_prev_handler(self, request):
            """
            前置处理器
            :param request:
            :return:
            """
            local_file_path = ''
            output_root_dir = ''
    
            source_url = request.get('source_url', {})
            download_flag = getattr(self, 'download_flag')
            store_local_flag = getattr(self, 'store_local_flag')
            allow_media_type = getattr(self, 'allow_media_type')
            source_type = source_url.split('.')[-1]
            origin_file_name = source_url.split('/')[-1]
    
            # 验证资源
            if source_type not in allow_media_type:
                raise AsyncHandlerException(res_code=-1, res_msg='source type not allow')
            # 下载资源
            if download_flag:
                try:
                    print(source_url)
                    ret, msg, local_file_path, gen_date = download_remote_source(remote_uri=source_url,
                                                                                 file_name=origin_file_name,
                                                                                 module_name=self._module_name)
                except Exception as err:
                    raise AsyncHandlerException(res_code=-2,
                                                res_msg='download remote source has occurred failed,detail=%s' % str(err))
            # 创建输出文件夹
            if download_flag and store_local_flag:
                output_root_dir = gen_store_root_dir(gen_date, 'output', self._module_name)
    
            return local_file_path, output_root_dir, request
    
        def _async_core_ability(self, *args):
            """
            核心能力
            :return:
            """
    
        def _async_post_handler(self, *args):
            """
            核心处理器完成之后,进行后续处理 如 upload cdn
            :return:
            """
    
        def _async_after_completion(self, *args):
            """post handler 完成后,进行回调通知 or 资源销毁等动作"""
            print(args)
            ret_data, request = args
            ret_data['data']['extra'] = request.get('extra')
            call_back_url = request.get('call_back_url')
            payload = "{\"bussiness_data\": %s\n}" % json.dumps(ret_data)
            headers = {
                'content-type': "application/json"
            }
            response = requests.request("POST", call_back_url, data=payload, headers=headers)
            print(response.text)
    
        def async_handler(self, request):
            try:
                # 异步前置处理
                args = self._async_prev_handler(request)
                # 异步核心能力调用
                args = self._async_core_ability(args)
                # 异步后置处理,如上传cdn
                ret_data = self._async_post_handler(args)
            except Exception as err:
                if isinstance(err, AsyncHandlerException):
                    ret_data = err.res_data if err.res_data else {"res_code": err.res_code, "err_msg": err.res_msg}
                else:
                    ret_data = {"res_code": -10,
                                "err_msg": "server has occurred error,detail=%s,please call bofengliu" % str(err)}
            finally:
                # 异步清理资源,回调处理,没想好怎么写
                self._async_after_completion(ret_data, request)
    
        def handle(self, request):
            """
            核心处理器
            :param request:
            :return:
            """
            ret, msg = self._check_request_param(request)
            if not ret:
                if self.next_module_handler:
                    return self.next_module_handler.handle(request)
                return json.dumps(
                    {"res_code": -6, "err_msg": "request params not match %s required params" % self._module_name})
            # 执行之前先返回,确认接受参数的响应
            EXECUTOR.submit(self.async_handler, request)
            return json.dumps({"res_code": 0, "err_msg": "received %s task success, running now!" % self._module_name})
    
    

    最重要的是 handler() & async_handler()方法,在这个类中,我们定义了处理请求的核心接口,其他具体实现这个抽象契约类的子类,只需要关注 _async_core_ability()_async_post_handler() 后置处理了。

    统一化阉割版API 网关:

    @app.route('/module/v1/api', methods=['POST'])
    def dispatcher_handler():
        """
        统一网管
        :return:
        """
        if 'POST' != request.method:
            return jsonify({
                "res_code": 1,
                "err_msg": "request method must be POST"
            })
    
        request_params = request.get_json()
        return all_module_handlers.handle(request_params)
    

    优化后可以预想到的提升效果

    采用策略模式后,如果新上线一个模型,我们只需要新建一个继承策略类的module类,且实现 _async_core_ability()_async_post_handler() 就可以了,接口完全不需要改变。工作效率提升 70% 没有问题,至此就已经逃离了苦力模式,美滋滋。

    发现这里面其实还应用到了模板方法模式,其实就是 AbstractHandler 类;

    相关文章

      网友评论

          本文标题:巧用策略模式-逃离苦力模式

          本文链接:https://www.haomeiwen.com/subject/inybphtx.html