chenzihao/sac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

接口超时配置

Browse Source 
支持不同请求模式open,sac
This commit is contained in:
ztzh_xieyun 2024-05-14 17:56:05 +08:00
parent 8312736200
commit 9e0de4ba4f
10 changed files with 67 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
sf-vertx-api/src/main/java/com/sf/vertx/api/pojo/ApiConfig.java
View File

@ -11,6 +11,7 @@ public class ApiConfig implements Serializable {
private String apiCode;
private String uri;
private String method; // 大写
private Long timeout = 3000L; // 超时时间，单位毫秒
private String mockResponse;
private List<Strategy> strategy; // 策略
}

1
sf-vertx-api/src/main/java/com/sf/vertx/api/pojo/SacService.java
View File

@ -9,6 +9,7 @@ import lombok.Data;
public class SacService implements Serializable {
private static final long serialVersionUID = -5171112142954536813L;
private String serviceName; // 服务名
private String serviceType; // 服务类型，SAC=SAC规范服务，OPEN=开放服务
private String serviceModel; // 模式, NORMAL, ROUTE
private ServerAddress serverAddress; // NORMAL模式的服务地址
private List<ApiConfig> apiConfig; // request set header sacApiCode

2
sf-vertx-api/src/main/java/com/sf/vertx/api/pojo/VertxConfig.java
View File

@ -10,7 +10,7 @@ public class VertxConfig implements Serializable {
private Integer port; // 启动端口
private String appCodeHeaderKey = "sacAppCode";
private String apiCodeHeaderKey = "sacApiCode";
private String rateLimitModel = "redis"; // local,redis 负载均衡模式
private String rateLimitModel = "local"; // local,redis 负载均衡模式
private VertxOptionsConfig vertxOptionsConfig;
private HttpClientOptionsConfig httpClientOptionsConfig; // 配置Vert端口连接池
private AddressRetryStrategy addressRetryStrategy;

3
sf-vertx/src/main/java/com/sf/AdminApplication.java
View File

@ -5,14 +5,11 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.context.ApplicationContext;
import lombok.extern.slf4j.Slf4j;
/**
* 启动程序
*
* @author ztzh
*/
@Slf4j
@SpringBootApplication(exclude = { DataSourceAutoConfiguration.class })
public class AdminApplication {
// 全局 ApplicationContext 实例，方便 getBean 拿到 Spring/SpringBoot 注入的类实例

58
sf-vertx/src/main/java/com/sf/vertx/handle/AppConfigHandler.java
View File

@ -80,7 +80,8 @@ public class AppConfigHandler {
private static ConcurrentHashMap<String, ApiConfig> APICODE_CONFIG_MAP = new ConcurrentHashMap<>();
// apiCode限流配置 appCode:apiCode - RateLimiterRegistry
private static ConcurrentHashMap<String, SacCurrentLimiting> APICODE_CONFIG_CURRENT_LIMITING_MAP = new ConcurrentHashMap<>();
// 服务类型, apiCode限流配置 appCode:apiCode - 服务类型，SAC=SAC规范服务，OPEN=开放服务
private static ConcurrentHashMap<String, String> APICODE_CONFIG_SERVICE_TYPE_MAP = new ConcurrentHashMap<>();
// 负载均衡路由类型 appCode:apiCode - routerType
// 执行流程 routerType= <br/>
// 1、serviceNodel="NORMAL", serviceNodel="ROUTE" and RouteType = "WEIGHT_ROUTE"
@ -90,7 +91,7 @@ public class AppConfigHandler {
// return APICODE_CONFIG_ROUTERCONENT_MAP
private static ConcurrentHashMap<String, Integer> APICODE_CONFIG_ROUTERTYPE_MAP = new ConcurrentHashMap<>();
private static ConcurrentHashMap<String, List<RouteContent>> APICODE_CONFIG_ROUTERCONENT_MAP = new ConcurrentHashMap<>();
private static ConcurrentHashMap<String, List<RouteContent>> APICODE_CONFIG_HEADER_ROUTERCONENT_MAP = new ConcurrentHashMap<>();
// apiCode熔断配置 appCode:apiCode - CircuitBreaker
private static ConcurrentHashMap<String, CircuitBreaker> APICODE_CONFIG_CIRCUIT_BREAKER_MAP = new ConcurrentHashMap<>();
@ -101,12 +102,18 @@ public class AppConfigHandler {
return APICODE_CONFIG_ROUTERTYPE_MAP.get(key) != null ? APICODE_CONFIG_ROUTERTYPE_MAP.get(key) : 1;
}
public static List<RouteContent> routerConentList(String key) {
return APICODE_CONFIG_ROUTERCONENT_MAP.get(key);
public static List<RouteContent> routerHeaderConentList(String key) {
return APICODE_CONFIG_HEADER_ROUTERCONENT_MAP.get(key);
}
public static Integer requestModel() {
return sacVertxConfig.getRequestModel();
public static Boolean isServiceTypeOpen(String key) {
return APICODE_CONFIG_SERVICE_TYPE_MAP.get(key) != null
&& StringUtils.equals(APICODE_CONFIG_SERVICE_TYPE_MAP.get(key), "OPEN");
}
public static Boolean isServiceTypeSac(String key) {
return APICODE_CONFIG_SERVICE_TYPE_MAP.get(key) != null
&& StringUtils.equals(APICODE_CONFIG_SERVICE_TYPE_MAP.get(key), "SAC");
}
public static String sacResponseHeaderKey() {
@ -199,12 +206,20 @@ public class AppConfigHandler {
return LOADBALANCING_MAP.get(key);
}
public static ApiConfig getApicodeConfigMap(String key) {
public static ApiConfig getApicodeConfig(String key) {
return APICODE_CONFIG_MAP.get(key);
}
public static Long getApicodeConfigTimeOut(String key) {
return APICODE_CONFIG_MAP.get(key) != null ? APICODE_CONFIG_MAP.get(key).getTimeout() : 3000L;
}
public static boolean isApicodeUri(String key, String uri, String httpMethod) {
return StringUtils.equals(APICODE_CONFIG_MAP.get(key).getUri(), uri)
String apiCodeUri = APICODE_CONFIG_MAP.get(key).getUri();
if (StringUtils.equals(apiCodeUri, "*")) {
return true;
}
return StringUtils.equals(apiCodeUri, uri)
&& StringUtils.equals(httpMethod, APICODE_CONFIG_MAP.get(key).getMethod());
}
@ -264,6 +279,7 @@ public class AppConfigHandler {
String key = appCode + ":" + apiConfig.getApiCode();
APICODE_CONFIG_MAP.remove(key);
LOADBALANCING_MAP.remove(key);
APICODE_CONFIG_SERVICE_TYPE_MAP.remove(key);
APICODE_CONFIG_CURRENT_LIMITING_MAP.remove(key);
String keyCircuitBreaker = key + ":" + "CIRCUIT_BREAKER";
CircuitBreaker circuitBreaker = APICODE_CONFIG_CIRCUIT_BREAKER_MAP.get(keyCircuitBreaker);
@ -339,6 +355,7 @@ public class AppConfigHandler {
for (ApiConfig apiConfig : sacService.getApiConfig()) {
String key = appCode + ":" + apiConfig.getApiCode();
APICODE_CONFIG_MAP.put(key, apiConfig);
APICODE_CONFIG_SERVICE_TYPE_MAP.put(key, sacService.getServiceType());
// 负载均衡模式
APICODE_CONFIG_ROUTERTYPE_MAP.put(key, routerType);
@ -351,14 +368,13 @@ public class AppConfigHandler {
}
break;
case 2:
APICODE_CONFIG_ROUTERCONENT_MAP.put(key, routeContentList);
APICODE_CONFIG_HEADER_ROUTERCONENT_MAP.put(key, routeContentList);
default:
break;
}
if (apiConfig.getStrategy() != null && apiConfig.getStrategy().size() > 0) {
for (Strategy strategy : apiConfig.getStrategy()) {
if (StringUtils.equals(strategy.getType(), "CURRENT_LIMITING")) {
RateLimiterRegistry registry = createRateLimiter(strategy);
SacCurrentLimiting sacCurrentLimiting = new SacCurrentLimiting();
@ -381,7 +397,7 @@ public class AppConfigHandler {
.create(keyCircuitBreaker + "-circuit-breaker", VERTX,
new CircuitBreakerOptions().setMaxFailures(strategy.getThreshold()) // 最大失败数
.setFailuresRollingWindow(strategy.getTimeWindow() * 1000) // 毫秒
// .setTimeout(2000) // 超时时间
//.setTimeout(apiConfig.getTimeout()) // 超时时间,不要开启, 配置会设置接口超时, 这个参数有bug,半开超时会卡死
.setFallbackOnFailure(true) // 失败后是否调用回退函数（fallback）
.setResetTimeout(strategy.getRecovery_interval() * 1000) // 在开启状态下，尝试重试之前所需时间
).openHandler(v -> {
@ -534,16 +550,14 @@ public class AppConfigHandler {
// // 会跳转到 RestfulFailureHandlerImpl
// throw new HttpException(10003);
// }
if (AppConfigHandler.requestModel() == 2) {
String appCode = context.request().headers().get(getAppCodeHeaderKey());
String apiCode = context.request().headers().get(getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
if (isServiceTypeSac(key)) {
String uri = APICODE_CONFIG_MAP.get(key).getUri();
String method = APICODE_CONFIG_MAP.get(key).getMethod();
context.request().setURI(uri).setMethod(HttpMethod.valueOf(method));
}
return context.sendRequest();
}
@ -559,19 +573,17 @@ public class AppConfigHandler {
}
});
WebClient mainWebClient = WebClient.create(VERTX);
String rateLimitModel = vertxConfig.getRateLimitModel();
rateLimitModel = "local";
Route route = null;
if (requestModel() == 2) {
route = mainHttpRouter.route(rpcUri());
} else {
route = mainHttpRouter.route();
}
route.handler(ParameterCheckHandler.create()).handler(AppRateLimitHandler.create(rateLimitModel))
Route routeSac = mainHttpRouter.route(rpcUri());
Route routeOpen = mainHttpRouter.route();
routeSac.handler(ParameterCheckHandler.create()).handler(AppRateLimitHandler.create(rateLimitModel))
.handler(ApiRateLimitHandler.create(rateLimitModel)).handler(BodyHandler.create())
.handler(ProxyHandler.create(mainWebClient, proxy)).failureHandler(RestfulFailureHandler.create());
// mainHttpRouter.route().handler(ProxyHandler.create(mainWebClient, proxy));
routeOpen.handler(ParameterCheckHandler.create()).handler(AppRateLimitHandler.create(rateLimitModel))
.handler(ApiRateLimitHandler.create(rateLimitModel)).handler(BodyHandler.create())
.handler(ProxyHandler.create(mainWebClient, proxy)).failureHandler(RestfulFailureHandler.create());
}
/***

5
sf-vertx/src/main/java/com/sf/vertx/handle/ParameterCheckHandlerImpl.java
View File

@ -12,13 +12,12 @@ public class ParameterCheckHandlerImpl implements ParameterCheckHandler {
@Override
public void handle(RoutingContext rc) {
try {
// TODO 测试异常
String appCode = rc.request().headers().get(AppConfigHandler.getAppCodeHeaderKey());
String apiCode = rc.request().headers().get(AppConfigHandler.getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
if (StringUtils.isBlank(appCode) || StringUtils.isBlank(apiCode)
|| AppConfigHandler.getAppConfig(appCode) == null
|| AppConfigHandler.getApicodeConfigMap(key) == null) {
|| AppConfigHandler.getApicodeConfig(key) == null) {
rc.fail(new HttpException(10012));
return;
}
@ -31,7 +30,7 @@ public class ParameterCheckHandlerImpl implements ParameterCheckHandler {
// 通过请求模式来区分
String uri = rc.request().uri();
String httpMethod = rc.request().method().name();
if(AppConfigHandler.requestModel() == 1 && AppConfigHandler.isApicodeUri(key, uri, httpMethod) == false) {
if(AppConfigHandler.isServiceTypeOpen(key) && AppConfigHandler.isApicodeUri(key, uri, httpMethod) == false) {
rc.fail(new HttpException(10018));
return;
}

3
sf-vertx/src/main/java/com/sf/vertx/init/SacVertxConfig.java
View File

@ -28,9 +28,6 @@ public class SacVertxConfig {
@Value("${server.vertx.sacResponseHeaderKey:sacErrorCode}")
private String sacResponseHeaderKey;
@Value("${server.vertx.requestModel:2}")
private Integer requestModel;
@Value("${server.vertx.rpcUri:/rpc}")
private String rpcUri;

2
sf-vertx/src/main/java/com/sf/vertx/utils/ProxyTool.java
View File

@ -39,7 +39,7 @@ public class ProxyTool {
log.info("sacLoadBalancing address:{},port:{}", socketAddress.host(), socketAddress.port());
return socketAddress;
case 2:
List<RouteContent> routeContentList = AppConfigHandler.routerConentList(key);
List<RouteContent> routeContentList = AppConfigHandler.routerHeaderConentList(key);
if(routeContentList != null && routeContentList.size() > 0) {
for (RouteContent routeContent : routeContentList) {
String headerValue = request.getHeader(routeContent.getHeaderKey());

26
sf-vertx/src/main/java/io/vertx/httpproxy/impl/ReverseProxy.java
View File

@ -22,7 +22,6 @@ import org.apache.commons.lang3.StringUtils;
import com.sf.vertx.api.pojo.DataSecurity;
import com.sf.vertx.constans.SacErrorCode;
import com.sf.vertx.handle.AppConfigHandler;
import com.sf.vertx.init.SacVertxConfig;
import com.sf.vertx.security.MainSecurity;
import com.sf.vertx.utils.ProxyTool;
@ -245,9 +244,12 @@ public class ReverseProxy implements HttpProxy {
private Future<ProxyResponse> breakerAndSecurity(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
String apiCode = proxyRequest.headers().get(AppConfigHandler.getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
return Future.future(p -> {
circuitBreaker.executeWithFallback(promise -> {
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
requestBuffer.timeout(AppConfigHandler.getApicodeConfigTimeOut(key));
requestBuffer.putHeaders(proxyRequest.headers())
.sendJson(bodyDecrypt(ctx.getBodyAsString(), appCode), h -> {
if (h.succeeded()) {
@ -302,13 +304,17 @@ public class ReverseProxy implements HttpProxy {
}
private Future<ProxyResponse> breaker(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
String apiCode = proxyRequest.headers().get(AppConfigHandler.getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
return Future.future(p -> {
circuitBreaker.executeWithFallback(promise -> {
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
requestBuffer.timeout(AppConfigHandler.getApicodeConfigTimeOut(key));
requestBuffer.putHeaders(proxyRequest.headers()).sendJson(ctx.getBodyAsString(), h -> {
log.info("==========uri:{},response http code:{}, succeeded:{}", proxyRequest.getURI(),
h.result().statusCode(), h.succeeded());
if (h.succeeded()) {
log.info("==========uri:{},response http code:{}", proxyRequest.getURI(),
h.result().statusCode());
if (h.result().statusCode() == 200) {
// promise.complete();
promise.complete("1");
@ -351,8 +357,11 @@ public class ReverseProxy implements HttpProxy {
private Future<ProxyResponse> security(CircuitBreaker circuitBreaker, ProxyRequest proxyRequest) {
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
String apiCode = proxyRequest.headers().get(AppConfigHandler.getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
return Future.future(p -> {
HttpRequest<Buffer> requestBuffer = methodGetRequestBuffer(proxyRequest);
requestBuffer.timeout(AppConfigHandler.getApicodeConfigTimeOut(key));
requestBuffer.putHeaders(proxyRequest.headers()).sendJson(bodyDecrypt(ctx.getBodyAsString(), appCode),
h -> {
if (h.succeeded()) {
@ -451,9 +460,18 @@ public class ReverseProxy implements HttpProxy {
}
private Future<ProxyResponse> sendProxyRequest(ProxyRequest proxyRequest, HttpClientRequest request) {
String appCode = proxyRequest.headers().get(AppConfigHandler.getAppCodeHeaderKey());
String apiCode = proxyRequest.headers().get(AppConfigHandler.getApiCodeHeaderKey());
String key = appCode + ":" + apiCode;
request.setTimeout(AppConfigHandler.getApicodeConfigTimeOut(key));
Future<ProxyResponse> fut = proxyRequest.send(request);
fut.onFailure(err -> {
proxyRequest.proxiedRequest().response().setStatusCode(502).end();
JsonObject json = SacErrorCode.returnErrorMsg(502);
proxyRequest.response().release().setStatusCode(500).putHeader("content-type", "application/json")
.putHeader(AppConfigHandler.sacResponseHeaderKey(), String.valueOf(502))
.putHeader(HttpHeaders.CONTENT_LENGTH, String.valueOf(json.size())).setBody(Body.body(json.toBuffer()))
.send();
//proxyRequest.proxiedRequest().response().setStatusCode(502).end();
});
return fut;
}

1
sf-vertx/src/main/resources/application.yml
View File

@ -2,7 +2,6 @@
server:
vertx:
deploymentMode: 1 # 1:单机 2:集群
requestModel: 2 # 1: 客户端传递uri. 2: uri vertx代理,不对客户端暴露uri
isSSL: false # vertx https或者http启动, ssl 证书有过期时间
rpcUri: /rpc
sacResponseHeaderKey: sacErrorCode