public interface Processor<T> {        ProcessorType getType();        int getOrder();        boolean shouldProcess(RequestContext context);        //对外统一封装为Maybe        Maybe<T> process(RequestContext context) throws Exception; }

public abstract class AbstractProcessor implements Processor {     //同步&无响应，继承此方法     //场景：常规业务处理     protected void processSync(RequestContext context) throws Exception {}
    //同步&有响应，继承此方法，健康检测    //场景：健康检测、未通过校验时的静态响应    protected T processSyncAndGetReponse(RequestContext context) throws Exception {        process(context);        return null;    };
    //异步，继承此方法    //场景：认证、鉴权等涉及远程调用的模块    protected Maybe<T> processAsync(RequestContext context) throws Exception {        T response = processSyncAndGetReponse(context);        if (response == null) {            return Maybe.empty();        } else {            return Maybe.just(response);        }    };
    @Override    public Maybe<T> process(RequestContext context) throws Exception {        Maybe<T> maybe = processAsync(context);        if (maybe instanceof ScalarCallable) {            //标识同步方法，无需额外封装            return maybe;        } else {            //统一加超时，默认忽略错误            return maybe.timeout(getAsyncTimeout(context), TimeUnit.MILLISECONDS,                    Schedulers.from(context.getEventloop()), timeoutFallback(context));        }    }
    protected long getAsyncTimeout(RequestContext context) {        return 2000;    }
    protected Maybe<T> timeoutFallback(RequestContext context) {        return Maybe.empty();    }

public class RxUtil{      //组合某阶段（如Inbound）内的多个filter（即Callable<Maybe<T>>）      public static <T> Maybe<T> concat(Iterable<? extends Callable<Maybe<T>>> iterable) {          Iterator<? extends Callable<Maybe<T>>> sources = iterable.iterator();          while (sources.hasNext()) {              Maybe<T> maybe;              try {                  maybe = sources.next().call();              } catch (Exception e) {                  return Maybe.error(e);              }              if (maybe != null) {                  if (maybe instanceof ScalarCallable) {                      //同步方法                      T response = ((ScalarCallable<T>)maybe).call();                      if (response != null) {                          //有response，中断                          return maybe;                      }                  } else {                      //异步方法                      if (sources.hasNext()) {                          //将sources传入回调，后续filter重复此逻辑                          return new ConcattedMaybe(maybe, sources);                      } else {                          return maybe;                      }                  }              }          }          return Maybe.empty();      }  }

public class ProcessEngine{      //各个阶段，增加默认超时与错误处理    private void process(RequestContext context) {          List<Callable<Maybe<Response>>> inboundTask = get(ProcessorType.INBOUND, context);          List<Callable<Maybe<Void>>> outboundTask = get(ProcessorType.OUTBOUND, context);          List<Callable<Maybe<Response>>> errorTask = get(ProcessorType.ERROR, context);          List<Callable<Maybe<Void>>> logTask = get(ProcessorType.LOG, context);           RxUtil.concat(inboundTask)    //inbound阶段                                  .toSingle()        //获取response                                        .flatMapMaybe(response -> {                  context.setOriginResponse(response);                  return RxUtil.concat(outboundTask);              })            //进入outbound              .onErrorResumeNext(e -> {                  context.setThrowable(e);                  return RxUtil.concat(errorTask).flatMap(response -> {                      context.resetResponse(response);                      return RxUtil.concat(outboundTask);                  });              })            //异常则进入error，并重新进入outbound              .flatMap(response -> RxUtil.concat(logTask))  //日志阶段              .timeout(asyncTimeout.get(), TimeUnit.MILLISECONDS, Schedulers.from(context.getEventloop()),                      Maybe.error(new ServerException(500, "Async-Timeout-Processing"))              )            //全局兜底超时              .subscribe(        //释放资源                      unused -> {                          logger.error("this should not happen, " + context);                          context.release();                      },                      e -> {                          logger.error("this should not happen, " + context, e);                          context.release();                      },                      () -> context.release()              );      }     }

{      //匹配方式      "type": "uri",            //HTTP默认采用uri前缀匹配，内部通过树结构寻址；私有协议（SOTP）通过服务唯一标识定位。      "value": "/hotel/order",      "matcherType": "prefix",            //标签与属性      //用于portal端权限管理、切面逻辑运行（如按核心/非核心）等      "tags": [          "owner_admin",          "org_framework",          "appId_123456"      ],      "properties": {          "core": "true"      },          //endpoint信息      "routes": [{          //condition用于二级路由，如按app版本划分、按query重分配等          "condition": "true",          "conditionParam": {},          "zone": "PRO",                    //具体服务地址，权重用于灰度场景          "targets": [{              "url": "http://test.ctrip.com/hotel",              "weight": 100          }        ]      }]  }

 {      //模块名称，对应网关内部某个具体模块      "name": "addResponseHeader",            //执行阶段      "stage": "PRE_RESPONSE",            //执行顺序      "ruleOrder": 0,            //灰度比例      "grayRatio": 100,            //执行条件      "condition": "true",      "conditionParam": {},            //执行参数      //大量${}形式的内置模板，用于获取运行时数据      "actionParam": {        "connection": "keep-alive",        "x-service-call": "${request.func.remoteCost}",        "Access-Control-Expose-Headers": "x-service-call",        "x-gate-root-id": "${func.catRootMessageId}"      },            //异常处理方式，可以抛出或忽略      "exceptionHandle": "return"   }