自建API网关-架构设计篇

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline p = ch.pipeline();
    p.addLast(new HttpRequestDecoder());
    p.addLast(new HttpResponseEncoder());
    int maxContentLength = 2000;
    try {
        maxContentLength = Integer.parseInt(AppConfigUtil.getValue("netty.maxContentLength"));
    } catch (Exception e) {
        logger.warn("netty.maxContentLength配置异常，系统默认为：2000KB");
 }
    p.addLast(new HttpObjectAggregator(maxContentLength * 1024));// HTTP 消息的合并处理
    p.addLast(new HttpServerInboundHandler());

}

@Override
public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    try {
        if (msg instanceof HttpRequest && msg instanceof HttpContent) {
            CmptRequest cmptRequest = CmptRequestUtil.convert(ctx, msg);
            CmptResult cmptResult = this.gatewayExecutor.execute(cmptRequest);
            FullHttpResponse response = encapsulateResponse(cmptResult);
            ctx.write(response);
            ctx.flush();
        }
    } catch (Exception e) {
        logger.error("网关入口异常，" + e.getMessage());
        e.printStackTrace();
    }

}

/**
 * 微信token认证，token格式：
 * {appID:'',openID:'',timestamp:132525144172,sessionKey: ''}
 */
public class WeixinAuthTokenCmpt extends AbstractCmpt {
    private static Logger logger = LoggerFactory.getLogger(WeixinAuthTokenCmpt.class);
    private final CmptResult SUCCESS_RESULT;

    public WeixinAuthTokenCmpt() {
        SUCCESS_RESULT = buildSuccessResult();
    }

    @Override
    public CmptResult execute(CmptRequest request, Map<String, FieldDTO> config) {
        if (logger.isDebugEnabled()) {
            logger.debug("WeixinTokenCmpt ......");
        }
        CmptResult cmptResult = null;

        //Token认证超时间(传入单位:分)
        long authTokenExpireTime = getAuthTokenExpireTime(config);
        WeixinTokenDTO authTokenDTO = this.getAuthTokenDTO(request);
        logger.debug("Token=" + authTokenDTO);
        AuthTokenState authTokenState = validateToken(authTokenDTO, authTokenExpireTime);
        switch (authTokenState) {
            case ACCESS: {
                cmptResult = SUCCESS_RESULT;
                Map<String, String> header = new HashMap<>();
                header.put(HeaderKeyConstants.HEADER_APP_ID_KEY, authTokenDTO.getAppID());
                header.put(CmptHeaderKeyConstants.HEADER_WEIXIN_OPENID_KEY, authTokenDTO.getOpenID());
                header.put(CmptHeaderKeyConstants.HEADER_WEIXIN_SESSION_KEY, authTokenDTO.getSessionKey());
                cmptResult.setHeader(header);
                break;
            }
            case EXPIRED: {
                cmptResult = buildCmptResult(RespErrCode.AUTH_TOKEN_EXPIRED, "token过期,请重新获取Token！");
                break;
            }
            case INVALID: {
                cmptResult = buildCmptResult(RespErrCode.AUTH_INVALID_TOKEN, "Token无效！");
                break;
            }
        }
        return cmptResult;
    }
...
}

public interface ICmpt {
    /**
     * 组件执行入口
     *
     * @param request
     * @param config，组件实例的参数配置
     * @return
     */
    CmptResult execute(CmptRequest request, Map<String, FieldDTO> config);

    /**
     * 销毁组件持有的特殊资源，比如线程。
     */
    void destroy();
}

public class FieldDTO {

    private String title;

    private String name;

    private FieldType fieldType = FieldType.STRING;

    private String defaultValue;

    private boolean required;

    private String regExp;

    private String description;

}

public class CmptResult {

    RespErrMsg respErrMsg;//组件返回错误信息

    private boolean passed;//组件过滤是否通过

    private byte[] data;//组件返回数据

    private Map<String, String> header = new HashMap<String, String>();//透传后端服务响应头信息

    private MediaType mediaType;//返回响应数据类型

    private Integer statusCode = 200;//默认返回状态码为200

}

public static ICmpt newInstance(final CmptDef cmptDef) {
    ICmpt cmpt = null;
    try {
        final String jarPath = getJarPath(cmptDef);
        if (logger.isDebugEnabled()) {
            logger.debug("尝试载入jar包,jar包路径: " + jarPath);
        }
        //加载依赖jar
        CmptClassLoader cmptClassLoader = CmptClassLoaderManager.loadJar(jarPath, true);
        // 创建实例
        if (null != cmptClassLoader) {
            cmpt = LoadClassUtil.newObject(cmptDef.getFullQualifiedName(), ICmpt.class, cmptClassLoader);
        } else {
            logger.error("加载组件jar包失败! jarPath: " + jarPath);
        }
    } catch (Exception e) {
        logger.error("组件类加载失败，请检查类名和版本是否正确。ClassName=" + cmptDef.getFullQualifiedName() + ", Version=" + cmptDef.getVersion());
        e.printStackTrace();
    }
    return cmpt;
}

Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey(groupKey))
    .andCommandKey(HystrixCommandKey.Factory.asKey(commandKey ))
    .andCommandPropertiesDefaults(HystrixCommandProperties.Setter() // 舱壁隔离策略-信号量
       .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE) //设置每组command可以申请的信号量最大数         .withExecutionIsolationSemaphoreMaxConcurrentRequests(CmptInvoker.maxSemaphore)
/*开启超时设置*/
.withExecutionIsolationThreadInterruptOnTimeout(true)
/*超时时间设置*/
.withExecutionIsolationThreadTimeoutInMilliseconds(timeout)
.withCircuitBreakerEnabled(true)//开启熔断
.withCircuitBreakerSleepWindowInMilliseconds(Constants.DEFAULT_CIRCUIT_BREAKER_SLEEP_WINDOW_IN_MILLISECONDS)// 5秒后会尝试闭合回路

PoolingHttpClientConnectionManager cmOfHttp = new
PoolingHttpClientConnectionManager();
cmOfHttp.setMaxTotal(maxConn);
cmOfHttp.setDefaultMaxPerRoute(maxPerRoute);
httpClient = HttpClients.custom()
        .setConnectionManager(cmOfHttp)
        .setConnectionManagerShared(true)
        .build();