如何解决Spring 集成 - 轮询器触发太频繁
@Bean
public IntegrationFlow aggregatingFlow(
AmqpInboundChannelAdapter aggregatorInboundAdapter,PollableChannel aggregatingChannel,AmqpOutboundEndpointEnhanced amqpOutboundEndpoint,PollSkipAdvice pollSkipAdvice) {
return IntegrationFlows.from(aggregatorInboundAdapter)
.wireTap(wtChannel())
.channel(aggregatingChannel)
.handle(
amqpOutboundEndpoint,e ->
e.poller(
Pollers.fixedDelay(1,TimeUnit.SECONDS,1)
.maxMessagesPerPoll(DEFAULT_MESSAGES_PER_POLL)
.advice(pollSkipAdvice))
.id("pollingConsumer"))
.get();
}
这里 pollSkipAdvice
被定义为
@Bean
public PollSkipAdvice pollSkipAdvice() {
return new PollSkipAdvice(
new PollSkipStrategy() {
int currentPoll = 1;
@Override
public synchronized boolean skipPoll() {
int hitRate = 0; //just not to add to the code; its dynamic
if (currentPoll >= hitRate) {
System.out.println(messageformat.format("{0} : Hit poll number {1} for message number of {2} | {3}",DateTimeFormatter.ofPattern("HH:mm:ss. SSS")
.withZone(ZoneOffset.UTC)
.format(Instant.Now()),currentPoll,currentMessagesInQueue,Thread.currentThread().getName()));
currentPoll = 1;
return false;
}
currentPolL++;
return true;
}
});
}
我面临的问题是 Poller 每秒被调用多次。这是我在测试代码时得到的日志:
21:41:47. 652 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:47. 654 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=190)
21:41:47. 656 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=191)
21:41:47. 658 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:47. 660 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=192)
21:41:47. 662 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=193)
21:41:47. 665 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=194)
21:41:47. 667 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=195)
21:41:47. 669 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:47. 671 : Hit poll number 1 for message number of 0 | pool-3-thread-1
EventChange(id=196)
EventChange(id=197)
EventChange(id=198)
EventChange(id=199)
21:41:48. 653 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:49. 656 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:50. 659 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:51. 661 : Hit poll number 1 for message number of 0 | pool-3-thread-1
21:41:52. 662 : Hit poll number 1 for message number of 0 | pool-3-thread-1
所以似乎每当我从 AMQP 收到一些东西时,这个 Poller 就会触发很多次。您可以在日志中看到 21:41:47. 652
到 21:41:47. 671
在 1 秒内触发了 10 次。另一方面,一旦它结束 - 它就会正常工作(每秒一次)。
当我尝试为这两种情况打印堆栈跟踪时,只有几毫秒的距离,我得到了这个:
20:32:56. 406 : Hit poll number 2 for message number of 0 | pool-3-thread-1
java.lang.Thread.getStackTrace(Thread.java:1559)
com.a.configuration.component.EventAggregatorConfig$1.skipPoll(EventAggregatorConfig.java:313)
org.springframework.integration.scheduling.PollSkipAdvice.invoke(PollSkipAdvice.java:51)
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
com.sun.proxy.$Proxy136.call(UnkNown Source)
org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:328)
org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:275)
org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)
20:32:56. 408 : Hit poll number 1 for message number of 0 | pool-3-thread-1
java.lang.Thread.getStackTrace(Thread.java:1559)
com.a.configuration.component.EventAggregatorConfig$1.skipPoll(EventAggregatorConfig.java:313)
org.springframework.integration.scheduling.PollSkipAdvice.invoke(PollSkipAdvice.java:51)
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:212)
com.sun.proxy.$Proxy136.call(UnkNown Source)
org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:328)
org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$null$1(AbstractPollingEndpoint.java:275)
org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:748)
我是否以错误的方式配置了它?我试图定义单独的 taskScheduler,但它似乎根本没有使用那个。
更新
大概经过一番分析,有可能是spring AbstractPollingEndpoint
类中的这部分造成的:
while (this.initialized && (this.maxMessagesPerPoll <= 0 || count < this.maxMessagesPerPoll)) {
if (pollForMessage() == null) {
break;
}
count++;
}
所以基本上即使我说每 1 秒轮询一次,如果有的话就取 10 条消息;但是如果没有消息,什么都不做 - 如果什么也没找到,它会继续尝试轮询?
更新2
是的,似乎每次消息轮询都会调用 skipPoll
建议,而不是 maxMessagesPerPoll
一次。
解决方法
我不确定问题是什么。逻辑绝对是我们从这条建议中所期望的:
private Runnable createPoller() {
return () ->
this.taskExecutor.execute(() -> {
int count = 0;
while (this.initialized && (this.maxMessagesPerPoll <= 0 || count < this.maxMessagesPerPoll)) {
if (this.maxMessagesPerPoll == 0) {
logger.info("Polling disabled while 'maxMessagesPerPoll == 0'");
break;
}
if (pollForMessage() == null) {
break;
}
count++;
}
});
}
因此,如果 pollForMessage()
返回 null
,我们将中断 while()
循环并退出当前轮询周期。当 null
返回 PollSkipAdvice
时,PollSkipStrategy
从 false
发生。
您可能只是误解了轮询周期并真正进行了轮询。第一个确实是触发器命中时间及其maxMessagesPerPoll
的时刻。真正的 poll
确实是关于单个 pollForMessage
调用。您可能需要修改自定义策略的逻辑,以满足应用此类建议的现有要求。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。