sentinel-reactor适配,限流计数有问题
- Dominant language
- Java
- Stars
- 23.1k
- Forks
- 8.1k
- PR merge metrics
- No merged PRs in 30d
Description
### 业务背景
使用sentinel对限制订阅着接受QPS,期望与自定义资源的限流策略相同
### 测试代码
@Test
public void testEmitMultipleValuesWhenFlowControlTriggered() {
String resourceName = createResourceName("testEmitMultipleValuesWhenFlowControlTriggered");
FlowRuleManager.loadRules(Collections.singletonList(
new FlowRule(resourceName).setCount(3)
));
StepVerifier.create(Flux.just(1, 3, 5, 7)
.map(e -> e * 2)
.transform(new SentinelReactorTransformer<>(resourceName)))
.expectNext(2)
.expectNext(6)
.expectNext(10)
.expectNext(14)
.verifyComplete();
ClusterNode cn = ClusterBuilderSlot.getClusterNode(resourceName);
assertNotNull(cn);
assertEquals(1, cn.passQps(), 0.01);
assertEquals(1, cn.totalRequest(), 0.01);
assertEquals(0, cn.blockRequest());
FlowRuleManager.loadRules(new ArrayList<>());
}
### 期望
这种配置方式,不应该是 totalRequest =4,blockRequest=1,Flux的订阅者也只能收到3条消息
### 问题分析
看源码只有在onSubscribe 时候才触发SphU.asyncEntry
onNext 时没有触发SphU.asyncEntry
private void entryWhenSubscribed() {
ContextConfig sentinelContextConfig = entryConfig.getContextConfig();
if (sentinelContextConfig != null) {
// If current we're already in a context, the context config won't work.
ContextUtil.enter(sentinelContextConfig.getContextName(), sentinelContextConfig.getOrigin());
}
try {
AsyncEntry entry = SphU.asyncEntry(entryConfig.getResourceName(), entryConfig.getResourceType(),
entryConfig.getEntryType(), entryConfig.getAcquireCount(), entryConfig.getArgs());
this.currentEntry = entry;
actual.onSubscribe(this);
} catch (BlockException ex) {
// Mark as completed (exited) explicitly.
entryExited.set(true);
// Signal cancel and propagate the {@code BlockException}.
cancel();
actual.onSubscribe(this);
actual.onError(ex);
} finally {
if (sentinelContextConfig != null) {
ContextUtil.exit();
}
}
}
Contributor guide
Research direction
Start with testEmitMultipleValuesWhenFlowControlTriggered and the SentinelReactorTransformer path, then inspect entryWhenSubscribed and its SphU.asyncEntry call. Verify how subscription and onNext events affect the counters and subscriber output. Done means the configured count limits the Flux to three values and the counters report totalRequest=4, blockRequest=1, and passQps=1.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100