使用任务执行器的事务轮询器是否将数据库资源用于队列消息

Question

我有一个服务激活器，它使用轮询器从通道中提取消息。通道有一个队列，该队列由数据库的持久存储支持。该轮询器还配置了一个任务执行器，以便为来自通道的消息处理添加一些并发性。

任务执行器配置有队列容量。

由于轮询器从数据库中的通道检索消息，并且这被配置为事务性的，那么如果任务执行器没有更多可用的线程，那么在任务执行器中排队的消息的事务会发生什么。任务执行器上的线程请求是排队的，由于这些消息没有自己的线程，事务会发生什么呢？我假设将提交由轮询器从持久通道存储区移除由任务执行器（在）排队的消息。因此，如果服务器在任务执行器中已排队的运行程序中出现故障，它们将丢失？

由于事务性持久通道队列的思想是确保在服务器发生故障时不会丢失消息，那么在通道数据库支持的队列/存储中，队列消息（在任务执行器中）的事务是如何被处理的呢？

    <bean id="store" class="org.springframework.integration.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="channelServerDataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
    <property name="region" value="${user.name}_${channel.queue.region:default}"/>
    <property name="usingIdCache" value="false"/>
</bean> 

<int:transaction-synchronization-factory id="syncFactory">
    <int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
    <int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/> 
</int:transaction-synchronization-factory>

<int:channel id="transacitonAsyncServiceQueue">
    <int:queue message-store="store"/> 
    <!--  <int:queue/>  --> 
</int:channel>

<bean id="rxPollingTrigger" class="org.springframework.scheduling.support.PeriodicTrigger">
    <constructor-arg value="500"/>
    <constructor-arg value="MILLISECONDS"/>
    <property name = "initialDelay" value = "30000"/> 
    <!-- initialDelay important to ensure channel doesnt start processing before the datasources have been initialised becuase we
         now persist transactions in the queue, at startup (restart) there maybe some ready to go which get processed before the
         connection pools have been created which happens when the servlet is first hit -->
</bean> 

<int:service-activator ref="asyncChannelReceiver" method="processMessage" input-channel="transacitonAsyncServiceQueue">
    <int:poller trigger="rxPollingTrigger" max-messages-per-poll="20"  task-executor="taskExecutor" receive-timeout="400">
        <int:transactional transaction-manager="transactionManagerAsyncChannel" /> 
    </int:poller>
    <int:request-handler-advice-chain>
        <ref bean="databaseSessionContext" />
    </int:request-handler-advice-chain>
</int:service-activator>

<task:executor id="taskExecutor" pool-size="100-200" queue-capacity="200" keep-alive="1" rejection-policy="CALLER_RUNS" />  

Answer 1

...然后，如果任务执行器没有更多可用线程，那么在任务执行器中排队的消息的事务将发生什么。任务执行器上的线程请求是排队的，由于这些消息没有自己的线程，事务会发生什么呢？

它不是这样工作的--直到任务执行（包括来自通道的receive())之后，事务才开始。

轮询器调度任务，任务启动，事务启动，工作进行，事务提交/回滚。

请参见abstractPollingEndpoint-pollingTask.call()是事务启动的地方。