From eb9c25615b852f9264918ae39ecae9ee9bfca95a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 3 Jul 2012 12:44:26 -0400 Subject: [PATCH] INT-2648 UDP - Don't Hold a Scheduler Thread Previously, the UDP inbound adapter was incorrectly using the default task scheduler (taskScheduler) to run its main receive activity. Once a packet is read, it is handed off to another thread for message processing, using a configured (or default) task executor. We should not hold on to a scheduler thread permanently; the default scheduler has only 10 threads and one user's app stopped working when using 10 inbound adapters. The work around was to define an explicit 'taskScheduler' bean with more threads. --- ...ternetProtocolReceivingChannelAdapter.java | 22 ++++++++++--------- .../udp/UnicastReceivingChannelAdapter.java | 2 +- .../integration/ip/udp/MultiClientTests.java | 10 --------- .../ip/udp/UdpChannelAdapterTests.java | 17 +------------- 4 files changed, 14 insertions(+), 37 deletions(-) diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java index 23391ef289..8847124d53 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java @@ -16,18 +16,15 @@ package org.springframework.integration.ip; -import java.util.Date; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.ThreadFactory; import org.springframework.integration.endpoint.MessageProducerSupport; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.util.Assert; /** * Base class for inbound TCP/UDP Channel Adapters. - * + * * @author Mark Fisher * @author Gary Russell * @since 2.0 @@ -59,7 +56,7 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter } /** - * + * * @return The port on which this receiver is listening. */ public int getPort() { @@ -99,17 +96,22 @@ public abstract class AbstractInternetProtocolReceivingChannelAdapter return receiveBufferSize; } + /** + * Protected by lifecycleLock + */ @Override protected void doStart() { - TaskScheduler taskScheduler = this.getTaskScheduler(); - Assert.state(taskScheduler != null, "taskScheduler is required"); - this.active = true; - taskScheduler.schedule(this, new Date()); + if (!this.active) { + this.active = true; + String beanName = this.getComponentName(); + checkTaskExecutor((beanName == null ? "" : beanName + "-") + this.getComponentType()); + this.taskExecutor.execute(this); + } } /** * Creates a default task executor if none was supplied. - * + * * @param threadName */ protected void checkTaskExecutor(final String threadName) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java index 0b25640db4..dab9fa892e 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java @@ -77,7 +77,6 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece if (logger.isDebugEnabled()) { logger.debug("UDP Receiver running on port:" + this.getPort()); } - checkTaskExecutor("UDP-Incoming-Msg-Handler"); this.setListening(true); @@ -231,6 +230,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece this.mapper.setLookupHost(lookupHost); } + @Override public String getComponentType(){ return "ip:udp-inbound-channel-adapter"; } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java index 3adac3dd10..e14eda265a 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java @@ -25,7 +25,6 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; /** @@ -54,9 +53,6 @@ public class MultiClientTests { adapter.setPoolSize(drivers); QueueChannel queue = new QueueChannel(drivers * 3); adapter.setOutputChannel(queue); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); @@ -95,9 +91,6 @@ public class MultiClientTests { adapter.setPoolSize(drivers); QueueChannel queue = new QueueChannel(drivers * 3); adapter.setOutputChannel(queue); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); @@ -140,9 +133,6 @@ public class MultiClientTests { adapter.setPoolSize(drivers); QueueChannel queue = new QueueChannel(drivers * 3); adapter.setOutputChannel(queue); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); adapter.start(); final QueueChannel queueIn = new QueueChannel(1000); SocketTestUtils.waitListening(adapter); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java index 6954fe15ef..396ba73b51 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java @@ -19,7 +19,6 @@ import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.SocketUtils; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; public class UdpChannelAdapterTests { @@ -31,9 +30,6 @@ public class UdpChannelAdapterTests { int port = SocketUtils.findAvailableUdpSocket(); UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); adapter.setOutputChannel(channel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); // SocketUtils.setLocalNicIfPossible(adapter); adapter.start(); SocketTestUtils.waitListening(adapter); @@ -53,10 +49,8 @@ public class UdpChannelAdapterTests { QueueChannel channel = new QueueChannel(2); int port = SocketUtils.findAvailableUdpSocket(); UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); + adapter.setBeanName("test"); adapter.setOutputChannel(channel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); // SocketUtils.setLocalNicIfPossible(adapter); adapter.start(); SocketTestUtils.waitListening(adapter); @@ -82,9 +76,6 @@ public class UdpChannelAdapterTests { int port = SocketUtils.findAvailableUdpSocket(); MulticastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.8", port); adapter.setOutputChannel(channel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); String nic = SocketTestUtils.chooseANic(true); if (nic == null) { // no multicast support LogFactory.getLog(this.getClass()).error("No Multicast support"); @@ -112,9 +103,6 @@ public class UdpChannelAdapterTests { int port = SocketUtils.findAvailableUdpSocket(); UnicastReceivingChannelAdapter adapter = new MulticastReceivingChannelAdapter("225.6.7.9", port); adapter.setOutputChannel(channel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); String nic = SocketTestUtils.chooseANic(true); if (nic == null) { // no multicast support LogFactory.getLog(this.getClass()).error("No Multicast support"); @@ -140,9 +128,6 @@ public class UdpChannelAdapterTests { int port = SocketUtils.findAvailableUdpSocket(); UnicastReceivingChannelAdapter adapter = new UnicastReceivingChannelAdapter(port); adapter.setOutputChannel(channel); - ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler(); - taskScheduler.initialize(); - adapter.setTaskScheduler(taskScheduler); // SocketUtils.setLocalNicIfPossible(adapter); adapter.setOutputChannel(channel); ServiceActivatingHandler handler = new ServiceActivatingHandler(new FailingService());