Spring-Integration - Queue Kanal + Service activator-Poller-threadpool erschöpft
Habe ich eine einfache spring-integration-app, wo ich bin versucht zu veröffentlichen, eine Aufgabe, die zu einem queue-Kanal, und dann einen Arbeiter Holen sich die Aufgabe und führen Sie es aus. (aus einem pool mit mehreren gleichzeitigen Arbeitskräfte zur Verfügung).
Ich finde, dass der thread-pool ist schnell erschöpft, und die Aufgaben werden abgelehnt.
Hier ist meine config:
<int:annotation-config />
<task:annotation-driven executor="executor" scheduler="scheduler"/>
<task:executor id="executor" pool-size="5-20" rejection-policy="CALLER_RUNS" />
<task:scheduler id="scheduler" pool-size="5"/>
<int:gateway service-interface="com.example.MyGateway">
<int:method name="queueForSync" request-channel="worker.channel" />
</int:gateway>
<int:channel id="worker.channel">
<int:queue />
</int:channel>
<bean class="com.example.WorkerBean" id="workerBean" />
<int:service-activator ref="workerBean" method="doWork" input-channel="worker.channel">
<int:poller fixed-delay="50" task-executor="executor" receive-timeout="0" />
</int:service-activator>
Diese Frage ist sehr ähnlich zu einem anderen, fragte ich eine Weile zurück, hier. Der wesentliche Unterschied ist, dass ich nicht mit einem AMQP message broker hier, nur interne Feder-Nachricht-Kanäle.
Ich habe nicht in der Lage zu finden, die eine Analogie für die concurrent-consumer
Konzept in Vanille-Feder-Kanäle.
Darüber hinaus habe ich angenommen, Gary Russell empfohlene config:
Um dies zu vermeiden, stellen Sie einfach den receive-timeout auf 0 auf
<poller/>
Trotz, ich bin noch immer der pool erschöpft.
Was ist die richtige Konfiguration für dieses Ziel?
Nebenbei - zwei andere riecht hier vorschlagen, dass meine config ist falsch:
- Warum werde ich immer abgelehnt Ausnahmen, wenn die
rejection-policy
istCALLER_RUNS
? - Den Ausnahmen beginnen, wenn das
queued tasks = 1000
. Gegeben, es gibt keine Warteschlange-Kapazität auf den Testamentsvollstrecker sollte nicht die Warteschlange Grenzenlos?
Ausnahme-stack-trace angezeigt:
[Mon Dec 2013 17:44:57.172] ERROR [task-scheduler-6] (org.springframework.integration.handler.LoggingHandler:126) - org.springframework.core.task.TaskRejectedException: Executor [java.util.concurrent.ThreadPoolExecutor@48e83911[Running, pool size = 20, active threads = 20, queued tasks = 1000, completed tasks = 48]] did not accept task: org.springframework.integration.util.ErrorHandlingTaskExecutor$1@a5798e
at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:244)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:49)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:231)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:53)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
at java.util.concurrent.FutureTask.run(FutureTask.java:262)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)
Caused by: java.util.concurrent.RejectedExecutionException: Task org.springframework.integration.util.ErrorHandlingTaskExecutor$1@a5798e rejected from java.util.concurrent.ThreadPoolExecutor@48e83911[Running, pool size = 20, active threads = 20, queued tasks = 1000, completed tasks = 48]
at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2048)
at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:821)
at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1372)
at org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.execute(ThreadPoolTaskExecutor.java:241)
... 11 more
- Downvoter - Pflege, Stellung zu nehmen, warum?
Du musst angemeldet sein, um einen Kommentar abzugeben.
Beste Vermutung ist, Sie haben eine andere
executor
bean irgendwo anders im Kontext.Schalten Sie die debug-Protokollierung und suchen
...DefaultListableBeanFactory] Overriding bean definition for bean 'executor'
.Den default-queue-Kapazität ist
Integer.MAX_VALUE
.