tanghc
5 years ago
parent
b76684c3c6
commit
68e9f173f2
@ -0,0 +1,114 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.bean; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.interceptor.RouteInterceptorContext; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public class DefaultRouteInterceptorContext implements RouteInterceptorContext { |
||||||
|
|
||||||
|
/** 请求参数 */ |
||||||
|
private ApiParam apiParam; |
||||||
|
/** 错误信息 */ |
||||||
|
private String serviceErrorMsg; |
||||||
|
/** 微服务返回状态 */ |
||||||
|
private int responseStatus; |
||||||
|
/** 开始时间 */ |
||||||
|
private long beginTimeMillis; |
||||||
|
/** 结束时间 */ |
||||||
|
private long finishTimeMillis; |
||||||
|
/** 请求上下文 */ |
||||||
|
private Object requestContext; |
||||||
|
/** 微服务返回结果 */ |
||||||
|
private String serviceResult; |
||||||
|
/** 请求包大小 */ |
||||||
|
private long requestDataSize; |
||||||
|
/** 返回内容大小 */ |
||||||
|
private long responseDataSize; |
||||||
|
|
||||||
|
@Override |
||||||
|
public ApiParam getApiParam() { |
||||||
|
return apiParam; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public String getServiceResult() { |
||||||
|
return serviceResult; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public int getResponseStatus() { |
||||||
|
return responseStatus; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public long getBeginTimeMillis() { |
||||||
|
return beginTimeMillis; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public long getFinishTimeMillis() { |
||||||
|
return finishTimeMillis; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public Object getRequestContext() { |
||||||
|
return requestContext; |
||||||
|
} |
||||||
|
|
||||||
|
public void setApiParam(ApiParam apiParam) { |
||||||
|
this.apiParam = apiParam; |
||||||
|
} |
||||||
|
|
||||||
|
public void setServiceResult(String serviceResult) { |
||||||
|
this.serviceResult = serviceResult; |
||||||
|
} |
||||||
|
|
||||||
|
public void setResponseStatus(int responseStatus) { |
||||||
|
this.responseStatus = responseStatus; |
||||||
|
} |
||||||
|
|
||||||
|
public void setBeginTimeMillis(long beginTimeMillis) { |
||||||
|
this.beginTimeMillis = beginTimeMillis; |
||||||
|
} |
||||||
|
|
||||||
|
public void setFinishTimeMillis(long finishTimeMillis) { |
||||||
|
this.finishTimeMillis = finishTimeMillis; |
||||||
|
} |
||||||
|
|
||||||
|
public void setRequestContext(Object requestContext) { |
||||||
|
this.requestContext = requestContext; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public String getServiceErrorMsg() { |
||||||
|
return serviceErrorMsg; |
||||||
|
} |
||||||
|
|
||||||
|
public void setServiceErrorMsg(String serviceErrorMsg) { |
||||||
|
this.serviceErrorMsg = serviceErrorMsg; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public long getRequestDataSize() { |
||||||
|
return requestDataSize; |
||||||
|
} |
||||||
|
|
||||||
|
public void setRequestDataSize(long requestDataSize) { |
||||||
|
// spring cloud gateway get请求contentLength返回-1
|
||||||
|
if (requestDataSize < 0) { |
||||||
|
requestDataSize = 0; |
||||||
|
} |
||||||
|
this.requestDataSize = requestDataSize; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public long getResponseDataSize() { |
||||||
|
return responseDataSize; |
||||||
|
} |
||||||
|
|
||||||
|
public void setResponseDataSize(long responseDataSize) { |
||||||
|
this.responseDataSize = responseDataSize; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,21 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.gateway.controller; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.support.BaseMonitorController; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import org.springframework.web.bind.annotation.RestController; |
||||||
|
import org.springframework.web.server.ServerWebExchange; |
||||||
|
|
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
@RestController |
||||||
|
public class GatewayMonitorController extends BaseMonitorController<ServerWebExchange> { |
||||||
|
|
||||||
|
@Override |
||||||
|
protected ApiParam getApiParam(ServerWebExchange request) { |
||||||
|
Map<String, String> params = request.getRequest().getQueryParams().toSingleValueMap(); |
||||||
|
return ApiParam.build(params); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,96 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.interceptor; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.bean.ApiConfig; |
||||||
|
import com.gitee.sop.gatewaycommon.monitor.MonitorInfo; |
||||||
|
import com.gitee.sop.gatewaycommon.monitor.MonitorManager; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
|
||||||
|
import java.util.ArrayList; |
||||||
|
import java.util.concurrent.LinkedBlockingQueue; |
||||||
|
import java.util.concurrent.ThreadPoolExecutor; |
||||||
|
import java.util.concurrent.TimeUnit; |
||||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||||
|
|
||||||
|
/** |
||||||
|
* SOP默认的拦截器，用于收集监控数据 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public class MonitorRouteInterceptor implements RouteInterceptor { |
||||||
|
|
||||||
|
private ThreadPoolExecutor threadPoolExecutor; |
||||||
|
|
||||||
|
public MonitorRouteInterceptor(int threadPoolSize) { |
||||||
|
threadPoolExecutor = new ThreadPoolExecutor(threadPoolSize, threadPoolSize, |
||||||
|
0L, TimeUnit.MILLISECONDS, |
||||||
|
new LinkedBlockingQueue<>()); |
||||||
|
} |
||||||
|
|
||||||
|
public MonitorRouteInterceptor() { |
||||||
|
this(4); |
||||||
|
} |
||||||
|
|
||||||
|
public MonitorRouteInterceptor(ThreadPoolExecutor threadPoolExecutor) { |
||||||
|
this.threadPoolExecutor = threadPoolExecutor; |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void preRoute(RouteInterceptorContext context) { |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void afterRoute(RouteInterceptorContext context) { |
||||||
|
threadPoolExecutor.execute(() -> this.storeRequestInfo(context)); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* 记录接口调用流量，最大时间，最小时间，总时长，平均时长，调用次数，成功次数，失败次数. |
||||||
|
* 需要考虑并发情况。 |
||||||
|
*/ |
||||||
|
protected void storeRequestInfo(RouteInterceptorContext context) { |
||||||
|
MonitorManager monitorManager = ApiConfig.getInstance().getMonitorManager(); |
||||||
|
ApiParam apiParam = context.getApiParam(); |
||||||
|
String routeId = apiParam.fetchNameVersion(); |
||||||
|
long spendTime = context.getFinishTimeMillis() - context.getBeginTimeMillis(); |
||||||
|
// 这步操作是线程安全的，底层调用了ConcurrentHashMap.computeIfAbsent
|
||||||
|
MonitorInfo monitorInfo = monitorManager.getMonitorInfo(routeId, (k) -> this.createMonitorInfo(apiParam)); |
||||||
|
|
||||||
|
monitorInfo.storeMaxTime(spendTime); |
||||||
|
monitorInfo.storeMinTime(spendTime); |
||||||
|
monitorInfo.getTotalCount().incrementAndGet(); |
||||||
|
monitorInfo.getTotalTime().addAndGet(spendTime); |
||||||
|
monitorInfo.getTotalRequestDataSize().addAndGet(context.getRequestDataSize()); |
||||||
|
monitorInfo.getTotalResponseDataSize().addAndGet(context.getResponseDataSize()); |
||||||
|
if (context.isSuccessRequest()) { |
||||||
|
monitorInfo.getSuccessCount().incrementAndGet(); |
||||||
|
} else { |
||||||
|
monitorInfo.getErrorCount().incrementAndGet(); |
||||||
|
String errorMsg = context.getServiceErrorMsg(); |
||||||
|
monitorInfo.addErrorMsg(errorMsg); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
private MonitorInfo createMonitorInfo(ApiParam apiParam) { |
||||||
|
MonitorInfo monitorInfo = new MonitorInfo(); |
||||||
|
monitorInfo.setName(apiParam.fetchName()); |
||||||
|
monitorInfo.setVersion(apiParam.fetchVersion()); |
||||||
|
monitorInfo.setServiceId(apiParam.fetchServiceId()); |
||||||
|
monitorInfo.setTotalRequestDataSize(new AtomicLong()); |
||||||
|
monitorInfo.setTotalResponseDataSize(new AtomicLong()); |
||||||
|
monitorInfo.setTotalTime(new AtomicLong()); |
||||||
|
monitorInfo.setMaxTime(0L); |
||||||
|
monitorInfo.setMinTime(0L); |
||||||
|
monitorInfo.setSuccessCount(new AtomicLong()); |
||||||
|
monitorInfo.setTotalCount(new AtomicLong()); |
||||||
|
monitorInfo.setErrorCount(new AtomicLong()); |
||||||
|
monitorInfo.setErrorMsgList(new ArrayList<>(10)); |
||||||
|
return monitorInfo; |
||||||
|
} |
||||||
|
|
||||||
|
|
||||||
|
@Override |
||||||
|
public int getOrder() { |
||||||
|
return -1000; |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,40 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.interceptor; |
||||||
|
|
||||||
|
/** |
||||||
|
* 路由拦截器 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public interface RouteInterceptor { |
||||||
|
|
||||||
|
/** |
||||||
|
* 在路由转发前执行，签名校验通过后会立即执行此方法 |
||||||
|
* |
||||||
|
* @param context context |
||||||
|
*/ |
||||||
|
void preRoute(RouteInterceptorContext context); |
||||||
|
|
||||||
|
/** |
||||||
|
* 微服务返回结果后执行，可能返回成功，也可能返回失败结果， |
||||||
|
* 可通过 {@link RouteInterceptorContext#getResponseStatus()} 来判断是否返回正确结果。 |
||||||
|
* |
||||||
|
* @param context context |
||||||
|
*/ |
||||||
|
void afterRoute(RouteInterceptorContext context); |
||||||
|
|
||||||
|
/** |
||||||
|
* 拦截器执行顺序，值小优先执行，建议从0开始，小于0留给系统使用 |
||||||
|
* |
||||||
|
* @return 返回顺序 |
||||||
|
*/ |
||||||
|
int getOrder(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 是否匹配，返回true执行拦截器，默认true |
||||||
|
* @param context context |
||||||
|
* @return 返回true执行拦截器 |
||||||
|
*/ |
||||||
|
default boolean match(RouteInterceptorContext context) { |
||||||
|
return true; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,85 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.interceptor; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.bean.SopConstants; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import org.springframework.http.HttpStatus; |
||||||
|
|
||||||
|
/** |
||||||
|
* 拦截器参数 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public interface RouteInterceptorContext { |
||||||
|
|
||||||
|
/** |
||||||
|
* 返回ApiParam |
||||||
|
* |
||||||
|
* @return 返回ApiParam |
||||||
|
*/ |
||||||
|
ApiParam getApiParam(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取微服务返回的内容 |
||||||
|
* |
||||||
|
* @return 微服务返回内容 |
||||||
|
*/ |
||||||
|
String getServiceResult(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取微服务端的错误信息，status为200时，返回null。 |
||||||
|
* |
||||||
|
* @return 返回错误信息 |
||||||
|
*/ |
||||||
|
String getServiceErrorMsg(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取微服务返回状态码 |
||||||
|
* |
||||||
|
* @return 返回状态码，正确为200，错误为非200 |
||||||
|
*/ |
||||||
|
int getResponseStatus(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取路由开始时间 |
||||||
|
* |
||||||
|
* @return 返回开始时间 |
||||||
|
*/ |
||||||
|
long getBeginTimeMillis(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取路由结束时间 |
||||||
|
* |
||||||
|
* @return 返回结束时间 |
||||||
|
*/ |
||||||
|
long getFinishTimeMillis(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取上下文信息，zuul返回RequestContext，Gateway返回ServerWebExchange |
||||||
|
* |
||||||
|
* @return 返回上下文对象 |
||||||
|
*/ |
||||||
|
Object getRequestContext(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取请求内容大小 |
||||||
|
* |
||||||
|
* @return 返回请求内容大小 |
||||||
|
*/ |
||||||
|
long getRequestDataSize(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 获取返回结果内容大小 |
||||||
|
* @return 返回返回结果内容大小 |
||||||
|
*/ |
||||||
|
long getResponseDataSize(); |
||||||
|
|
||||||
|
/** |
||||||
|
* 是否是成功请求，微服务主动抛出的异常也算作成功，JSR303校验失败也算作成功。 |
||||||
|
* 只有微服务返回未知的错误算作失败。 |
||||||
|
* @return true：成功请求 |
||||||
|
*/ |
||||||
|
default boolean isSuccessRequest() { |
||||||
|
int responseStatus = getResponseStatus(); |
||||||
|
return responseStatus == HttpStatus.OK.value() || responseStatus == SopConstants.BIZ_ERROR_STATUS; |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,88 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.monitor; |
||||||
|
|
||||||
|
import lombok.Data; |
||||||
|
|
||||||
|
import java.util.List; |
||||||
|
import java.util.concurrent.atomic.AtomicLong; |
||||||
|
|
||||||
|
/** |
||||||
|
* 每个接口 总调用流量，最大时间，最小时间，总时长，平均时长，调用次数，成功次数，失败次数，错误查看。 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
@Data |
||||||
|
public class MonitorInfo { |
||||||
|
|
||||||
|
/** |
||||||
|
* 接口名 |
||||||
|
*/ |
||||||
|
private String name; |
||||||
|
/** |
||||||
|
* 版本号 |
||||||
|
*/ |
||||||
|
private String version; |
||||||
|
/** |
||||||
|
* serviceId |
||||||
|
*/ |
||||||
|
private String serviceId; |
||||||
|
/** |
||||||
|
* 总请求数据量 |
||||||
|
*/ |
||||||
|
private AtomicLong totalRequestDataSize; |
||||||
|
/** |
||||||
|
* 总返回数据量 |
||||||
|
*/ |
||||||
|
private AtomicLong totalResponseDataSize; |
||||||
|
/** |
||||||
|
* 请求耗时最长时间 |
||||||
|
*/ |
||||||
|
private Long maxTime; |
||||||
|
/** |
||||||
|
* 请求耗时最小时间 |
||||||
|
*/ |
||||||
|
private Long minTime; |
||||||
|
/** |
||||||
|
* 总时长 |
||||||
|
*/ |
||||||
|
private AtomicLong totalTime; |
||||||
|
/** |
||||||
|
* 总调用次数 |
||||||
|
*/ |
||||||
|
private AtomicLong totalCount; |
||||||
|
/** |
||||||
|
* 成功次数 |
||||||
|
*/ |
||||||
|
private AtomicLong successCount; |
||||||
|
/** |
||||||
|
* 失败次数（业务主动抛出的异常算作成功，如参数校验，未知的错误算失败） |
||||||
|
*/ |
||||||
|
private AtomicLong errorCount; |
||||||
|
/** |
||||||
|
* 错误信息 |
||||||
|
*/ |
||||||
|
private List<String> errorMsgList; |
||||||
|
|
||||||
|
public synchronized void storeMaxTime(long spendTime) { |
||||||
|
if (spendTime > maxTime) { |
||||||
|
maxTime = spendTime; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
public synchronized void storeMinTime(long spendTime) { |
||||||
|
if (minTime == 0 || spendTime < minTime) { |
||||||
|
minTime = spendTime; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
public void addErrorMsg(String errorMsg) { |
||||||
|
if (errorMsg == null || "".equals(errorMsg)) { |
||||||
|
return; |
||||||
|
} |
||||||
|
synchronized (this) { |
||||||
|
if (errorMsgList != null && errorMsgList.size() < 10) { |
||||||
|
errorMsgList.add(errorMsg); |
||||||
|
} |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,22 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.monitor; |
||||||
|
|
||||||
|
import java.util.Map; |
||||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||||
|
import java.util.function.Function; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public class MonitorManager { |
||||||
|
|
||||||
|
private static Map<String, MonitorInfo> monitorMap = new ConcurrentHashMap<>(128); |
||||||
|
|
||||||
|
public Map<String, MonitorInfo> getMonitorData() { |
||||||
|
return monitorMap; |
||||||
|
} |
||||||
|
|
||||||
|
public MonitorInfo getMonitorInfo(String routeId, Function<String, MonitorInfo> createFun) { |
||||||
|
return monitorMap.computeIfAbsent(routeId, createFun); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,21 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.support; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.bean.ApiConfig; |
||||||
|
import com.gitee.sop.gatewaycommon.monitor.MonitorManager; |
||||||
|
import com.gitee.sop.gatewaycommon.result.ApiResult; |
||||||
|
import org.springframework.web.bind.annotation.GetMapping; |
||||||
|
|
||||||
|
/** |
||||||
|
* 提供监控数据 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public abstract class BaseMonitorController<T> extends SopBaseController<T> { |
||||||
|
|
||||||
|
@GetMapping("/sop/getMonitorData") |
||||||
|
public ApiResult doExecute(T request) { |
||||||
|
MonitorManager monitorManager = ApiConfig.getInstance().getMonitorManager(); |
||||||
|
return execute(request, monitorManager::getMonitorData); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,45 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.support; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import com.gitee.sop.gatewaycommon.result.ApiResult; |
||||||
|
import com.gitee.sop.gatewaycommon.result.JsonResult; |
||||||
|
import com.gitee.sop.gatewaycommon.validate.taobao.TaobaoSigner; |
||||||
|
import org.springframework.beans.factory.annotation.Value; |
||||||
|
|
||||||
|
import java.util.function.Supplier; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public abstract class SopBaseController<T> { |
||||||
|
|
||||||
|
TaobaoSigner signer = new TaobaoSigner(); |
||||||
|
|
||||||
|
@Value("${sop.secret}") |
||||||
|
private String secret; |
||||||
|
|
||||||
|
protected abstract ApiParam getApiParam(T t); |
||||||
|
|
||||||
|
public ApiResult execute(T request, Supplier<Object> supplier) { |
||||||
|
try { |
||||||
|
this.check(request); |
||||||
|
JsonResult apiResult = new JsonResult(); |
||||||
|
apiResult.setData(supplier.get()); |
||||||
|
return apiResult; |
||||||
|
} catch (Exception e) { |
||||||
|
ApiResult apiResult = new ApiResult(); |
||||||
|
apiResult.setCode("505050"); |
||||||
|
apiResult.setMsg(e.getMessage()); |
||||||
|
return apiResult; |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
protected void check(T request) { |
||||||
|
ApiParam apiParam = getApiParam(request); |
||||||
|
boolean right = signer.checkSign(apiParam, secret); |
||||||
|
if (!right) { |
||||||
|
throw new RuntimeException("签名校验失败"); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,63 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.util; |
||||||
|
|
||||||
|
import java.util.List; |
||||||
|
import java.util.Map; |
||||||
|
import java.util.concurrent.ConcurrentHashMap; |
||||||
|
import java.util.concurrent.ThreadLocalRandom; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public class LoadBalanceUtil { |
||||||
|
|
||||||
|
/** |
||||||
|
* key：serviceId，value：指示变量i |
||||||
|
*/ |
||||||
|
private static Map<String, Integer> serviceIdRoundMap = new ConcurrentHashMap<>(8); |
||||||
|
|
||||||
|
/** |
||||||
|
* 轮询选择一台机器。<br> |
||||||
|
* <p> |
||||||
|
* 假设有N台服务器：S = {S1, S2, …, Sn}，一个指示变量i表示上一次选择的服务器ID。变量i被初始化为N-1。 |
||||||
|
* </p> |
||||||
|
* 参考：https://blog.csdn.net/qq_37469055/article/details/87991327
|
||||||
|
* @param serviceId serviceId，不同的serviceId对应的服务器数量不一样，需要区分开 |
||||||
|
* @param servers 服务器列表 |
||||||
|
* @return 返回一台服务器实例 |
||||||
|
*/ |
||||||
|
public static <T> T chooseByRoundRobin(String serviceId, List<T> servers) { |
||||||
|
if (servers == null || servers.isEmpty()) { |
||||||
|
return null; |
||||||
|
} |
||||||
|
int n = servers.size(); |
||||||
|
int i = serviceIdRoundMap.computeIfAbsent(serviceId, (k) -> n - 1); |
||||||
|
int j = i; |
||||||
|
do { |
||||||
|
j = (j + 1) % n; |
||||||
|
i = j; |
||||||
|
serviceIdRoundMap.put(serviceId, i); |
||||||
|
return servers.get(i); |
||||||
|
} while (j != i); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* 随机选取一台实例 |
||||||
|
* |
||||||
|
* @param servers 服务列表 |
||||||
|
* @return 返回实例，没有返回null |
||||||
|
*/ |
||||||
|
public static <T> T chooseByRandom(List<T> servers) { |
||||||
|
if (servers.isEmpty()) { |
||||||
|
return null; |
||||||
|
} |
||||||
|
int serverCount = servers.size(); |
||||||
|
// 随机选取一台实例
|
||||||
|
int index = chooseRandomInt(serverCount); |
||||||
|
return servers.get(index); |
||||||
|
} |
||||||
|
|
||||||
|
private static int chooseRandomInt(int serverCount) { |
||||||
|
return ThreadLocalRandom.current().nextInt(serverCount); |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
@ -0,0 +1,58 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.util; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.bean.ApiConfig; |
||||||
|
import com.gitee.sop.gatewaycommon.bean.DefaultRouteInterceptorContext; |
||||||
|
import com.gitee.sop.gatewaycommon.interceptor.RouteInterceptor; |
||||||
|
import com.gitee.sop.gatewaycommon.interceptor.RouteInterceptorContext; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import lombok.extern.slf4j.Slf4j; |
||||||
|
|
||||||
|
import java.util.Collection; |
||||||
|
import java.util.Comparator; |
||||||
|
import java.util.List; |
||||||
|
import java.util.function.Consumer; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
@Slf4j |
||||||
|
public class RouteInterceptorUtil { |
||||||
|
|
||||||
|
public static void runPreRoute(Object requestContext, ApiParam param, Consumer<RouteInterceptorContext> saveContext) { |
||||||
|
DefaultRouteInterceptorContext defaultRouteInterceptorContext = new DefaultRouteInterceptorContext(); |
||||||
|
saveContext.accept(defaultRouteInterceptorContext); |
||||||
|
defaultRouteInterceptorContext.setBeginTimeMillis(System.currentTimeMillis()); |
||||||
|
defaultRouteInterceptorContext.setRequestContext(requestContext); |
||||||
|
defaultRouteInterceptorContext.setApiParam(param); |
||||||
|
getRouteInterceptors().forEach(routeInterceptor -> { |
||||||
|
if (routeInterceptor.match(defaultRouteInterceptorContext)) { |
||||||
|
routeInterceptor.preRoute(defaultRouteInterceptorContext); |
||||||
|
} |
||||||
|
}); |
||||||
|
} |
||||||
|
|
||||||
|
public static void runAfterRoute(RouteInterceptorContext routeInterceptorContext) { |
||||||
|
if (routeInterceptorContext == null) { |
||||||
|
return; |
||||||
|
} |
||||||
|
try { |
||||||
|
getRouteInterceptors().forEach(routeInterceptor -> { |
||||||
|
if (routeInterceptor.match(routeInterceptorContext)) { |
||||||
|
routeInterceptor.afterRoute(routeInterceptorContext); |
||||||
|
} |
||||||
|
}); |
||||||
|
} catch (Exception e) { |
||||||
|
log.error("执行路由拦截器异常, apiParam:{}", routeInterceptorContext.getApiParam().toJSONString()); |
||||||
|
} |
||||||
|
} |
||||||
|
|
||||||
|
public static List<RouteInterceptor> getRouteInterceptors() { |
||||||
|
return ApiConfig.getInstance().getRouteInterceptors(); |
||||||
|
} |
||||||
|
|
||||||
|
public static void addInterceptors(Collection<RouteInterceptor> interceptors) { |
||||||
|
List<RouteInterceptor> routeInterceptors = getRouteInterceptors(); |
||||||
|
routeInterceptors.addAll(interceptors); |
||||||
|
routeInterceptors.sort(Comparator.comparing(RouteInterceptor::getOrder)); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,21 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon.zuul.controller; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.support.BaseMonitorController; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import com.gitee.sop.gatewaycommon.util.RequestUtil; |
||||||
|
import org.springframework.web.bind.annotation.RestController; |
||||||
|
|
||||||
|
import javax.servlet.http.HttpServletRequest; |
||||||
|
import java.util.Map; |
||||||
|
|
||||||
|
/** |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
@RestController |
||||||
|
public class ZuulMonitorController extends BaseMonitorController<HttpServletRequest> { |
||||||
|
@Override |
||||||
|
protected ApiParam getApiParam(HttpServletRequest request) { |
||||||
|
Map<String, String> params = RequestUtil.convertRequestParamsToMap(request); |
||||||
|
return ApiParam.build(params); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,62 @@ |
|||||||
|
package com.gitee.sop.gatewaycommon; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.util.LoadBalanceUtil; |
||||||
|
import junit.framework.TestCase; |
||||||
|
|
||||||
|
import java.util.ArrayList; |
||||||
|
import java.util.Arrays; |
||||||
|
import java.util.List; |
||||||
|
|
||||||
|
/** |
||||||
|
* 轮询选择一台机器。 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
public class RoundRobinTest extends TestCase { |
||||||
|
|
||||||
|
public void testDo() { |
||||||
|
String serviceId = "order-service"; |
||||||
|
List<String> serverList = new ArrayList<>(Arrays.asList("server1", "server2", "server3")); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
} |
||||||
|
|
||||||
|
public void testAdd() { |
||||||
|
String serviceId = "order-service"; |
||||||
|
List<String> serverList = new ArrayList<>(Arrays.asList("server1", "server2", "server3")); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
// 中途添加一个服务器
|
||||||
|
serverList.add("server4"); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
} |
||||||
|
|
||||||
|
public void testRemove() { |
||||||
|
String serviceId = "order-service"; |
||||||
|
List<String> serverList = new ArrayList<>(Arrays.asList("server1", "server2", "server3")); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
// 中途减少一台服务器
|
||||||
|
serverList.remove(2); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
System.out.println(chooseRoundRobinServer(serviceId, serverList)); |
||||||
|
} |
||||||
|
|
||||||
|
/** |
||||||
|
* 轮询选择一台机器。 |
||||||
|
* 假设有N台服务器：S = {S1, S2, …, Sn}，一个指示变量i表示上一次选择的服务器ID。变量i被初始化为N-1。 |
||||||
|
* |
||||||
|
* @param serviceId serviceId |
||||||
|
* @param servers 服务器列表 |
||||||
|
* @return 返回一台服务器实例 |
||||||
|
*/ |
||||||
|
private <T> T chooseRoundRobinServer(String serviceId, List<T> servers) { |
||||||
|
return LoadBalanceUtil.chooseByRoundRobin(serviceId, servers); |
||||||
|
} |
||||||
|
} |
||||||
@ -0,0 +1,32 @@ |
|||||||
|
package com.gitee.sop.gateway.interceptor; |
||||||
|
|
||||||
|
import com.gitee.sop.gatewaycommon.interceptor.RouteInterceptor; |
||||||
|
import com.gitee.sop.gatewaycommon.interceptor.RouteInterceptorContext; |
||||||
|
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||||
|
import org.springframework.stereotype.Component; |
||||||
|
|
||||||
|
/** |
||||||
|
* 演示拦截器 |
||||||
|
* |
||||||
|
* @author tanghc |
||||||
|
*/ |
||||||
|
@Component |
||||||
|
public class MyRouteInterceptor implements RouteInterceptor { |
||||||
|
|
||||||
|
@Override |
||||||
|
public void preRoute(RouteInterceptorContext context) { |
||||||
|
ApiParam apiParam = context.getApiParam(); |
||||||
|
System.out.println("请求接口：" + apiParam.fetchNameVersion()); |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public void afterRoute(RouteInterceptorContext context) { |
||||||
|
System.out.println("请求成功，微服务返回结果：" + context.getServiceResult()); |
||||||
|
} |
||||||
|
|
||||||
|
@Override |
||||||
|
public int getOrder() { |
||||||
|
return 0; |
||||||
|
} |
||||||
|
|
||||||
|
} |
||||||
Loading…
Reference in new issue