開源框架中的責(zé)任鏈模式實踐
作者：Wang Zhi
責(zé)任鏈模式作為常用的設(shè)計模式而被大家熟知和使用。本文介紹責(zé)任鏈的常見實現(xiàn)方式，并結(jié)合開源框架如Dubbo、Sentinel等進(jìn)行延伸探討。
一、責(zé)任鏈介紹
在GoF 的《設(shè)計模式》一書中對責(zé)任鏈模定義的：將請求的發(fā)送和接收解耦，讓多個接收對象都有機(jī)會處理這個請求。將這些接收對象串成一條鏈，并沿著這條鏈傳遞這個請求，直到鏈上的某個接收對象能夠處理它為止或者所有接收對象處理一遍。
用通俗的話解釋在責(zé)任鏈模式中，多個處理器（接收對象）依次處理同一個請求。一個請求先經(jīng)過 A 處理器處理，然后再把請求傳遞給 B 處理器，B 處理器處理完后再傳遞給 C 處理器，以此類推，形成一個鏈條。鏈條上的每個處理器各自承擔(dān)各自的處理職責(zé)，所以叫作責(zé)任鏈模式。
責(zé)任鏈模式有效地降低了發(fā)送和接收者之間的耦合度，增強(qiáng)了系統(tǒng)的可擴(kuò)展性。在責(zé)任鏈的模式下不僅能夠針對單個處理器對象進(jìn)行定制升級（每個處理器對象關(guān)注各自的任務(wù)），而且能夠?qū)φ麄€責(zé)任鏈的處理器對象的順序的調(diào)整以及增刪。
本文約定：責(zé)任鏈上的接收對象統(tǒng)一稱為處理器；本文中介紹的責(zé)任鏈屬于GOF定義中責(zé)任鏈的變種即責(zé)任鏈上的所有處理器都會參與任務(wù)的處理。
二、責(zé)任鏈實現(xiàn)
責(zé)任鏈模式有多種實現(xiàn)方式，從驅(qū)動責(zé)任鏈上處理器方式的角度可以分類兩類，即責(zé)任鏈驅(qū)動 和 責(zé)任鏈處理器自驅(qū)動。
2.1 處理器自驅(qū)動
// 1、定義抽象類
public abstract class AbstractHandler {
protected Handler next = null;
// 綁定處理器
public void setSuccessor(Handler next) {
this.next = next;
}
// 處理器執(zhí)行操作并驅(qū)動下一個處理器
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、構(gòu)建責(zé)任鏈并添加處理器
public class HandlerChain {
// 通過鏈表的形式保存責(zé)任鏈
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、整體構(gòu)建責(zé)任鏈添加處理器并進(jìn)行驅(qū)動
public class Application {
public static void main(String[] args) {
// 構(gòu)建責(zé)任鏈并添加處理器
HandlerChain chain = new HandlerChain();
chain.addHandler(new HandlerA());
chain.addHandler(new HandlerB());
// 責(zé)任鏈負(fù)責(zé)觸發(fā)
chain.handle();
}
}說明：
- 責(zé)任鏈上的每個處理器對象維護(hù)下一個處理器對象，整個責(zé)任鏈的驅(qū)動由每個處理器對象自行驅(qū)動。
- 每個處理器對象Handler中包含下一個處理器對象next的變量，通過鏈表形式維護(hù)責(zé)任鏈的關(guān)系。
2.2 責(zé)任鏈驅(qū)動
// 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、構(gòu)建責(zé)任鏈并添加處理器
public class HandlerChain {
// 通過數(shù)組的形式保存處理器
private List<IHandler> handlers = new ArrayList<>();
public void addHandler(IHandler handler) {
handlers.add(handler);
}
// 由責(zé)任鏈負(fù)責(zé)遍歷所有的處理器并進(jìn)行調(diào)用
public void handle() {
for (IHandler handler : handlers) {
handler.handle();
}
}
}
// 5、整體構(gòu)建責(zé)任鏈添加處理器并進(jìn)行驅(qū)動
public class Application {
public static void main(String[] args) {
HandlerChain chain = new HandlerChain();
chain.addHandler(new HandlerA());
chain.addHandler(new HandlerB());
chain.handle();
}
}說明：
- 責(zé)任鏈對象本身以數(shù)組的形式維護(hù)處理器對象，即上述代碼中的handlers 。
- 責(zé)任鏈的處理器的執(zhí)行由責(zé)任鏈對象循環(huán)調(diào)用處理器對象驅(qū)動，即上述代碼中的handle方法。
三、開源框架中責(zé)任鏈應(yīng)用
責(zé)任鏈低耦合高擴(kuò)展的特點讓它在很多開源的框架中被采用，本文選取了開源框架中的Spring Interceptor、Servlet Filter、Dubbo、Sentinel進(jìn)行責(zé)任鏈的實現(xiàn)介紹，通過對常用框架中責(zé)任鏈應(yīng)用的了解能夠更好掌握責(zé)任鏈落地并在日常的開發(fā)中積極的使用。
3.1 Spring Interceptor
3.1.1 Interceptor介紹
圖片
- Spring中的攔截器(Interceptor) 用于攔截控制器方法的執(zhí)行，可以在方法執(zhí)行前后添加自定義邏輯類似于AOP編程思想。
- Inteceptor的作用時機(jī)是在請求（request）進(jìn)入servlet后，在進(jìn)入Controller之前進(jìn)行預(yù)處理。
- Inteceptor的實際應(yīng)用包括：認(rèn)證授權(quán)、日志記錄、字符編碼轉(zhuǎn)換，敏感詞過濾等等。
- Inteceptor中責(zé)任鏈的實現(xiàn)會從處理器的介紹，責(zé)任鏈的構(gòu)建以及責(zé)任鏈的執(zhí)行三個角度進(jìn)行闡述。
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");
}
}說明：
- 處理器Interceptor的接口HandlerInterceptor定義了三個方法，可在控制器方法執(zhí)行前后添加自定義邏輯。
- 自定義處理器如上的TimeInterceptor需要自定義實現(xiàn)上述3個方法實現(xiàn)自我的邏輯。
- 所有的自定義處理會串聯(lián)在HandlerExecutionChain類實現(xiàn)的責(zé)任鏈上。
3.1.3 責(zé)任鏈構(gòu)建
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;
}
}說明：
- HandlerExecutionChain類作為串聯(lián)Interceptor處理器的責(zé)任鏈負(fù)責(zé)責(zé)任鏈的構(gòu)建和執(zhí)行。
- HandlerExecutionChain類通過集合對象interceptorList保存所有相關(guān)的處理器對象。
3.1.4 責(zé)任鏈執(zhí)行
public class DispatcherServlet extends FrameworkServlet {
protected void doDispatch(HttpServletRequest request, HttpServletResponse response) throws Exception {
try {
try {
// mappedHandler代表的是HandlerExecutionChain責(zé)任鏈 mappedHandler = getHandler(processedRequest);
HandlerAdapter ha = getHandlerAdapter(mappedHandler.getHandler());
// 1、執(zhí)行mappedHandler的applyPreHandle方法
if (!mappedHandler.applyPreHandle(processedRequest, response)) {
return;
}
// 2、執(zhí)行controller的執(zhí)行邏輯
mv = ha.handle(processedRequest, response, mappedHandler.getHandler());
if (asyncManager.isConcurrentHandlingStarted()) {
return;
}
applyDefaultViewName(processedRequest, mv);
// 執(zhí)行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)) {
// 責(zé)任鏈從前往后的順序執(zhí)行
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)) {
// 責(zé)任鏈從后往前的順序執(zhí)行
for (int i = interceptors.length - 1; i >= 0; i--) {
HandlerInterceptor interceptor = interceptors[i];
interceptor.postHandle(request, response, this.handler, mv);
}
}
}
}說明：
- 在servlet的doDispatch方法中依次觸發(fā)責(zé)任鏈的applyPreHandle的前置處理方法、applyPostHandle的后置處理方法。
- 前置處理方法applyPreHandle會遍歷責(zé)任鏈上的處理器從前往后依次處理，后置處理方法applyPostHandle會遍歷責(zé)任鏈上的處理器從后往前依次處理。
- 處理器的驅(qū)動由責(zé)任鏈對象負(fù)責(zé)依次觸發(fā)，非處理器對象自驅(qū)執(zhí)行。
3.2 Servlet Filter
3.2.1 Filter介紹
- Servlet過濾器是在Java Servlet規(guī)范2.3中定義的，它能夠?qū)ervlet容器的請求和響應(yīng)對象進(jìn)行檢查和修改，是個典型的責(zé)任鏈。
- 在Servlet被調(diào)用之前檢查Request對象并支持修改Request Header和Request內(nèi)容。
- 在Servlet被調(diào)用之后檢查Response對象并支修改Response Header和Response內(nèi)容。
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、執(zhí)行處理的邏輯
System.out.println("time filter doFilter");
// 2、執(zhí)行責(zé)任鏈當(dāng)中的下一個 Filter 對象，等價于執(zhí)行 FilterChain 的internalDoFilter方法
filterChain.doFilter(servletRequest, servletResponse);
}
}說明：
- Servlet過濾器類要實現(xiàn)javax.servlet.Filter接口，該接口定義了通用的3個方法。
- init方法：負(fù)責(zé)Servlet過濾器的初始化方法，Servlet容器創(chuàng)建Servlet過濾器實例過程中調(diào)用這個方法。
- doFilter方法：當(dāng)客戶請求訪問與過濾器關(guān)聯(lián)的URL時，Servlet容器會調(diào)用該方法。
- destroy方法：Servlet容器在銷毀過濾器實例前調(diào)用該方法，可以釋放過濾器占用的資源。
3.2.3 責(zé)任鏈構(gòu)建
public final class ApplicationFilterChain implements FilterChain {
// 責(zé)任鏈上 Filter 的維護(hù)對象
private ApplicationFilterConfig[] filters = new ApplicationFilterConfig[0];
//責(zé)任鏈上待執(zhí)行的 Filter 對象
private int pos = 0;
// 責(zé)任鏈上擁有的 Filter 數(shù)量
private int n = 0;
void addFilter(ApplicationFilterConfig filterConfig) {
// 避免重復(fù)添加Filter
for(ApplicationFilterConfig filter:filters)
if(filter==filterConfig)
return;
// 按需進(jìn)行擴(kuò)容
if (n == filters.length) {
ApplicationFilterConfig[] newFilters =
new ApplicationFilterConfig[n + INCREMENT];
System.arraycopy(filters, 0, newFilters, 0, n);
filters = newFilters;
}
// 保存Filter 對象
filters[n++] = filterConfig;
}
}說明：
- ApplicationFilterChain作為Filter的責(zé)任鏈，負(fù)責(zé)責(zé)任鏈的構(gòu)建和執(zhí)行。
- 責(zé)任鏈通過ApplicationFilterConfig類型的數(shù)組對象filters保存Filter處理器。
- 責(zé)任鏈上處理器的添加通過保存到數(shù)組filters來實現(xiàn)。
3.2.4 責(zé)任鏈執(zhí)行
public final class ApplicationFilterChain implements FilterChain {
// 責(zé)任鏈上 Filter 的維護(hù)對象
private ApplicationFilterConfig[] filters = new ApplicationFilterConfig[0];
//責(zé)任鏈上待執(zhí)行的 Filter 對象
private int pos = 0;
// 責(zé)任鏈上擁有的 Filter 數(shù)量
private int n = 0;
// 責(zé)任鏈的執(zhí)行
private void internalDoFilter(ServletRequest request, ServletResponse response)
throws IOException, ServletException {
// 在責(zé)任鏈未執(zhí)行完的情況下執(zhí)行責(zé)任鏈 if (pos < n) {
// 獲取當(dāng)前待執(zhí)行的 Filter，同時遞增下一次待執(zhí)行責(zé)任鏈的下標(biāo)
ApplicationFilterConfig filterConfig = filters[pos++];
try {
Filter filter = filterConfig.getFilter();
if( Globals.IS_SECURITY_ENABLED ) {
// 省略相關(guān)代碼
} else {
filter.doFilter(request, response, this);
}
} catch (Throwable e) {
}
return;
}
try {
if ((request instanceof HttpServletRequest) &&
(response instanceof HttpServletResponse) &&
Globals.IS_SECURITY_ENABLED ) {
// 執(zhí)行正常的業(yè)務(wù)邏輯
} else {
servlet.service(request, response);
}
} catch (Throwable e) {
e = ExceptionUtils.unwrapInvocationTargetException(e);
throw new ServletException(sm.getString("filterChain.servlet"), e);
}
}
}說明：
- 整個責(zé)任鏈上Filter處理器的執(zhí)行通過處理器自驅(qū)進(jìn)行實現(xiàn)，而非由責(zé)任鏈對象驅(qū)動。
- Filter處理器的在處理過程中除了執(zhí)行自我邏輯，會通過filterChain.doFilter
(servletRequest, servletResponse)觸發(fā)下一個處理器的執(zhí)行。
3.3 Dubbo
3.3.1 Dubbo Filter介紹
- Dubbo的Filter作用時機(jī)如上圖所示，F(xiàn)ilter實現(xiàn)是專門為服務(wù)提供方和服務(wù)消費(fèi)方調(diào)用過程進(jìn)行攔截，Dubbo本身的大多功能均基于此擴(kuò)展點實現(xiàn)，每次遠(yuǎn)程方法執(zhí)行該攔截都會被執(zhí)行。
- Dubbo官方針對Filter做了很多的原生支持，目前大致有20來個吧，包括我們熟知的RpcContext，accesslog功能都是通過filter來實現(xiàn)了。
- 在實際業(yè)務(wù)開發(fā)中會對Filter接口進(jìn)行擴(kuò)展，在服務(wù)調(diào)用鏈路中嵌入我們自身的處理邏輯，如日志打印、調(diào)用耗時統(tǒng)計等。
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) {
}
// 執(zhí)行下一個invoker
return invoker.invoke(inv);
}
}說明：
- Dubbo中的自定義Filter需要實現(xiàn)org.apache.dubbo.rpc.Filter類，內(nèi)部通過實現(xiàn)invoke方法來實現(xiàn)自定義邏輯。
- 自定義Filter內(nèi)部除了實現(xiàn)必要的自定義邏輯外，核心的需要通過invoker.invoke(inv)觸發(fā)下一個過濾器的執(zhí)行。
3.3.3 責(zé)任鏈構(gòu)建
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 對象，構(gòu)建責(zé)任鏈 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進(jìn)行串聯(lián)
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;
}
}說明：
- ProtocolFilterWrapper通過
buildInvokerChain構(gòu)建Dubbo Filter的責(zé)任鏈。 - 責(zé)任鏈上的處理器對象是將Filter封裝的Invoker對象，每個Invoker對象指向下一個處理器封裝的Invoker對象。
3.3.4 責(zé)任鏈執(zhí)行
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 {
// 執(zhí)行封裝了Filter的invoker對象，驅(qū)動處理器的執(zhí)行
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;
}說明：
- 每個Invoker對象invoke方法會執(zhí)行自定義邏輯，并觸發(fā)下一個處理器的執(zhí)行。
- 整個責(zé)任鏈上處理器的執(zhí)行通過Invoker對象的驅(qū)動，而非責(zé)任鏈對象的驅(qū)動。
3.4 Sentinel
3.4.1 Sentinel Slot介紹
- Sentinel是面向分布式服務(wù)架構(gòu)的流量治理組件，以流量為切入點提供熔斷限流的功能保證系統(tǒng)的穩(wěn)定性。
- Sentinel 里面以Entry作為限流的資源對象，每個Entry創(chuàng)建的同時會關(guān)聯(lián)一系列功能插槽（slot chain）。
- Sentinel提供了通用的原生Slot處理不同的邏輯，同時支持自定義Slot來定制功能。
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 {
// 觸發(fā)下一個處理器對象的處理
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;
// 執(zhí)行具體處理器的邏輯，由具體的處理器自行實現(xiàn)
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、處理器驅(qū)動觸發(fā)下一個處理器
fireEntry(context, resourceWrapper, node, count, prioritized, args);
}
}說明：
- Sentinel中的Slot需要實現(xiàn)
com.alibaba.csp.sentinel.slotchain.ProcessorSlot的通用接口。 - 自定義Slot一般繼承抽象類AbstractLinkedProcessorSlot且只要改寫entry/exit方法實現(xiàn)自定義邏輯。
- Slot通過next變量保存下一個處理器Slot對象。
- 在自定義實現(xiàn)的entry方法中需要通過fireEntry觸發(fā)下一個處理器的執(zhí)行，在exit方法中通過fireExit觸發(fā)下一個處理器的執(zhí)行。
3.4.3 責(zé)任鏈構(gòu)建
public class DefaultSlotChainBuilder implements SlotChainBuilder {
@Override
public ProcessorSlotChain build() {
// 責(zé)任鏈的頭部對象ProcessorSlotChain
ProcessorSlotChain chain = new DefaultProcessorSlotChain();
// sortedSlotList獲取所有的處理器對象
List<ProcessorSlot> sortedSlotList = SpiLoader.of(ProcessorSlot.class).loadInstanceListSorted();
for (ProcessorSlot slot : sortedSlotList) {
if (!(slot instanceof AbstractLinkedProcessorSlot)) {
continue;
}
// 通過尾添法將職責(zé)slot添加到DefaultProcessorSlotChain當(dāng)中
chain.addLast((AbstractLinkedProcessorSlot<?>) slot);
}
return chain;
}
}
public class DefaultProcessorSlotChain extends ProcessorSlotChain {
// 創(chuàng)建DefaultProcessorSlotChain的頭尾節(jié)點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;
}
}說明：
- ProcessorSlotChain作為Slot的責(zé)任鏈，負(fù)責(zé)責(zé)任鏈的構(gòu)建和執(zhí)行。
- 責(zé)任鏈上的處理器對象
AbstractLinkedProcessorSlot通過保存指向下一個處理器的對象的進(jìn)行關(guān)聯(lián)，整體以鏈表的形式進(jìn)行串聯(lián)。 - 責(zé)任鏈上的第一個處理器對象first本身不起任何作用，只是保存鏈表的頭部。
3.4.4 責(zé)任鏈執(zhí)行
public class CtSph implements Sph {
private Entry entryWithPriority(ResourceWrapper resourceWrapper, int count, boolean prioritized, Object... args)
throws BlockException {
Context context = ContextUtil.getContext();
// 省略相關(guān)代碼
ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
Entry e = new CtEntry(resourceWrapper, chain, context);
// 驅(qū)動責(zé)任鏈上的第一個處理器，進(jìn)而由處理器自驅(qū)動執(zhí)行下一個處理器
chain.entry(context, resourceWrapper, null, count, prioritized, args);
return e;
}
}
public class DefaultProcessorSlotChain extends ProcessorSlotChain {
// 創(chuàng)建DefaultProcessorSlotChain的頭尾節(jié)點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 {
// 觸發(fā)下一個處理器對象的處理
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;
// 執(zhí)行具體處理器的邏輯，由具體的處理器自行實現(xiàn)
entry(context, resourceWrapper, t, count, prioritized, args);
}
public void setNext(AbstractLinkedProcessorSlot<?> next) {
// 綁定下一個處理器的邏輯
this.next = next;
}
}說明：
- 整個責(zé)任鏈上處理器的執(zhí)行通過Invoker對象的驅(qū)動，而非責(zé)任鏈對象的驅(qū)動。
- DefaultProcessorSlotChain的entry首先頭部對象first，進(jìn)而觸發(fā)處理器的自驅(qū)實現(xiàn)處理器的執(zhí)行。
- 整體按照entry →fireEntry →
transformEntry→ entry的循環(huán)順序依次觸發(fā)處理器的自驅(qū)。
四、實踐總結(jié)
在日常項目實踐中，責(zé)任鏈的設(shè)計模式會在很多業(yè)務(wù)場景中落地。
譬如對于支持用戶生成內(nèi)容（UGC）的應(yīng)用來說，用戶生成的內(nèi)容可能包含一些敏感內(nèi)容如敏感言論或者圖片等。針對這種應(yīng)用場景，可以通過責(zé)任鏈模式設(shè)置多個處理器來處理不同的任務(wù)，如文本過濾器處理敏感詞，圖片過濾器處理敏感圖片等等。
譬如對于電商服務(wù)中的下單流程來說，一個下單流程包含訂單拆合單，優(yōu)惠計算，訂單生成等多個步驟，我們可以通過責(zé)任鏈模式設(shè)置多個處理器來處理不同的任務(wù)等等。
責(zé)任鏈的應(yīng)用場景非常廣泛，在常見的開源框架中有豐富的落地場景，同樣在業(yè)務(wù)開發(fā)中也可以根據(jù)場景靈活使用。
責(zé)任編輯：龐桂玉
來源：
vivo互聯(lián)網(wǎng)技術(shù)
