自定义责任链Filter实现

核心接口 Filter package com.xxx.arch.mw.nbp mon.extension; import com.xxx mons.data.domain.Result; /** * @date 2023/08/25 */ public interface Filter { Result invoke(final Invoker invoker, final Invocation invocation); } Invoker package com.xxx.arch.mw.nbp mon.extension; import com.xxx mons.data.domain.Result; /** * @date 2023/08/25 */ public interface Invoker { Result invoke(Invocation invocation); }

Invocation package com.xxx.arch.mw.nbp mon.extension; import com.xxx.arch.mw.nbp.share.dto.PropertyDTO; /** * @date 2023/08/25 */ public interface Invocation<T> { T getDetail(); PropertyDTO getProperty(); boolean isAsync(); }

关键类 FilterChain package com.xxx.arch.mw.nbp mon.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import java.util.ArrayList; import java.util.Collections; import java.util.List; /** * @date 2023/08/28 */ public class FilterChain { private final List<Filter> filters = new ArrayList<>(); public FilterChain() { } public void addFilter(Filter filter) { if (filter == null || filter.getClass().getAnnotation(DispatchFilter.class) == null) { return; } if (this.filters.stream().noneMatch(existFilter -> existFilter.getClass().equals(filter.getClass()))) { this.filters.add(filter); } } public void removeFilter(String filterName) { this.filters.removeIf(filter -> filter.getClass().getAnnotation(DispatchFilter.class).value().equals(filterName)); } public void removeFilter(Filter filter) { this.filters.removeIf(aFilter -> aFilter.getClass().equals(filter.getClass())); } public Invoker buildInvokerChain(final Invoker invoker) { Invoker last = invoker; if (!filters.isEmpty()) { this.sort(); for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker next = last; last = invocation -> filter.invoke(next, invocation); } } return last; } public void sort() { Collections.sort(this.filters, FilterComparator.COMPARATOR); } public List<Filter> getFilters() { return filters; } }

FilterComparator package com.xxx.arch.mw.nbp mon.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import java.util.Comparator; /** * @date 2023/08/28 */ public class FilterComparator implements Comparator<Object> { public static final Comparator<Object> COMPARATOR = new FilterComparator(); @Override public int compare(Object o1, Object o2) { if (o1 == null && o2 == null) { return 0; } if (o1 == null) { return -1; } if (o2 == null) { return 1; } if (o1.equals(o2)) { return 0; } DispatchFilter a1 = o1.getClass().getAnnotation(DispatchFilter.class); DispatchFilter a2 = o2.getClass().getAnnotation(DispatchFilter.class); int n1 = a1 == null ? 0 : a1.order(); int n2 = a2 == null ? 0 : a2.order(); // never return 0 even if n1 equals n2, otherwise, o1 and o2 will override each other in collection like HashSet return n1 > n2 ? 1 : -1; } }

SingleInvocation package com.xxx.arch.mw.nbp mon.extension; import com.xxx.arch.mw.nbp.share.dto.PropertyDTO; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; /** * @date 2023/08/28 */ public class SingleInvocation implements Invocation<SingleDetailDTO> { private final SingleDetailDTO singleDetailDTO; private final PropertyDTO propertyDTO; private final boolean async; public SingleInvocation(SingleDetailDTO singleDetailDTO, PropertyDTO propertyDTO) { this.singleDetailDTO = singleDetailDTO; this.propertyDTO = propertyDTO; this.async = false; } public SingleInvocation(SingleDetailDTO singleDetailDTO, PropertyDTO propertyDTO, boolean async) { this.singleDetailDTO = singleDetailDTO; this.propertyDTO = propertyDTO; this.async = async; } @Override public SingleDetailDTO getDetail() { return singleDetailDTO; } @Override public PropertyDTO getProperty() { return propertyDTO; } @Override public boolean isAsync() { return this.async; } }

MultiInvocation package com.xxx.arch.mw.nbp mon.extension; import com.xxx.arch.mw.nbp.share.dto.PropertyDTO; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; import java.util.List; /** * @date 2023/08/28 */ public class MultiInvocation implements Invocation<List<SingleDetailDTO>> { private final List<SingleDetailDTO> singleDetailDTOS; private final PropertyDTO propertyDTO; private final boolean async; public MultiInvocation(List<SingleDetailDTO> singleDetailDTOS, PropertyDTO propertyDTO) { this.singleDetailDTOS = singleDetailDTOS; this.propertyDTO = propertyDTO; this.async = false; } public MultiInvocation(List<SingleDetailDTO> singleDetailDTOS, PropertyDTO propertyDTO, boolean async) { this.singleDetailDTOS = singleDetailDTOS; this.propertyDTO = propertyDTO; this.async = async; } @Override public List<SingleDetailDTO> getDetail() { return singleDetailDTOS; } @Override public PropertyDTO getProperty() { return propertyDTO; } @Override public boolean isAsync() { return this.async; } }

自定义Filter

选取几个经典实现

ValidationFilter package com.xxx.arch.mw.nbp.client.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import com.xxx.arch.mw.nbp mon.constant.FilterConstants; import com.xxx.arch.mw.nbp mon.domain.NbpCode; import com.xxx.arch.mw.nbp mon.exception.NbpException; import com.xxx.arch.mw.nbp mon.extension.*; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; import com.xxx mons.data.domain.Result; import java.util.List; /** * @date 2023/08/28 */ @DispatchFilter(group = {FilterConstants.PUBLISHER, FilterConstants.EXECUTOR}, value = "validation", order = 20) public class ValidationFilter implements Filter { @Override public Result invoke(final Invoker invoker, final Invocation invocation) { if (invoker == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "invoker can't be null"); } if (invocation == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "invocation can't be null"); } if (invocation.getDetail() == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "detail can't be null"); } if (invocation.getProperty() == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "property can't be null"); } if (invocation instanceof SingleInvocation) { this.validate(((SingleInvocation) invocation).getDetail()); } else if (invocation instanceof MultiInvocation) { List<SingleDetailDTO> singleDetailDTOList = ((MultiInvocation) invocation).getDetail(); if (singleDetailDTOList.isEmpty()) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "singleDetailList can't be null or empty"); } for (SingleDetailDTO singleDetail : singleDetailDTOList) { this.validate(singleDetail); } } return invoker.invoke(invocation); } private void validate(SingleDetailDTO singleDetail) { if (singleDetail == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "singleDetail can't be null"); } if (singleDetail.getTemplateCode() == null) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "templateCode can't be null"); } if (singleDetail.getUserContext() == null || singleDetail.getUserContext().size() == 0) { throw new NbpException(NbpCode.ILLEGAL_PARAM.getCode(), "userContext can't be null or empty"); } } }

PublishMetricFilter package com.xxx.arch.mw.nbp.client.extension; import com.xxx.arch.mw.nbp.client.util.EnvUtils; import com.xxx.arch.mw.nbp.client.util.VersionUtils; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import com.xxx.arch.mw.nbp mon.constant.FilterConstants; import com.xxx.arch.mw.nbp mon.constant.InstanceKey; import com.xxx.arch.mw.nbp mon.constant.TraceKey; import com.xxx.arch.mw.nbp mon.domain.NbpCode; import com.xxx.arch.mw.nbp mon.exception.NbpException; import com.xxx.arch.mw.nbp mon.extension.*; import com.xxx.arch.mw.nbp mon.util.TraceUtil; import com.xxx.arch.mw.nbp.share.dto.MultiResult; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; import com.xxx mons.data.domain.Result; import com.xxx mons.data.exception.ServiceException; import org.apache mons.lang3.time.StopWatch; import java.util.List; import java.util.Map; import static com.xxx.arch.mw.nbp.client.logger.LoggerInit.LOGGER_PUBLISH; /** * @date 2023/08/28 */ @DispatchFilter(group = {FilterConstants.PUBLISHER}, value = "publishMetric", order = 200) public class PublishMetricFilter implements Filter { @Override public Result invoke(final Invoker invoker, final Invocation invocation) { if (Boolean.TRUE.equals(invocation.getProperty().getPublisher().getMetricDisabled())) { return invoker.invoke(invocation); } Result result; StopWatch stopWatch = new StopWatch(); try { stopWatch.start(); result = invoker.invoke(invocation); stopWatch.stop(); if (result.isSuccess()) { MultiResult<SingleDetailDTO> resultData = (MultiResult<SingleDetailDTO>) result.getData(); for (SingleDetailDTO singleDetail : resultData.getSuccessList()) { this.log(singleDetail, true, stopWatch, null); } for (SingleDetailDTO singleDetail : resultData.getFailureList()) { this.log(singleDetail, false, stopWatch, null); } } else { // 仅PublishExceptionFilter被禁用才会走该分支 if (invocation instanceof SingleInvocation) { this.log(((SingleInvocation) invocation).getDetail(), false, stopWatch, new NbpException(result.getCode(), result.getMessage(), result.getCause())); } else if (invocation instanceof MultiInvocation) { List<SingleDetailDTO> singleDetailDTOList = ((MultiInvocation) invocation).getDetail(); for (SingleDetailDTO singleDetail : singleDetailDTOList) { this.log(singleDetail, false, stopWatch, new NbpException(result.getCode(), result.getMessage(), result.getCause())); } } } } catch (Throwable e) { if (stopWatch.isStarted()) { stopWatch.stop(); } if (invocation instanceof SingleInvocation) { this.log(((SingleInvocation) invocation).getDetail(), false, stopWatch, e); } else if (invocation instanceof MultiInvocation) { List<SingleDetailDTO> singleDetailDTOList = ((MultiInvocation) invocation).getDetail(); for (SingleDetailDTO singleDetail : singleDetailDTOList) { this.log(singleDetail, false, stopWatch, e); } } throw e; } return result; } private void log(SingleDetailDTO singleDetail, boolean success, StopWatch stopWatch, Throwable throwable) { Map<String, Object> systemContext = singleDetail.getSystemContext(); LOGGER_PUBLISH.info("NBP-CLIENT METRIC PUBLISH", "success={},elapsed={},env={},shadow={},traceId={},rpcId={},version={}," + "code={},id={},bizKey={},triggerTime={},strategyId={},ruleId={},msgId={}," + "publishedTime={},publishedIp={},receivedTime={},receivedIp={}," + "errorCode={}", success, stopWatch.getTime(), EnvUtils.getCurrentEnv().name(), TraceUtil.isShadow(), TraceUtil.getTraceId(), TraceUtil.getRpcId(), VersionUtils.VERSION, singleDetail.getTemplateCode(), singleDetail.getInstanceId(), singleDetail.getBizKey(), singleDetail.getTriggerTime(), systemContext.get(InstanceKey.STRATEGY_ID), systemContext.get(InstanceKey.RULE_ID), singleDetail.getUserContext().get(InstanceKey.DISASTER_MSG_ID), systemContext.get(TraceKey.PUBLISHED_TIME), systemContext.get(TraceKey.PUBLISHED_IP), systemContext.get(TraceKey.RECEIVED_TIME), systemContext.get(TraceKey.RECEIVED_IP), throwable == null ? null : throwable instanceof ServiceException ? ((ServiceException) throwable).getCode() : NbpCode.UNKNOWN.getCode(), throwable == null ? null : throwable); } }

PublishTraceFilter package com.xxx.arch.mw.nbp.client.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import com.xxx.arch.mw.nbp mon.constant.CommonConstants; import com.xxx.arch.mw.nbp mon.constant.FilterConstants; import com.xxx.arch.mw.nbp mon.extension.Filter; import com.xxx.arch.mw.nbp mon.extension.Invocation; import com.xxx.arch.mw.nbp mon.extension.Invoker; import com.xxx.arch.mw.nbp mon.util.TraceUtil; import com.xxx.arch.mw.nbp.share.dto.PropertyDTO; import com.xxx.arch.mw.nbp.share.facade.DispatchPublishService; import com.xxx mons.data.domain.Result; import com.xxx.arch.mw.util.RequestCtxUtil; import static com.xxx.arch.mw.nbp mon.util.TraceUtil.*; import static com.xxx.arch.mw.nbp mon.util.TraceUtil.NBP_RPC_PUB_NAME; /** * @date 2023/08/28 */ @DispatchFilter(group = {FilterConstants.PUBLISHER}, value = "publishTrace", order = 100) public class PublishTraceFilter implements Filter { @Override public Result invoke(final Invoker invoker, final Invocation invocation) { final PropertyDTO property = invocation.getProperty(); if (Boolean.TRUE.equals(property.getPublisher().getEagleEyeDisabled())) { return invoker.invoke(invocation); } Result result; try { TraceUtil.startRpc(String.join(CommonConstants.COLON, DispatchPublishService.class.getCanonicalName(), invocation.getProperty().getTemplateCode()), NBP_PUB_SEND_METHOD_NAME, NBP_RPC_PUB_TYPE); TraceUtil.rpcClientSend(); result = invoker.invoke(invocation); final String remoteIp = RequestCtxUtil.getProviderIp(); if (remoteIp != null) { TraceUtil.remoteIp(remoteIp); } if (result.isSuccess()) { TraceUtil.attribute(RPC_NAME_KEY, NBP_RPC_PUB_NAME); TraceUtil.rpcClientRecv(RPC_RESULT_SUCCESS, "success"); } else { TraceUtil.attribute(RPC_NAME_KEY, NBP_RPC_PUB_NAME); TraceUtil.rpcClientRecv(RPC_RESULT_FAILED, result.getMessage()); } } catch (Throwable e) { TraceUtil.attribute(RPC_NAME_KEY, NBP_RPC_PUB_NAME); TraceUtil.rpcClientRecv(RPC_RESULT_FAILED, e.getMessage()); throw e; } return result; } }

PublishExceptionFilter package com.xxx.arch.mw.nbp.client.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import com.xxx.arch.mw.nbp mon.constant.FilterConstants; import com.xxx.arch.mw.nbp mon.domain.NbpCode; import com.xxx.arch.mw.nbp mon.exception.DegradeException; import com.xxx.arch.mw.nbp mon.exception.FlowException; import com.xxx.arch.mw.nbp mon.exception.NbpException; import com.xxx.arch.mw.nbp mon.exception.RpcException; import com.xxx.arch.mw.nbp mon.extension.Filter; import com.xxx.arch.mw.nbp mon.extension.Invocation; import com.xxx.arch.mw.nbp mon.extension.Invoker; import com.xxx mons.data.domain.Result; import com.xxx.arch.mw.exception.RpcException; /** * @date 2023/08/28 */ @DispatchFilter(group = {FilterConstants.PUBLISHER}, value = "publishException", order = 300) public class PublishExceptionFilter implements Filter { @Override public Result invoke(final Invoker invoker, final Invocation invocation) { try { Result result = invoker.invoke(invocation); if (!result.isSuccess()) { if (NbpCode.FLOW_CONTROL_DENIED.getCode().equals(result.getCode())) { throw new FlowException(NbpCode.FLOW_CONTROL_DENIED.getCode(), result.getMessage(), result.getCause()); } else if (NbpCode.BLACKLIST_DENIED.getCode().equals(result.getCode())) { throw new FlowException(NbpCode.BLACKLIST_DENIED.getCode(), result.getMessage(), result.getCause()); } else if (NbpCode.DEGRADED_DENIED.getCode().equals(result.getCode())) { throw new DegradeException(NbpCode.DEGRADED_DENIED.getCode(), result.getMessage(), result.getCause()); } else { throw new NbpException(result.getCode(), result.getMessage(), result.getCause()); } } return result; } catch (NbpException e) { throw e; } catch (RpcException e) { throw new RpcException(NbpCode.RPC_ERROR.getCode(), e.getMessage(), e.getCause()); } catch (Throwable e) { throw new NbpException(NbpCode.UNKNOWN.getCode(), e.getMessage(), e.getCause()); } } }

PublishCompressFilter package com.xxx.arch.mw.nbp.client.extension; import com.xxx.arch.mw.nbp mon.annotation.DispatchFilter; import com.xxx.arch.mw.nbp mon.constant.CommonConstants; import com.xxx.arch.mw.nbp mon.constant.FilterConstants; import com.xxx.arch.mw.nbp mon.converter.ConverterUtil; import com.xxx.arch.mw.nbp mon.csp.Compressor; import com.xxx.arch.mw.nbp mon.csp.CompressorEnum; import com.xxx.arch.mw.nbp mon.csp.CompressorFactory; import com.xxx.arch.mw.nbp mon.extension.*; import com.xxx.arch.mw.nbp.share.dto.PropertyDTO; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; import com.xxx mons.data.domain.Result; import java.util.Base64; import java.util.HashMap; import java.util.List; import java.util.Map; import static com.xxx.arch.mw.nbp mon.constant.InstanceKey.COMPRESSED_ALGORITHM_KEY; import static com.xxx.arch.mw.nbp mon.constant.InstanceKey.COMPRESSED_CONTEXT_KEY; /** * @date 2023/08/28 */ @DispatchFilter(group = {FilterConstants.PUBLISHER}, value = "publishCompress", order = 700) public class PublishCompressFilter implements Filter { @Override public Result invoke(final Invoker invoker, final Invocation invocation) { final PropertyDTO propertyDTO = invocation.getProperty(); final boolean compressEnabled = propertyDTO.getPublisher().getCompressDisabled() != null ? !propertyDTO.getPublisher().getCompressDisabled() : CommonConstants.USER_CONTEXT_COMPRESS_ENABLED; if (compressEnabled) { final int userContextCompressSizeThreshold = propertyDTO.getPublisher().getCompressSizeThreshold() != null ? propertyDTO.getPublisher().getCompressSizeThreshold() : CommonConstants.USER_CONTEXT_COMPRESS_SIZE_THRESHOLD; final CompressorEnum compressorEnum = CompressorEnum.toEnumFromName(propertyDTO.getPublisher().getCompressAlgorithm()); final Compressor compressor = compressorEnum != null ? CompressorFactory.getCompressor(compressorEnum) : CompressorFactory.getDefaultCompressor(); if (invocation instanceof SingleInvocation) { this.doProcess(((SingleInvocation) invocation).getDetail(), propertyDTO, userContextCompressSizeThreshold, compressorEnum, compressor); } else if (invocation instanceof MultiInvocation) { List<SingleDetailDTO> singleDetailDTOList = ((MultiInvocation) invocation).getDetail(); for (SingleDetailDTO singleDetail : singleDetailDTOList) { this.doProcess(singleDetail, propertyDTO, userContextCompressSizeThreshold, compressorEnum, compressor); } } } return invoker.invoke(invocation); } private void doProcess(SingleDetailDTO singleDetail, PropertyDTO propertyDTO, int userContextCompressSizeThreshold, CompressorEnum compressorEnum, Compressor compressor) { byte[] body = ConverterUtil.toBody(singleDetail.getUserContext()); if (body.length > userContextCompressSizeThreshold) { final Map<String, String> remainUnchangedFields = new HashMap<>(); propertyDTO.getPublisher().getRemainUnchangedFields().forEach(filed -> { if (singleDetail.getUserContext().containsKey(filed)) { remainUnchangedFields.put(filed, singleDetail.getUserContext().remove(filed)); } }); if (remainUnchangedFields.size() > 0) { body = ConverterUtil.toBody(singleDetail.getUserContext(), Map.class); } final byte[] compressedBody = compressor press(body); final String compressedContext = Base64.getEncoder().encodeToString(compressedBody); if (body.length - compressedContext.length() > CommonConstants.USER_CONTEXT_COMPRESS_REVENUE_THRESHOLD) { singleDetail.getUserContext().clear(); singleDetail.getUserContext().put(COMPRESSED_CONTEXT_KEY, compressedContext); if (compressorEnum != null && compressorEnum != Compressor.DEFAULT_COMPRESSOR_ALGORITHM) { singleDetail.getUserContext().put(COMPRESSED_ALGORITHM_KEY, compressorEnum.getName()); } } singleDetail.getUserContext().putAll(remainUnchangedFields); } } }

自定义Invoker PublishInvoker package com.xxx.arch.mw.nbp.client.invoker; import com.xxx.arch.mw.nbp.client.configuration.DispatchProperty; import com.xxx.arch.mw.nbp mon.extension.Invocation; import com.xxx.arch.mw.nbp mon.extension.Invoker; import com.xxx.arch.mw.nbp mon.extension.MultiInvocation; import com.xxx.arch.mw.nbp mon.extension.SingleInvocation; import com.xxx.arch.mw.nbp.share.dto.SingleDetailDTO; import com.xxx.arch.mw.nbp.share.facade.DispatchPublishService; import com.xxx mons.data.domain.Result; import java.util.ArrayList; import java.util.List; /** * @date 2023/08/30 */ public class PublishInvoker implements Invoker { private DispatchPublishService publishService; private DispatchProperty dispatchProperty; public PublishInvoker(DispatchPublishService publishService, DispatchProperty dispatchProperty) { this.publishService = publishService; this.dispatchProperty = dispatchProperty; } @Override public Result invoke(Invocation invocation) { List<SingleDetailDTO> singleDetailDTOS = new ArrayList<>(); if (invocation instanceof SingleInvocation) { singleDetailDTOS.add(((SingleInvocation) invocation).getDetail()); } else if (invocation instanceof MultiInvocation) { singleDetailDTOS = ((MultiInvocation) invocation).getDetail(); } return this.publishService.publish(singleDetailDTOS); } public DispatchProperty getDispatchProperty() { return dispatchProperty; } } Invoker构建与使用 public static final List<Filter> PUBLISH_FILTERS = new ArrayList<Filter>() { { add(new ValidationFilter()); add(new PublishMetricFilter()); add(new PublishExceptionFilter()); add(new PublishTraceFilter()); add(new PublishCompressFilter()); } }; FilterChain publishChain = new FilterChain(); for (Filter filter : ClientFilterConstants.PUBLISH_FILTERS) { publishChain.addFilter(filter); } Invoker invoker = publishChain.buildInvokerChain(new PublishInvoker(publishService, this.property)); Invocation invocation = singleDetailDTOList.size() == 1 ? new SingleInvocation(singleDetailDTOList.get(0), propertyDTO) : new MultiInvocation(singleDetailDTOList, propertyDTO); Result<MultiResult<SingleDetailDTO>> result = invoker.invoke(invocation);