java - 使用 Spring 集成执行器 channel 时超出 GC 限制

Question

   I have got the below exception , I suspect heap memory is full so GC exception was thrown . Kindly explain if any other perspective for the below application solution


  2017:06:07 21:18:36.275 [loginputtaskexecutor-7] ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessageHandlingException: nested exception is java.lang.IllegalStateException: Cannot process message
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:96)
    at org.springframework.integration.handler.ServiceActivatingHandler.handleRequestMessage(ServiceActivatingHandler.java:89)
    at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:109)
    at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
    at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:148)
    at org.springframework.integration.dispatcher.UnicastingDispatcher.access$000(UnicastingDispatcher.java:53)
    at org.springframework.integration.dispatcher.UnicastingDispatcher$3.run(UnicastingDispatcher.java:129)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.IllegalStateException: Cannot process message
    at org.springframework.integration.util.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:333)
    at org.springframework.integration.util.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:155)
    at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:93)
    ... 11 more
Caused by: java.lang.OutOfMemoryError: GC overhead limit exceeded


  **Application flow in detail :**

Spring集成应用程序构建监听来自ActiveMQ的消息，在使用来自ActiveMQ的消息后，它将被移交给以订阅者作为服务激活器的输入 channel (执行器 channel )。在 Service Activator 中，消息被转换为 json 然后存储到 Cassandra 中。在Service activator方法中提到了@transaction。

通过上述解决方案，我想到通过实现执行器 channel 来打破消息事务流程，消息消费后将移交给执行器 channel ，事务结束。之后，执行程序 channel 中的线程将负责对 cassandra 执行并行写入。

有没有更好的方法使用java spring集成尽可能快地将大量数据写入casandra

Answer 1

如果数据接收器无法跟上，请在 TaskExecutor 中添加队列大小限制，并在队列已满时使用 CallerRunsPolicy 或 CallerBlocksPolicy。

这自然会以接收器可以处理的速度限制工作负载。