java - 线程池执行器 : : TaskRejectedException from Executor

Question

我的应用程序正在通过 Jms MessageListener 类读取消息，并且在某个时间点它抛出 TaskRejectedException。我知道你们中的大多数人会说线程数超过了 maxPoolSize 并且队列也已满。

但我观察到了一些东西。发送到 MessageListener 类从中获取消息的队列的消息数是 10353，我的 threadPoolExecutor spring 属性如下:

<bean id="ticketReaderThreadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" scope="singleton" destroy-method="destroy">
    <property name="corePoolSize" value="10" />
    <property name="maxPoolSize" value="150" />
    <property name="queueCapacity" value="11000" />
</bean>

现在根据我的说法，maxPoolSize 足以处理这么多请求。因此，如果你们中的任何人都可以给出 maxPoolSize 违规之外的原因，那么请这样做。

我们现在第二次遇到这个问题，之前我们已经尝试增加 maxPoolSize，但 15 天后我们再次遇到此异常，每天大约 5000 到 8000 次。

更新:

这是异常的完整堆栈跟踪:

General Exception occurred while reading from Queue/Processing the message org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@408b9775] did not accept task: com.batman.rapid.rapidserver.sla.TicketHandler@1be5e598 at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:244) at com.batman.rapid.rapidserver.sla.JmsTicketReceiver.onMessage(JmsTicketReceiver.java:58) at org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:560) at org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:498) at org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467) at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:325) at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:263) at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1058) at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1050) at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947) at java.lang.Thread.run(Thread.java:662) Caused by: java.util.concurrent.RejectedExecutionException at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:1774) at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:768) at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:656) at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:241) ... 10 more

这是相关代码:

if (message instanceof TextMessage)
{
    textMessage = (TextMessage) message;
    ticketReaderThreadPool.execute(new TicketHandler(textMessage.getText()));
}

下面是请求的配置:

    <!-- End of JMS Queue Support -->

    <bean id="ticketReaderThreadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" scope="singleton" destroy-method="destroy">
    <property name="corePoolSize" value="10" />
    <property name="maxPoolSize" value="150" />
    <property name="queueCapacity" value="11000" />
</bean>

    <bean id="notificationThreadPool" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" scope="singleton" destroy-method="destroy">
            <property name="corePoolSize" value="10" />
            <property name="maxPoolSize" value="100" />
            <property name="queueCapacity" value="10000" />
    </bean>

    <bean id="notificationManager" class="com.batman.rapid.rapidserver.sla.scheduler.NotificationManager" scope="singleton">
            <property name="defaultPercent" value="80"></property>
    </bean>

    <bean id="dbUpdateThreads" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor" scope="singleton" destroy-method="destroy">
            <property name="corePoolSize" value="1" />
            <property name="maxPoolSize" value="100" />
            <property name="queueCapacity" value="10000" />
    </bean>

Answer 1

不确定具体原因是什么。但是我强烈认为，最大消息数可能会超出您设置的 maxqueueCapacity。但是，您使用的 vary 方法与 JMS 的一般用法不同。一般在 DefaultMessageListener 中我们配置 Max-consumer 的数量来并行处理异步处理。基于服务器资源，我们配置服务器可以处理的最大消费者

但是在您的情况下，从 DefaultMessageListener 读取消息，您生成新线程并在新线程中执行业务逻辑。由于消息监听器读取速度比业务逻辑快，因此任务会累积在线程任务队列中。

我建议重新处理您当前的实现。因为它不支持交易。我的意思是当服务器崩溃时，任务中所有待处理的消息处理都将被终止/丢失，并且由于您将设置自动确认，消息将从队列中删除。其他问题如手动设置线程队列限制等。

终于回到你的问题，你可以快速检查一下你设置的150个线程中是否有锁