java - Spring 集成: Application leaking SimpleAsyncTaskExecutor threads?

Question

我们在生产中有一个 Spring 应用程序，它似乎正在泄漏线程。

当我进行线程转储时，我看到以下线程似乎正在等待。下面是一个示例，但数量有数千个

    "SimpleAsyncTaskExecutor-1801" prio=10 tid=0x00007fa668413800 nid=0x376c waiting on condition [0x00007fa4fbaff000]
   java.lang.Thread.State: WAITING (parking)
        at sun.misc.Unsafe.park(Native Method)
        - parking to wait for  <0x00000007777e2ba8> (a java.util.concurrent.CountDownLatch$Sync)
        at java.util.concurrent.locks.LockSupport.park(LockSupport.java:186)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:834)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:994)
        at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1303)
        at java.util.concurrent.CountDownLatch.await(CountDownLatch.java:236)
        at org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel.receive(MessagingTemplate.java:415)
        at org.springframework.integration.core.MessagingTemplate$TemporaryReplyChannel.receive(MessagingTemplate.java:409)
        at org.springframework.integration.core.MessagingTemplate.doReceive(MessagingTemplate.java:317)
        at org.springframework.integration.core.MessagingTemplate.doSendAndReceive(MessagingTemplate.java:341)
        at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:255)
        at org.springframework.integration.core.MessagingTemplate.convertSendAndReceive(MessagingTemplate.java:290)
        at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:224)
        at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceive(MessagingGatewaySupport.java:203)
        at org.springframework.integration.gateway.GatewayProxyFactoryBean.invokeGatewayMethod(GatewayProxyFactoryBean.java:306)
        at org.springframework.integration.gateway.GatewayProxyFactoryBean.doInvoke(GatewayProxyFactoryBean.java:269)
        at org.springframework.integration.gateway.GatewayProxyFactoryBean.access$200(GatewayProxyFactoryBean.java:71)
        at org.springframework.integration.gateway.GatewayProxyFactoryBean$AsyncInvocationTask.call(GatewayProxyFactoryBean.java:499)
        at java.util.concurrent.FutureTask.run(FutureTask.java:262)
        at java.lang.Thread.run(Thread.java:745)

最终，它会导致服务器变得无响应，只有重新启动才能恢复。

这似乎涉及一个 Spring Integration Gateway，该网关发送了一条消息并正在等待一些它永远不会得到的答案。我们的应用程序中唯一使用此功能的地方是通知 channel ，该 channel 使用服务接口(interface)将这些消息发送(发布)到 RabbitMQ 交换

这是服务接口(interface)代码:

@Component
public interface INotificationSender {

    Future<Void> sendNotification(@Payload Object notification,
            @Header("routingKey") String routingKey,
            @Header("notificationType") String type);
}

这是相关的 spring 配置:

<!-- Spring Integration RabbitMQ adapter -->        
<rabbit:template
    id="amqpTemplate"
    connection-factory="notificationConnectionFactory" />

<rabbit:connection-factory
    id="notificationConnectionFactory"
    addresses="${notificationChannel.rabbitHost1}:${notificationChannel.rabbitPort}, ${notificationChannel.rabbitHost2}:${notificationChannel.rabbitPort}"
    username="${notificationChannel.rabbitUsername}"
    password="${notificationChannel.rabbitPassword}"
    virtual-host="${notificationChannel.rabbitVirtualHost}"/>

<rabbit:topic-exchange
    name="notificationExchange"/>


<!-- Spring Integration AMQP -->        
<int-amqp:outbound-channel-adapter 
    id="notificationChannelAdapter"
    channel="notificationChannelEnc"
    exchange-name="notificationExchange"
    routing-key-expression="headers['routingKey']"
    mapped-request-headers="STANDARD_REQUEST_HEADERS, notificationType"/>


<!-- Spring Integration Core -->
<int:channel
    id="notificationChannelEnc">
    <int:interceptors>
        <int:wire-tap channel="loggingChannel" />
    </int:interceptors>
</int:channel>

<int:channel id="notificationChannel"/>

<int:object-to-json-transformer
    id="NotificationEncoder"
    content-type="text/x-json"
    input-channel="notificationChannel"
    output-channel="notificationChannelEnc"/>       

<int:gateway
    id="notificationGateway"
    default-request-channel="notificationChannel"
    service-interface="com.ericsson.ericloud.commander.notification.sender.INotificationSender"/>

我们正在使用 Spring 版本 3.2.3.RELEASE 和 Spring Integration 版本 3.0.0.M2

有人知道如何处理这个问题并让这些线程正确终止吗？

谢谢，/塞巴斯蒂安

Answer 1

您的Future<Void>是一个瓶颈。

我猜你想异步发送通知并且不等待任何回复。但无论如何Gateway尝试等待回复，因为它看到 Future作为启动异步进程的提示。

要克服它，您必须将返回更改为简单的 void并使用ExecutorChannel作为网关的default-request-channel .

查看更多信息Reference Manual .