tanghc
6 years ago
parent
f8fd24ce32
commit
59a41de04e
@ -0,0 +1,67 @@ |
||||
package com.gitee.sop.gatewaycommon.gateway.filter; |
||||
|
||||
import com.gitee.sop.gatewaycommon.bean.ApiConfig; |
||||
import com.gitee.sop.gatewaycommon.bean.ApiContext; |
||||
import com.gitee.sop.gatewaycommon.bean.RouteConfig; |
||||
import com.gitee.sop.gatewaycommon.bean.SopConstants; |
||||
import com.gitee.sop.gatewaycommon.exception.ApiException; |
||||
import com.gitee.sop.gatewaycommon.limit.LimitManager; |
||||
import com.gitee.sop.gatewaycommon.limit.LimitType; |
||||
import com.gitee.sop.gatewaycommon.manager.RouteConfigManager; |
||||
import com.gitee.sop.gatewaycommon.message.ErrorImpl; |
||||
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||
import com.gitee.sop.gatewaycommon.param.ParamNames; |
||||
import com.gitee.sop.gatewaycommon.util.RouteUtil; |
||||
import com.gitee.sop.gatewaycommon.validate.Validator; |
||||
import com.gitee.sop.gatewaycommon.zuul.ZuulContext; |
||||
import lombok.extern.slf4j.Slf4j; |
||||
import org.springframework.cloud.gateway.filter.GatewayFilterChain; |
||||
import org.springframework.cloud.gateway.filter.GlobalFilter; |
||||
import org.springframework.core.Ordered; |
||||
import org.springframework.web.server.ServerWebExchange; |
||||
import reactor.core.publisher.Mono; |
||||
|
||||
import java.util.Map; |
||||
|
||||
/** |
||||
* @author tanghc |
||||
*/ |
||||
@Slf4j |
||||
public class LimitFilter implements GlobalFilter, Ordered { |
||||
|
||||
@Override |
||||
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) { |
||||
ApiConfig apiConfig = ApiConfig.getInstance(); |
||||
// 限流功能未开启，直接返回
|
||||
if (!apiConfig.isOpenLimit()) { |
||||
return chain.filter(exchange); |
||||
} |
||||
Map<String, ?> apiParam = exchange.getAttribute(SopConstants.CACHE_API_PARAM); |
||||
String routeId = apiParam.get(ParamNames.API_NAME).toString() + apiParam.get(ParamNames.VERSION_NAME); |
||||
RouteConfigManager routeConfigManager = apiConfig.getRouteConfigManager(); |
||||
RouteConfig routeConfig = routeConfigManager.get(routeId); |
||||
if (routeConfig == null) { |
||||
return chain.filter(exchange); |
||||
} |
||||
// 某个路由限流功能未开启
|
||||
if (routeConfig.getLimitStatus() == RouteConfig.LIMIT_STATUS_CLOSE) { |
||||
return chain.filter(exchange); |
||||
} |
||||
byte limitType = routeConfig.getLimitType().byteValue(); |
||||
LimitManager limitManager = ApiConfig.getInstance().getLimitManager(); |
||||
if (limitType == LimitType.LEAKY_BUCKET.getType()) { |
||||
boolean acquire = limitManager.acquire(routeConfig); |
||||
if (!acquire) { |
||||
throw new ApiException(new ErrorImpl(routeConfig.getLimitCode(), routeConfig.getLimitMsg())); |
||||
} |
||||
} else if (limitType == LimitType.TOKEN_BUCKET.getType()) { |
||||
limitManager.acquireToken(routeConfig); |
||||
} |
||||
return chain.filter(exchange); |
||||
} |
||||
|
||||
@Override |
||||
public int getOrder() { |
||||
return Orders.LIMIT_ORDER; |
||||
} |
||||
} |
||||
@ -0,0 +1,14 @@ |
||||
package com.gitee.sop.gatewaycommon.gateway.filter; |
||||
|
||||
import org.springframework.core.Ordered; |
||||
|
||||
/** |
||||
* @author tanghc |
||||
*/ |
||||
public class Orders { |
||||
/** 验证拦截器order */ |
||||
public static final int VALIDATE_ORDER = Ordered.HIGHEST_PRECEDENCE + 1000; |
||||
|
||||
/** 验证拦截器order */ |
||||
public static final int LIMIT_ORDER = VALIDATE_ORDER + 1; |
||||
} |
||||
@ -0,0 +1,52 @@ |
||||
package com.gitee.sop.gatewaycommon.limit; |
||||
|
||||
import com.gitee.sop.gatewaycommon.bean.RouteConfig; |
||||
import com.google.common.cache.LoadingCache; |
||||
import lombok.extern.slf4j.Slf4j; |
||||
|
||||
import java.util.concurrent.ExecutionException; |
||||
import java.util.concurrent.atomic.AtomicLong; |
||||
|
||||
/** |
||||
* @author tanghc |
||||
*/ |
||||
@Slf4j |
||||
public class DefaultLimitManager implements LimitManager { |
||||
|
||||
@Override |
||||
public double acquireToken(RouteConfig routeConfig) { |
||||
if (routeConfig.getLimitStatus() == RouteConfig.LIMIT_STATUS_CLOSE) { |
||||
return 0; |
||||
} |
||||
if (LimitType.LEAKY_BUCKET.getType() == routeConfig.getLimitType().byteValue()) { |
||||
throw new IllegalStateException("漏桶策略无法调用此方法"); |
||||
} |
||||
return routeConfig.fetchRateLimiter().acquire(); |
||||
} |
||||
|
||||
|
||||
@Override |
||||
public boolean acquire(RouteConfig routeConfig) { |
||||
if (routeConfig.getLimitStatus() == RouteConfig.LIMIT_STATUS_CLOSE) { |
||||
return true; |
||||
} |
||||
if (LimitType.TOKEN_BUCKET.getType() == routeConfig.getLimitType().byteValue()) { |
||||
throw new IllegalStateException("令牌桶策略无法调用此方法"); |
||||
} |
||||
int execCountPerSecond = routeConfig.getExecCountPerSecond(); |
||||
long currentSeconds = System.currentTimeMillis() / 1000; |
||||
try { |
||||
LoadingCache<Long, AtomicLong> counter = routeConfig.getCounter(); |
||||
// 被限流了
|
||||
if (counter.get(currentSeconds).incrementAndGet() > execCountPerSecond) { |
||||
return false; |
||||
} else { |
||||
return true; |
||||
} |
||||
} catch (ExecutionException e) { |
||||
log.error("漏桶限流出错，routeConfig", routeConfig, e); |
||||
return false; |
||||
} |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,29 @@ |
||||
package com.gitee.sop.gatewaycommon.limit; |
||||
|
||||
import com.gitee.sop.gatewaycommon.bean.RouteConfig; |
||||
|
||||
/** |
||||
* 限流管理 |
||||
* @author tanghc |
||||
*/ |
||||
public interface LimitManager { |
||||
|
||||
/** |
||||
* 从令牌桶中获取令牌，如果使用{@link LimitType#TOKEN_BUCKET |
||||
* RateType.TOKEN_BUCKET}限流策略，则该方法生效 |
||||
* |
||||
* @param routeConfig 路由配置 |
||||
* @return 返回耗时时间，秒 |
||||
*/ |
||||
double acquireToken(RouteConfig routeConfig); |
||||
|
||||
/** |
||||
* 是否需要限流，如果使用{@link LimitType#LEAKY_BUCKET |
||||
* RateType.LIMIT}限流策略，则该方法生效 |
||||
* |
||||
* @param routeConfig 路由配置 |
||||
* @return 如果返回true，表示可以执行业务代码，返回false则需要限流 |
||||
*/ |
||||
boolean acquire(RouteConfig routeConfig); |
||||
|
||||
} |
||||
@ -0,0 +1,28 @@ |
||||
package com.gitee.sop.gatewaycommon.limit; |
||||
|
||||
/** |
||||
* 限流策略 |
||||
* |
||||
* @author tanghc |
||||
*/ |
||||
public enum LimitType { |
||||
/** |
||||
* 漏桶策略。每秒处理固定数量的请求，超出请求返回错误信息。 |
||||
*/ |
||||
LEAKY_BUCKET(1), |
||||
/** |
||||
* 令牌桶策略，每秒放置固定数量的令牌数，不足的令牌数做等待处理，直到拿到令牌为止。 |
||||
*/ |
||||
TOKEN_BUCKET(2); |
||||
|
||||
private byte type; |
||||
|
||||
LimitType(int type) { |
||||
this.type = (byte)type; |
||||
} |
||||
|
||||
public byte getType() { |
||||
return type; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,62 @@ |
||||
package com.gitee.sop.gatewaycommon.zuul.filter; |
||||
|
||||
import com.gitee.sop.gatewaycommon.bean.ApiConfig; |
||||
import com.gitee.sop.gatewaycommon.bean.RouteConfig; |
||||
import com.gitee.sop.gatewaycommon.bean.TargetRoute; |
||||
import com.gitee.sop.gatewaycommon.exception.ApiException; |
||||
import com.gitee.sop.gatewaycommon.limit.LimitManager; |
||||
import com.gitee.sop.gatewaycommon.limit.LimitType; |
||||
import com.gitee.sop.gatewaycommon.manager.RouteConfigManager; |
||||
import com.gitee.sop.gatewaycommon.manager.RouteRepositoryContext; |
||||
import com.gitee.sop.gatewaycommon.message.ErrorImpl; |
||||
import com.gitee.sop.gatewaycommon.param.ApiParam; |
||||
import com.gitee.sop.gatewaycommon.zuul.ZuulContext; |
||||
import com.netflix.zuul.context.RequestContext; |
||||
import com.netflix.zuul.exception.ZuulException; |
||||
|
||||
/** |
||||
* 限流拦截器 |
||||
* @author tanghc |
||||
*/ |
||||
public class PreLimitFilter extends BaseZuulFilter { |
||||
@Override |
||||
protected FilterType getFilterType() { |
||||
return FilterType.PRE; |
||||
} |
||||
|
||||
@Override |
||||
protected int getFilterOrder() { |
||||
return PRE_LIMIT_FILTER_ORDER; |
||||
} |
||||
|
||||
@Override |
||||
protected Object doRun(RequestContext requestContext) throws ZuulException { |
||||
ApiConfig apiConfig = ApiConfig.getInstance(); |
||||
// 限流功能未开启，直接返回
|
||||
if (!apiConfig.isOpenLimit()) { |
||||
return null; |
||||
} |
||||
ApiParam apiParam = ZuulContext.getApiParam(); |
||||
String routeId = apiParam.getRouteId(); |
||||
RouteConfigManager routeConfigManager = apiConfig.getRouteConfigManager(); |
||||
RouteConfig routeConfig = routeConfigManager.get(routeId); |
||||
if (routeConfig == null) { |
||||
return null; |
||||
} |
||||
// 某个路由限流功能未开启
|
||||
if (routeConfig.getLimitStatus() == RouteConfig.LIMIT_STATUS_CLOSE) { |
||||
return null; |
||||
} |
||||
byte limitType = routeConfig.getLimitType().byteValue(); |
||||
LimitManager limitManager = ApiConfig.getInstance().getLimitManager(); |
||||
if (limitType == LimitType.LEAKY_BUCKET.getType()) { |
||||
boolean acquire = limitManager.acquire(routeConfig); |
||||
if (!acquire) { |
||||
throw new ApiException(new ErrorImpl(routeConfig.getLimitCode(), routeConfig.getLimitMsg())); |
||||
} |
||||
} else if (limitType == LimitType.TOKEN_BUCKET.getType()) { |
||||
limitManager.acquireToken(routeConfig); |
||||
} |
||||
return null; |
||||
} |
||||
} |
||||
@ -0,0 +1,80 @@ |
||||
package com.gitee.sop; |
||||
|
||||
import com.alibaba.fastjson.JSON; |
||||
import com.gitee.sop.alipay.AlipaySignature; |
||||
import org.junit.Test; |
||||
|
||||
import java.text.SimpleDateFormat; |
||||
import java.util.Date; |
||||
import java.util.HashMap; |
||||
import java.util.Map; |
||||
import java.util.concurrent.CountDownLatch; |
||||
import java.util.concurrent.atomic.AtomicInteger; |
||||
|
||||
/** |
||||
* 限流测试 |
||||
*/ |
||||
public class LimitDemoPostTest extends TestBase { |
||||
|
||||
String url = "http://localhost:8081/api"; // zuul
|
||||
String appId = "2019032617262200001"; |
||||
// 支付宝私钥
|
||||
String privateKey = "MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQCXJv1pQFqWNA/++OYEV7WYXwexZK/J8LY1OWlP9X0T6wHFOvxNKRvMkJ5544SbgsJpVcvRDPrcxmhPbi/sAhdO4x2PiPKIz9Yni2OtYCCeaiE056B+e1O2jXoLeXbfi9fPivJZkxH/tb4xfLkH3bA8ZAQnQsoXA0SguykMRZntF0TndUfvDrLqwhlR8r5iRdZLB6F8o8qXH6UPDfNEnf/K8wX5T4EB1b8x8QJ7Ua4GcIUqeUxGHdQpzNbJdaQvoi06lgccmL+PHzminkFYON7alj1CjDN833j7QMHdPtS9l7B67fOU/p2LAAkPMtoVBfxQt9aFj7B8rEhGCz02iJIBAgMBAAECggEARqOuIpY0v6WtJBfmR3lGIOOokLrhfJrGTLF8CiZMQha+SRJ7/wOLPlsH9SbjPlopyViTXCuYwbzn2tdABigkBHYXxpDV6CJZjzmRZ+FY3S/0POlTFElGojYUJ3CooWiVfyUMhdg5vSuOq0oCny53woFrf32zPHYGiKdvU5Djku1onbDU0Lw8w+5tguuEZ76kZ/lUcccGy5978FFmYpzY/65RHCpvLiLqYyWTtaNT1aQ/9pw4jX9HO9NfdJ9gYFK8r/2f36ZE4hxluAfeOXQfRC/WhPmiw/ReUhxPznG/WgKaa/OaRtAx3inbQ+JuCND7uuKeRe4osP2jLPHPP6AUwQKBgQDUNu3BkLoKaimjGOjCTAwtp71g1oo+k5/uEInAo7lyEwpV0EuUMwLA/HCqUgR4K9pyYV+Oyb8d6f0+Hz0BMD92I2pqlXrD7xV2WzDvyXM3s63NvorRooKcyfd9i6ccMjAyTR2qfLkxv0hlbBbsPHz4BbU63xhTJp3Ghi0/ey/1HQKBgQC2VsgqC6ykfSidZUNLmQZe3J0p/Qf9VLkfrQ+xaHapOs6AzDU2H2osuysqXTLJHsGfrwVaTs00ER2z8ljTJPBUtNtOLrwNRlvgdnzyVAKHfOgDBGwJgiwpeE9voB1oAV/mXqSaUWNnuwlOIhvQEBwekqNyWvhLqC7nCAIhj3yvNQKBgQCqYbeec56LAhWP903Zwcj9VvG7sESqXUhIkUqoOkuIBTWFFIm54QLTA1tJxDQGb98heoCIWf5x/A3xNI98RsqNBX5JON6qNWjb7/dobitti3t99v/ptDp9u8JTMC7penoryLKK0Ty3bkan95Kn9SC42YxaSghzqkt+uvfVQgiNGQKBgGxU6P2aDAt6VNwWosHSe+d2WWXt8IZBhO9d6dn0f7ORvcjmCqNKTNGgrkewMZEuVcliueJquR47IROdY8qmwqcBAN7Vg2K7r7CPlTKAWTRYMJxCT1Hi5gwJb+CZF3+IeYqsJk2NF2s0w5WJTE70k1BSvQsfIzAIDz2yE1oPHvwVAoGAA6e+xQkVH4fMEph55RJIZ5goI4Y76BSvt2N5OKZKd4HtaV+eIhM3SDsVYRLIm9ZquJHMiZQGyUGnsvrKL6AAVNK7eQZCRDk9KQz+0GKOGqku0nOZjUbAu6A2/vtXAaAuFSFx1rUQVVjFulLexkXR3KcztL1Qu2k5pB6Si0K/uwQ="; |
||||
|
||||
@Test |
||||
public void testLimit() throws InterruptedException { |
||||
int threadsCount = 10; // threadsCount个线程同时提交
|
||||
final CountDownLatch countDownLatch = new CountDownLatch(1); |
||||
final CountDownLatch count = new CountDownLatch(threadsCount); |
||||
final AtomicInteger success = new AtomicInteger(); |
||||
for (int i = 0; i < threadsCount; i++) { |
||||
new Thread(new Runnable() { |
||||
@Override |
||||
public void run() { |
||||
try { |
||||
countDownLatch.await(); // 等在这里，执行countDownLatch.countDown();集体触发
|
||||
// 业务方法
|
||||
doBusiness(Thread.currentThread().getName()); |
||||
success.incrementAndGet(); |
||||
} catch (Exception e) { |
||||
} finally { |
||||
count.countDown(); |
||||
} |
||||
} |
||||
}).start(); |
||||
} |
||||
countDownLatch.countDown(); |
||||
count.await(); |
||||
System.out.println("成功次数：" + success); |
||||
} |
||||
|
||||
// 这个请求会路由到story服务
|
||||
public void doBusiness(String threadName) throws Exception { |
||||
|
||||
// 公共请求参数
|
||||
Map<String, String> params = new HashMap<String, String>(); |
||||
params.put("app_id", appId); |
||||
params.put("method", "alipay.story.get"); |
||||
params.put("format", "json"); |
||||
params.put("charset", "utf-8"); |
||||
params.put("sign_type", "RSA2"); |
||||
params.put("timestamp", new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())); |
||||
params.put("version", "1.2"); |
||||
|
||||
// 业务参数
|
||||
Map<String, String> bizContent = new HashMap<>(); |
||||
bizContent.put("id", "1"); |
||||
bizContent.put("name", "葫芦娃"); |
||||
|
||||
params.put("biz_content", JSON.toJSONString(bizContent)); |
||||
|
||||
String content = AlipaySignature.getSignContent(params); |
||||
String sign = AlipaySignature.rsa256Sign(content, privateKey, "utf-8"); |
||||
|
||||
params.put("sign", sign); |
||||
|
||||
String responseData = post(url, params);// 发送请求
|
||||
System.out.println(responseData); |
||||
} |
||||
|
||||
} |
||||
Loading…
Reference in new issue