fizzgate/fizz-gateway-node
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

optimize: scheduler executors initialization

Browse Source
This commit is contained in:
hongqiaowei
2021-01-12 15:16:35 +08:00
parent d0e3a8e8c7
commit 44bf38fa4e
3 changed files with 9 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
src/main/java/we/config/FlowStatSchedConfig.java
View File

@@ -99,6 +99,8 @@ public class FlowStatSchedConfig extends SchedConfig {
private long startTimeSlot = 0; private long startTimeSlot = 0;
private Map<String, AtomicLong> key2totalBlockMap = new HashMap<>();
@Scheduled(cron = "${flow-stat-sched.cron}") @Scheduled(cron = "${flow-stat-sched.cron}")
public void sched() { public void sched() {
@@ -118,11 +120,11 @@ public class FlowStatSchedConfig extends SchedConfig {
return; return;
} }
Map<String, AtomicLong> key2TotalBlockMap = new HashMap<>(8); key2totalBlockMap.clear();
resourceTimeWindowStats.forEach(rtws -> { resourceTimeWindowStats.forEach(rtws -> {
List<TimeWindowStat> wins = rtws.getWindows(); List<TimeWindowStat> wins = rtws.getWindows();
wins.forEach(w -> { wins.forEach(w -> {
AtomicLong totalBlock = key2TotalBlockMap.computeIfAbsent(String.format("%s%s", AtomicLong totalBlock = key2totalBlockMap.computeIfAbsent(String.format("%s%s",
ResourceRateLimitConfig.GLOBAL, w.getStartTime()), key -> new AtomicLong(0)); ResourceRateLimitConfig.GLOBAL, w.getStartTime()), key -> new AtomicLong(0));
totalBlock.addAndGet(w.getBlockRequests()); totalBlock.addAndGet(w.getBlockRequests());
}); });
@@ -154,7 +156,7 @@ public class FlowStatSchedConfig extends SchedConfig {
qps = rps.doubleValue(); qps = rps.doubleValue();
} }
AtomicLong totalBlock = key2TotalBlockMap.get(String.format("%s%s", resource, winStart)); AtomicLong totalBlock = key2totalBlockMap.get(String.format("%s%s", resource, winStart));
Long totalBlockReqs = totalBlock != null ? totalBlock.get() : w.getBlockRequests(); Long totalBlockReqs = totalBlock != null ? totalBlock.get() : w.getBlockRequests();
b.append(Constants.Symbol.LEFT_BRACE); b.append(Constants.Symbol.LEFT_BRACE);
@@ -176,7 +178,7 @@ public class FlowStatSchedConfig extends SchedConfig {
b.append(Constants.Symbol.RIGHT_BRACE); b.append(Constants.Symbol.RIGHT_BRACE);
String msg = b.toString(); String msg = b.toString();
if ("kafka".equals(dest)) { if ("kafka".equals(dest)) {
log.info(msg, LogService.HANDLE_STGY, LogService.toKF(queue)); log.warn(msg, LogService.HANDLE_STGY, LogService.toKF(queue));
} else { } else {
rt.convertAndSend(queue, msg).subscribe(); rt.convertAndSend(queue, msg).subscribe();
} }

2
src/main/java/we/controller/FlowControlController.java
View File

@@ -66,10 +66,8 @@ public class FlowControlController {
try { try {
FlowStat flowStat = flowControlFilter.getFlowStat(); FlowStat flowStat = flowControlFilter.getFlowStat();
long currentTimeSlot = flowStat.currentTimeSlotId(); long currentTimeSlot = flowStat.currentTimeSlotId();
// recent = 30;
long startTimeSlot = currentTimeSlot - recent * 1000; long startTimeSlot = currentTimeSlot - recent * 1000;
TimeWindowStat timeWindowStat = null; TimeWindowStat timeWindowStat = null;
// System.err.println("startTimeSlot: " + startTimeSlot + ", currentTimeSlot: " + currentTimeSlot + ", recent: " + recent);
List<ResourceTimeWindowStat> wins = flowStat.getResourceTimeWindowStats(ResourceRateLimitConfig.GLOBAL, startTimeSlot, currentTimeSlot, recent); List<ResourceTimeWindowStat> wins = flowStat.getResourceTimeWindowStats(ResourceRateLimitConfig.GLOBAL, startTimeSlot, currentTimeSlot, recent);
if (wins == null || wins.isEmpty()) { if (wins == null || wins.isEmpty()) {
result.put("rps", 0); result.put("rps", 0);

6
src/main/java/we/filter/FlowControlFilter.java
View File

@@ -115,8 +115,10 @@ public class FlowControlFilter extends ProxyAggrFilter {
if (concurrentOrRpsExceed) { if (concurrentOrRpsExceed) {
if (rlc.type == ResourceRateLimitConfig.Type.API || rlc.type == ResourceRateLimitConfig.Type.SERVICE || rlc.type == ResourceRateLimitConfig.Type.SERVICE_DEFAULT) { if (rlc.type == ResourceRateLimitConfig.Type.API || rlc.type == ResourceRateLimitConfig.Type.SERVICE || rlc.type == ResourceRateLimitConfig.Type.SERVICE_DEFAULT) {
if (globalConfig != null && globalConfig.isEnable()) {
flowStat.decrConcurrentRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot); flowStat.decrConcurrentRequest(ResourceRateLimitConfig.GLOBAL, currentTimeSlot);
} }
}
StringBuilder b = ThreadContext.getStringBuilder(); StringBuilder b = ThreadContext.getStringBuilder();
b.append(WebUtils.getClientService(exchange)).append(Constants.Symbol.SPACE).append(WebUtils.getClientReqPath(exchange)); b.append(WebUtils.getClientService(exchange)).append(Constants.Symbol.SPACE).append(WebUtils.getClientReqPath(exchange));
@@ -146,10 +148,6 @@ public class FlowControlFilter extends ProxyAggrFilter {
() -> { () -> {
inTheEnd(exchange, start, currentTimeSlot, false); inTheEnd(exchange, start, currentTimeSlot, false);
} }
)
.doOnDiscard(
Object.class,
o -> { inTheEnd(exchange, start, currentTimeSlot, false); }
); );
} }
} }