vivo 互联网技术

开源框架中的责任链模式实践

2023-09-20  本文已影响0人  vivo互联网技术

作者：vivo 互联网服务器团队-Wang Zhi

责任链模式作为常用的设计模式而被大家熟知和使用。本文介绍责任链的常见实现方式，并结合开源框架如Dubbo、Sentinel等进行延伸探讨。

一、责任链介绍

在GoF 的《设计模式》一书中对责任链模定义的：将请求的发送和接收解耦，让多个接收对象都有机会处理这个请求。将这些接收对象串成一条链，并沿着这条链传递这个请求，直到链上的某个接收对象能够处理它为止或者所有接收对象处理一遍。

用通俗的话解释在责任链模式中，多个处理器（接收对象）依次处理同一个请求。一个请求先经过 A 处理器处理，然后再把请求传递给 B 处理器，B 处理器处理完后再传递给 C 处理器，以此类推，形成一个链条。链条上的每个处理器各自承担各自的处理职责，所以叫作责任链模式。

责任链模式有效地降低了发送和接收者之间的耦合度，增强了系统的可扩展性。在责任链的模式下不仅能够针对单个处理器对象进行定制升级（每个处理器对象关注各自的任务），而且能够对整个责任链的处理器对象的顺序的调整以及增删。

本文约定：责任链上的接收对象统一称为处理器；本文中介绍的责任链属于GOF定义中责任链的变种即责任链上的所有处理器都会参与任务的处理。

二、责任链实现

责任链模式有多种实现方式，从驱动责任链上处理器方式的角度可以分类两类，即责任链驱动 和 责任链处理器自驱动。

2.1 处理器自驱动

// 1、定义抽象类
public abstract class AbstractHandler {
  protected Handler next = null;
 
  // 绑定处理器
  public void setSuccessor(Handler next) {
    this.next = next;
  }
   
  // 处理器执行操作并驱动下一个处理器
  public abstract void handle();
}
 
// 2、定义处理器A
public class HandlerA extends AbstractHandler {
  @Override
  public void handle() {
    // do something
 
    if (next != null) {
      next.handle();
    }
  }
}
 
// 3、定义处理器B
public class HandlerB extends AbstractHandler {
  @Override
  public void handle() {
    // do something
     
    if (next != null) {
      next.handle();
    }
  }
}
 
// 4、构建责任链并添加处理器
public class HandlerChain {
  // 通过链表的形式保存责任链
  private AbstractHandler head = null;
  private AbstractHandler tail = null;
 
  public void addHandler(AbstractHandler handler) {
    handler.setSuccessor(null);
 
    if (head == null) {
      head = handler;
      tail = handler;
      return;
    }
 
    tail.setSuccessor(handler);
    tail = handler;
  }
 
  public void handle() {
    if (head != null) {
      head.handle();
    }
  }
}
 
// 5、整体构建责任链添加处理器并进行驱动
public class Application {
  public static void main(String[] args) {
    // 构建责任链并添加处理器
    HandlerChain chain = new HandlerChain();
    chain.addHandler(new HandlerA());
    chain.addHandler(new HandlerB());
 
    // 责任链负责触发
    chain.handle();
  }
}

说明：

2.2 责任链驱动

// 1、定义抽象接口
public interface IHandler {
  void doSomething();
}
 
// 2、定义处理器A
public class HandlerA implements IHandler {
  @Override
  public void doSomething() {
    // do something
  }
}
 
// 3、定义处理器B
public class HandlerB implements IHandler {
  @Override
  public void doSomething() {
    // do something
  }
}
 
// 4、构建责任链并添加处理器
public class HandlerChain {
  // 通过数组的形式保存处理器
  private List<IHandler> handlers = new ArrayList<>();
 
  public void addHandler(IHandler handler) {
    handlers.add(handler);
  }
  // 由责任链负责遍历所有的处理器并进行调用
  public void handle() {
    for (IHandler handler : handlers) {
      handler.handle();
    }
  }
}
 
// 5、整体构建责任链添加处理器并进行驱动
public class Application {
  public static void main(String[] args) {
    HandlerChain chain = new HandlerChain();
    chain.addHandler(new HandlerA());
    chain.addHandler(new HandlerB());
    chain.handle();
  }
}

说明：

三、开源框架中责任链应用

责任链低耦合高扩展的特点让它在很多开源的框架中被采用，本文选取了开源框架中的Spring Interceptor、Servlet Filter、Dubbo、Sentinel进行责任链的实现介绍，通过对常用框架中责任链应用的了解能够更好掌握责任链落地并在日常的开发中积极的使用。

3.1 Spring Interceptor

3.1.1 Interceptor介绍

图1.jpg

3.1.2 处理器介绍

public interface HandlerInterceptor {
 
    boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception;
    void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception;
    void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception;
}
 
@Component
public class TimeInterceptor extends HandlerInterceptorAdapter {
 
    @Override
    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
        // 前置处理
        System.out.println("time interceptor preHandle");
 
        return true;
    }
 
    @Override
    public void postHandle(HttpServletRequest request, HttpServletResponse response, Object handler, ModelAndView modelAndView) throws Exception {
        // 后置处理
        System.out.println("time interceptor postHandle");
    }
 
    @Override
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception ex) throws Exception {
        System.out.println("time interceptor afterCompletion");
    }
}

说明：

3.1.3 责任链构建

public class HandlerExecutionChain {
 
    private final Object handler;
    private HandlerInterceptor[] interceptors;
    private List<HandlerInterceptor> interceptorList;
    private int interceptorIndex = -1;
 
    public void addInterceptor(HandlerInterceptor interceptor) {
        // 添加拦截器
        initInterceptorList().add(interceptor);
    }
 
    public void addInterceptors(HandlerInterceptor... interceptors) {
        if (!ObjectUtils.isEmpty(interceptors)) {
            CollectionUtils.mergeArrayIntoCollection(interceptors, initInterceptorList());
        }
    }
 
    private List<HandlerInterceptor> initInterceptorList() {
        if (this.interceptorList == null) {
            this.interceptorList = new ArrayList<HandlerInterceptor>();
            if (this.interceptors != null) {
                // An interceptor array specified through the constructor
                CollectionUtils.mergeArrayIntoCollection(this.interceptors, this.interceptorList);
            }
        }
        this.interceptors = null;
        return this.interceptorList;
    }
}

说明：

3.1.4 责任链执行

public class DispatcherServlet extends FrameworkServlet {
 
    protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
 
        try {
            try {
                // mappedHandler代表的是HandlerExecutionChain责任链                 mappedHandler = getHandler(processedRequest);
                HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());
 
                // 1、执行mappedHandler的applyPreHandle方法
                if (!mappedHandler.applyPreHandle(processedRequest, response)) {
                    return;
                }
 
                // 2、执行controller的执行逻辑
                mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
 
                if (asyncManager.isConcurrentHandlingStarted()) {
                    return;
                }
                applyDefaultViewName(processedRequest, mv);
 
                // 执行mappedHandler的applyPostHandle方法
                mappedHandler.applyPostHandle(processedRequest, response, mv);
            }
            catch (Exception ex) {
            }
            processDispatchResult(processedRequest, response, mappedHandler, mv, dispatchException);
        }
        catch (Exception ex) {
        }
        finally {
        }
    }
}
 
 
public class HandlerExecutionChain {
 
    private final Object handler;
    private HandlerInterceptor[] interceptors;
    private List<HandlerInterceptor> interceptorList;
    private int interceptorIndex = -1;
 
    boolean applyPreHandle(HttpServletRequest request, HttpServletResponse response) throws Exception {
        HandlerInterceptor[] interceptors = getInterceptors();
        if (!ObjectUtils.isEmpty(interceptors)) {
            // 责任链从前往后的顺序执行
            for (int i = 0; i < interceptors.length; i++) {
                HandlerInterceptor interceptor = interceptors[i];
                if (!interceptor.preHandle(request, response, this.handler)) {
                    triggerAfterCompletion(request, response, null);
                    return false;
                }
                this.interceptorIndex = i;
            }
        }
        return true;
    }
 
    void applyPostHandle(HttpServletRequest request, HttpServletResponse response, ModelAndView mv) throws Exception {
        HandlerInterceptor[] interceptors = getInterceptors();
        if (!ObjectUtils.isEmpty(interceptors)) {
            // 责任链从后往前的顺序执行
            for (int i = interceptors.length - 1; i >= 0; i--) {
                HandlerInterceptor interceptor = interceptors[i];
                interceptor.postHandle(request, response, this.handler, mv);
            }
        }
    }
}

说明：

3.2 Servlet Filter

3.2.1 Filter介绍

图2.jpg

3.2.2 处理器介绍

public interface Filter {
    public void init(FilterConfig filterConfig) throws ServletException;
    public void doFilter ( ServletRequest request, ServletResponse response, FilterChain chain ) throws IOException, ServletException;
    public void destroy();
}
 
public class TimeFilter implements Filter {
 
    @Override
    public void init(FilterConfig filterConfig) throws ServletException {
        System.out.println("time filter init");
    }
 
    @Override
    public void doFilter(ServletRequest servletRequest, ServletResponse servletResponse, FilterChain filterChain) throws IOException, ServletException {
        // 1、执行处理的逻辑
        System.out.println("time filter doFilter");
 
        // 2、执行责任链当中的下一个 Filter 对象，等价于执行 FilterChain 的internalDoFilter方法
        filterChain.doFilter(servletRequest, servletResponse);
    }
}

说明：

3.2.3 责任链构建

public final class ApplicationFilterChain implements FilterChain {
 
    // 责任链上 Filter 的维护对象
    private ApplicationFilterConfig[] filters = new ApplicationFilterConfig[0];
    //责任链上待执行的 Filter 对象
    private int pos = 0;
    // 责任链上拥有的 Filter 数量
    private int n = 0;
 
    void addFilter(ApplicationFilterConfig filterConfig) {
 
        // 避免重复添加Filter
        for(ApplicationFilterConfig filter:filters)
            if(filter==filterConfig)
                return;
        // 按需进行扩容
        if (n == filters.length) {
            ApplicationFilterConfig[] newFilters =
                new ApplicationFilterConfig[n + INCREMENT];
            System.arraycopy(filters, 0, newFilters, 0, n);
            filters = newFilters;
        }
        // 保存Filter 对象
        filters[n++] = filterConfig;
 
    }
}

说明：

3.2.4 责任链执行

public final class ApplicationFilterChain implements FilterChain {
 
    // 责任链上 Filter 的维护对象
    private ApplicationFilterConfig[] filters = new ApplicationFilterConfig[0];
    //责任链上待执行的 Filter 对象
    private int pos = 0;
    // 责任链上拥有的 Filter 数量
    private int n = 0;
 
    // 责任链的执行
    private void internalDoFilter(ServletRequest request,
                                  ServletResponse response)
        throws IOException, ServletException {
 
        // 在责任链未执行完的情况下执行责任链         if (pos < n) {
            // 获取当前待执行的 Filter，同时递增下一次待执行责任链的下标
            ApplicationFilterConfig filterConfig = filters[pos++];
            try {
                Filter filter = filterConfig.getFilter();
 
                if( Globals.IS_SECURITY_ENABLED ) {
                    // 省略相关代码
                } else {
                    filter.doFilter(request, response, this);
                }
            } catch (Throwable e) {
            }
            return;
        }
 
        try {
            if ((request instanceof HttpServletRequest) &&
                    (response instanceof HttpServletResponse) &&
                    Globals.IS_SECURITY_ENABLED ) {
                // 执行正常的业务逻辑
            } else {
                servlet.service(request, response);
            }
        } catch (Throwable e) {
            e = ExceptionUtils.unwrapInvocationTargetException(e);
            throw new ServletException(sm.getString("filterChain.servlet"), e);
        }
    }
}

说明：

3.3 Dubbo

3.3.1 Dubbo Filter介绍

图3.jpg

图片分享自《DUBBO官网》

3.3.2 处理器介绍

@Activate(group = PROVIDER, value = ACCESS_LOG_KEY)
public class AccessLogFilter implements Filter {
 
    @Override
    public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
        try {
            if (ConfigUtils.isNotEmpty(accessLogKey)) {
                AccessLogData logData = buildAccessLogData(invoker, inv);
                log(accessLogKey, logData);
            }
        } catch (Throwable t) {
        }
 
        // 执行下一个invoker
        return invoker.invoke(inv);
    }
}

说明：

3.3.3 责任链构建

public class ProtocolFilterWrapper implements Protocol {
 
    private final Protocol protocol;
 
    public ProtocolFilterWrapper(Protocol protocol) {
        this.protocol = protocol;
    }
 
    private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
 
        // 最后的 Invoker 对象
        Invoker<T> last = invoker;
 
        // 遍历所有 Filter 对象，构建责任链         List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
        if (!filters.isEmpty()) {
            for (int i = filters.size() - 1; i >= 0; i--) {
                // 每个 Filter 封装成一个 Invoker 对象，通过 filter.invoke进行串联
                final Filter filter = filters.get(i);
                final Invoker<T> next = last;
                last = new Invoker<T>() {
 
                    @Override
                    public Result invoke(Invocation invocation) throws RpcException {
                        return filter.invoke(next, invocation);
                    }
                };
            }
        }
        return last;
    }
}
 
 
// 封装了Filter的invoker对象
static final class ProtocolFilterWrapper.1 implements Invoker < T > {
    final Invoker val$invoker;
    final Filter val$filter;
    // 指向下一个Invoker的变量
    final Invoker val$next;
 
    public Result invoke(Invocation invocation) throws RpcException {
         return this.val$filter.invoke(this.val$next, invocation);
    }
 
    ProtocolFilterWrapper.1(Invoker invoker, Filter filter, Invoker invoker2) {
        this.val$invoker = invoker;
        this.val$filter = filter;
        this.val$next = invoker2;
    }
}

说明：

3.3.4 责任链执行

public class FailfastClusterInvoker<T> extends AbstractClusterInvoker<T> {
 
    public FailfastClusterInvoker(Directory<T> directory) {
        super(directory);
    }
 
    @Override
    public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        checkInvokers(invokers, invocation);
        Invoker<T> invoker = select(loadbalance, invocation, invokers, null);
        try {
            // 执行封装了Filter的invoker对象，驱动处理器的执行
            return invoker.invoke(invocation);
        } catch (Throwable e) {
             
        }
    }
}
 
static final class ProtocolFilterWrapper.1 implements Invoker < T > {
    final Invoker val$invoker;
    final Filter val$filter;
    final Invoker val$next;
 
    public Result invoke(Invocation invocation) throws RpcException {
         return this.val$filter.invoke(this.val$next, invocation);
    }
 
    ProtocolFilterWrapper.1(Invoker invoker, Filter filter, Invoker invoker2) {
        this.val$invoker = invoker;
        this.val$filter = filter;
        this.val$next = invoker2;
    }

说明：

3.4 Sentinel

3.4.1 Sentinel Slot介绍

图4.png

图片分享自《Sentinel官网》

3.4.2 处理器介绍

public interface ProcessorSlot<T> {
    void entry(Context context, ResourceWrapper resourceWrapper, T param, int count, boolean prioritized,Object... args) throws Throwable;
    void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized,Object... args) throws Throwable;
    void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args);
    void fireExit(Context context, ResourceWrapper resourceWrapper, int count, Object... args);
}
 
public abstract class AbstractLinkedProcessorSlot<T> implements ProcessorSlot<T> {
 
    private AbstractLinkedProcessorSlot<?> next = null;
 
    @Override
    public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
        throws Throwable {
        // 触发下一个处理器对象的处理
        if (next != null) {
            next.transformEntry(context, resourceWrapper, obj, count, prioritized, args);
        }
    }
 
    void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
        throws Throwable {
        T t = (T)o;
        // 执行具体处理器的逻辑，由具体的处理器自行实现
        entry(context, resourceWrapper, t, count, prioritized, args);
    }
 
    public void setNext(AbstractLinkedProcessorSlot<?> next) {
        // 绑定下一个处理器的逻辑
        this.next = next;
    }
}
 
 
public class NodeSelectorSlot extends AbstractLinkedProcessorSlot<Object> {
 
    private volatile Map<String, DefaultNode> map = new HashMap<String, DefaultNode>(10);
 
    @Override
    public void entry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
        throws Throwable {
 
        // 1、处理器处理本身的逻辑
        DefaultNode node = map.get(context.getName());
        context.setCurNode(node);
 
        // 2、处理器驱动触发下一个处理器
        fireEntry(context, resourceWrapper, node, count, prioritized, args);
    }
}

说明：

3.4.3 责任链构建

public class DefaultSlotChainBuilder implements SlotChainBuilder {
 
    @Override
    public ProcessorSlotChain build() {
        // 责任链的头部对象ProcessorSlotChain
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
 
        // sortedSlotList获取所有的处理器对象
        List<ProcessorSlot> sortedSlotList = SpiLoader.of(ProcessorSlot.class).loadInstanceListSorted();
        for (ProcessorSlot slot : sortedSlotList) {
            if (!(slot instanceof AbstractLinkedProcessorSlot)) {
                continue;
            }
            // 通过尾添法将职责slot添加到DefaultProcessorSlotChain当中
            chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
        }
 
        return chain;
    }
}
 
public class DefaultProcessorSlotChain extends ProcessorSlotChain {
 
    // 创建DefaultProcessorSlotChain的头尾节点first和end
    AbstractLinkedProcessorSlot<?> first = new AbstractLinkedProcessorSlot<Object>() {
 
        @Override
        public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
            throws Throwable {
            super.fireEntry(context, resourceWrapper, t, count, prioritized, args);
        }
 
        @Override
        public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
            super.fireExit(context, resourceWrapper, count, args);
        }
    };
    AbstractLinkedProcessorSlot<?> end = first;
 
    @Override
    public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        end.setNext(protocolProcessor);
        end = protocolProcessor;
    }
}

说明：

3.4.4 责任链执行

public class CtSph implements Sph {
 
    private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
        throws BlockException {
 
        Context context = ContextUtil.getContext();
 
        // 省略相关代码
 
        ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
        Entry e = new CtEntry(resourceWrapper, chain, context);
 
        // 驱动责任链上的第一个处理器，进而由处理器自驱动执行下一个处理器
        chain.entry(context, resourceWrapper, null, count, prioritized, args);
         
        return e;
    }
}
 
 
public class DefaultProcessorSlotChain extends ProcessorSlotChain {
 
    // 创建DefaultProcessorSlotChain的头尾节点first和end
    AbstractLinkedProcessorSlot<?> first = new AbstractLinkedProcessorSlot<Object>() {
 
        @Override
        public void entry(Context context, ResourceWrapper resourceWrapper, Object t, int count, boolean prioritized, Object... args)
            throws Throwable {
            super.fireEntry(context, resourceWrapper, t, count, prioritized, args);
        }
 
        @Override
        public void exit(Context context, ResourceWrapper resourceWrapper, int count, Object... args) {
            super.fireExit(context, resourceWrapper, count, args);
        }
    };
    AbstractLinkedProcessorSlot<?> end = first;
 
    @Override
    public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        end.setNext(protocolProcessor);
        end = protocolProcessor;
    }
}
 
public abstract class AbstractLinkedProcessorSlot<T> implements ProcessorSlot<T> {
  
    private AbstractLinkedProcessorSlot<?> next = null;
  
    @Override
    public void fireEntry(Context context, ResourceWrapper resourceWrapper, Object obj, int count, boolean prioritized, Object... args)
        throws Throwable {
        // 触发下一个处理器对象的处理
        if (next != null) {
            next.transformEntry(context, resourceWrapper, obj, count, prioritized, args);
        }
    }
  
    void transformEntry(Context context, ResourceWrapper resourceWrapper, Object o, int count, boolean prioritized, Object... args)
        throws Throwable {
        T t = (T)o;
        // 执行具体处理器的逻辑，由具体的处理器自行实现
        entry(context, resourceWrapper, t, count, prioritized, args);
    }
  
    public void setNext(AbstractLinkedProcessorSlot<?> next) {
        // 绑定下一个处理器的逻辑
        this.next = next;
    }
}

说明：

四、实践总结

在日常项目实践中，责任链的设计模式会在很多业务场景中落地。

譬如对于支持用户生成内容（UGC）的应用来说，用户生成的内容可能包含一些敏感内容如敏感言论或者图片等。针对这种应用场景，可以通过责任链模式设置多个处理器来处理不同的任务，如文本过滤器处理敏感词，图片过滤器处理敏感图片等等。

譬如对于电商服务中的下单流程来说，一个下单流程包含订单拆合单，优惠计算，订单生成等多个步骤，我们可以通过责任链模式设置多个处理器来处理不同的任务等等。

责任链的应用场景非常广泛，在常见的开源框架中有丰富的落地场景，同样在业务开发中也可以根据场景灵活使用。

上一篇下一篇

猜你喜欢

热点阅读