Spring消息驱动的通道适配器-停止一个特定的并发使用者

Question

我使用spring集成组件连接我的流程。

• 消息驱动的通道适配器从weblogic队列中选择消息。
• 使用消息转换器并将消息删除到通道。
• 来自通道的此消息由服务激活器选择。
• pojo on service激活器，从db进行一些数据处理，然后生成一个报告。
• 出于某些原因，我想停止此消费者处理此请求的原因如下(假设pojo处理时间过长)

我有什么选择？

• 我试过这些：
  • 通过JMX公开了spring的消息驱动通道适配器。
  • 在消息驱动的通道适配器上使用停止方法。
  • 这将阻止消息驱动的通道适配器选择新消息，这是很好的。
  • 但是并没有停止一个消费者已经选择的信息，并且目前正在处理中。
  • 我怎样才能阻止这个在我的pojo中花费太长时间的过程呢？

​

感谢你的帮助。

更新-2

这是我的密码。

<bean id="myListener" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
    <property name="destination" ref="requestQueue"/>
    <property name="concurrentConsumers" value="5"/>
    <property name="maxConcurrentConsumers" value="10"/>
    <property name="connectionFactory" ref="connectionFactory"/>
    <property name="taskExecutor" ref="threadPoolTaskExecutor"/>
    <property name="sessionTransacted" value="true"/>
</bean> 

    <bean id="threadPoolTaskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
    <property name="corePoolSize" value="10" />
    <property name="maxPoolSize" value="50" />
    <property name="queueCapacity" value="10" />
    <property name="waitForTasksToCompleteOnShutdown" value="false"/>
</bean> 

<!-- Channel where service activator drops message to -->
<si:channel id="jmsOutChannel" />

<!-- This bean code calls DB to get some data and build a report/-->
<bean id="simpleExecutor" class="com.poc.reports.executors.SimpleCode"/>

<!-- Initialize service activator -->
<si:service-activator id="activator" input-channel="jmsInChannel"
    ref="simpleExecutor" method="execute" output-channel="jmsOutChannel">
</si:service-activator>

<!-- outbound adaptor for response Queue -->
<jms:outbound-channel-adapter id="jmsout"
    channel="jmsOutChannel" destination="responseQueue" />

• 正如你所提到的，我看到交易被维持着。
• 但是我看到创建了5-10个threadPoolTaskExecutors，因为我有并发的使用者=5-10(在适配器上)
• 为什么要创建多个threadPoolTaskExecutors？它不应该只使用一个线程池任务执行器吗？
• 如果将并发使用者设置为1，则自会话处理以来，它将等待接收每条消息。

感谢你的回应。谢谢。

Update -3-当与适配器一起使用时，在executor服务中维护事务非常完美。

• 关于杀死特定线程，我向executor服务添加了一个名为interruptNamedThread(String threadName)的方法，并通过JMX公开了它。
• 触发它并设置中断标志。
• 我可以通过返回来检查中断状态和终止线程(只是因为我的测试代码运行在一个循环中，并且有机会检查中断状态)。
• 然而，这无助于我的设想。
• 因为我的线程没有在循环中运行。它与DB对话并生成报告。线的末端。返回池
• 当从DB调用或第三方库调用(报表生成器)返回的时间太长时，我想结束线程。
• 在这种情况下，它将没有机会检查中断标志，因为它被卡在一个调用(DB或报表生成器)中。

在我的用例中，重新启动组件是杀死线程的唯一方法吗？

谢谢你的建议。

Answer 1

您可以注入一个TaskExcecutor并将其waitForTasksToCompleteOnShutdown设置为false；这将导致它的shutdown()方法在调用底层ExecutorService时调用shutdownNow() (或注入ExecutorService并直接调用其shutdownNow()方法)。

这将取消(尝试中断)活动线程。

但是，如果您的服务正在做一些不可中断的事情，比如.

synchronize(foo) {}

lock.lock()

从套接字读取...or等。