Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-2648 UDP - Don't Hold a Scheduler Thread

Browse Source 
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.
This commit is contained in:
Gary Russell
2012-07-03 12:44:26 -04:00
committed by Oleg Zhurakousky
parent 62f96fa194
commit eb9c25615b
4 changed files with 14 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-integration-ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolReceivingChannelAdapter.java
View File

@@ -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) {

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -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";
}

10
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/MultiClientTests.java
View File

@@ -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);

17
spring-integration-ip/src/test/java/org/springframework/integration/ip/udp/UdpChannelAdapterTests.java
View File

@@ -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());