Spring集成轮询器以错误的时间间隔触发

2023-12-23 18:16:15

代码：https://github.com/giuliopulina/spring-integration-poller

尝试使用Spring集成创建jdbc轮询器时遇到问题.

当我用新数据提供表时,处理速度比预期慢：一切正常,除了每隔60秒触发一次轮询,我无法理解为什么.

2015-05-27 10：50：40,234 DEBUG ExpressionEvaluatingSqlParameterSourceFactory – 已解决的表达式#root.！[pk] to(pks列表)

2015-05-27 10：51：40,234 DEBUG ExpressionEvaluatingSqlParameterSourceFactory – 已解决的表达式#root.！[pk] to(pks列表)

这是spring集成配置xml的相关部分：

<task:executor id="pollerPool" pool-size="5-20" queue-capacity="0" rejection-policy="CALLER_RUNS" keep-alive="5"/>

<!--<task:executor id="processingPool" pool-size="5-20" queue-capacity="0" rejection-policy="CALLER_RUNS" keep-alive="5"/> -->

<bean id="jdbcSource" class="org.springframework.integration.jdbc.JdbcPollingChannelAdapter">
    <constructor-arg ref="dataSource"/>
    <constructor-arg value="XXXXXXXXXXXXXX"/>
    <property name="updateSql" value="XXXXXXXXXXXXXXXX"/>
    <property name="maxRowsPerPoll" value="50"/>
</bean>

<int:inbound-channel-adapter send-timeout="10000" auto-startup="false" id="inboundAdapter" ref="jdbcSource" channel="jdbcOutputChannel">
    <int:poller receive-timeout="3000" time-unit="MILLISECONDS" fixed-rate="0" error-channel="errorChannel" task-executor="pollerPool">
        <int:advice-chain>
            <ref bean="threadPrepareInterceptor"/>
            <ref bean="txAdvice"/>
        </int:advice-chain>
    </int:poller>
</int:inbound-channel-adapter>

<int:service-activator id="serviceActivator" input-channel="jdbcOutputChannel" ref="someServiceActivatorBean"/>

<tx:advice id="txAdvice" transaction-manager="txManager">
    <tx:attributes>
        <tx:method name="get*" read-only="true"/>
        <tx:method name="*"/>
    </tx:attributes>
</tx:advice>

<int:channel id="jdbcOutputChannel"  >
    <!-- using direct channel -->
    <!--<int:dispatcher task-executor="processingPool"/>-->
</int:channel>

你能帮我理解一下这个问题吗？

更新：

关于“jdbcOutputChannel”关于事务的建议,我同意并且我根据你的提示修改了我的配置,因为它更干净(无论如何,服务激活器也在单独的事务中运行,即使在xml示例中没有概述).

关于我的问题,我试图删除所有其他弹簧集成组件,并按照我的预期连续触发轮询器(我知道固定速率= 0太高了:))
相反,当项目中的其他轮询器配置如此时,我的轮询器似乎继承了相同的超时：

<int:service-activator id="someOtherServiceActivator">
    <int:poller fixed-rate="0" error-channel="someOtherPollerErrorChannel" receive-timeout="60000" />
</int:service-activator>

将其他轮询器的超时切换为10000ms,我的轮询器每10秒触发一次(而不是60秒).
我无法分享完整的spring集成配置,但我想问一下：完全分离的poller可以修改彼此的行为吗？

更新2：
我创建了一个试图重现问题的独立项目,但我仍无法做到这一点.
因此,我尝试删除以下配置,它是为了仅在应用程序完全启动并运行时启动轮询器而引入的：

<int:publish-subscribe-channel id="startupChannel" />
<int:control-bus input-channel="controlBusChannel" />

<int-event:inbound-channel-adapter channel="startupChannel" event-types="org.springframework.context.event.ContextRefreshedEvent"/>

<int:transformer input-channel="startupChannel" expression="'@inboundAdapter.start()'" output-channel="controlBusChannel" />
<int:transformer input-channel="startupChannel" expression="'@someOtherServiceActivator.start()'" output-channel="controlBusChannel" />

问题已经消失,即使我能完全理解其中的原因.
无论如何,为我的poller创建一个不同的startupChannel可以完美地工作：

<int:publish-subscribe-channel id="globalStartupChannel" />
<int:publish-subscribe-channel id="myStartupChannel" />
<int:control-bus input-channel="controlBusChannel" />

<int-event:inbound-channel-adapter channel="globalStartupChannel" event-types="org.springframework.context.event.ContextRefreshedEvent"/>
<int-event:inbound-channel-adapter channel="myStartupChannel" event-types="org.springframework.context.event.ContextRefreshedEvent"/>

<int:transformer input-channel="globalStartupChannel" expression="'@someOtherServiceActivator.start()'" output-channel="controlBusChannel" />
<int:transformer input-channel="myStartupChannel" expression="'@inboundAdapter.start()'" output-channel="controlBusChannel" />

更新3：

在为您准备项目代码时,我注意到以下日志：

信息：没有明确定义名为’taskScheduler’的bean.因此,将创建默认的ThreadPoolTaskScheduler.

所以,我添加了以下配置,现在一切正常：

<task:scheduler id="taskScheduler" pool-size="20" />

我猜默认的池大小是10,因此在使用totalNumberOfPollers>时会以某种方式覆盖配置. taskScheduler.size().
我对吗？

谢谢
朱利奥

解决方法:

我不能重现你的情况;我建议你在民意调查之间进行一次线程转储,看看线程在做什么.

也就是说,0的固定利率非常具有攻击性;你的DBA可能会抛出一个合适的轮询,没有这样的延迟.

此外,作为ExecutorChannel的jdbcOutputChannel表示事务将在将消息发送到该通道后立即提交.如果您希望流在事务中运行,则不应在此处使用调度程序.

编辑：

我仍然无法用这个重现你的情况……

<int:control-bus input-channel="input"/>

<int-event:inbound-channel-adapter channel="ps" event-types="org.springframework.context.event.ContextRefreshedEvent"/>

<int:publish-subscribe-channel id="ps" />

<int:transformer input-channel="ps" output-channel="input" expression="'@foo.start()'" />

<int:transformer input-channel="ps" output-channel="input" expression="'@sa.start()'" />

<int:inbound-channel-adapter id="foo" channel="bar" expression="'foo'" auto-startup="false">
    <int:poller fixed-rate="1000" />
</int:inbound-channel-adapter>

<int:channel id="bar">
    <int:queue />
</int:channel>

<int:service-activator id="sa" input-channel="bar" output-channel="baz" auto-startup="false"
        expression="payload.toUpperCase()">
    <int:poller fixed-rate="6000" receive-timeout="0" />
</int:service-activator>

<int:logging-channel-adapter id="baz" level="ERROR"/>

……正如预期的那样,我每6秒看到6个FOO(i-c-a每秒轮询一次,sa每6秒运行一次).

EDIT2：

我看了你的项目,问题的根本原因就像你说的那样,很多轮询端点,但实际上,就是这样：

fixed-rate="0" receive-timeout="60000"

使用此配置,调度程序资源(线程)在QueueChannels中被阻止,正如您所发现的那样,您已经耗尽了所有资源.

一种解决方案是增加调度程序池中的线程数.

通过这种配置,您似乎正在尝试通过让轮询器不断在队列中等待receive()方法来获得带轮询器的按需,零延迟消息传递.

如果您无法承受任何延迟,请考虑使用DirectChannel.如果您不希望下游端点在调用者的线程上运行,请使用ExecutorChannels …

<task:executor id="exec" pool-size="100"/>

<int:channel id="otherMessageChannel1">
    <int:dispatcher task-executor="exec" />
</int:channel>

这通常比您当前的设置更受欢迎.

码农公寓

相关文章