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

Update thread pool settings in STOMP/WebSocket config

Browse Source 
The clientInboundChannel and clientOutboundChannel now use twice
the number of available processors by default to accomodate for some
degree of blocking in task execution on average.

In practice these settings still need to be configured explicitly in
applications but these should serve as better default values than
the default values in ThreadPoolTaskExecutor.

Issue: SPR-11450
This commit is contained in:
Rossen Stoyanchev
2014-03-20 00:41:31 -04:00
parent 9cee70ff66
commit 10af128e96
7 changed files with 175 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
spring-messaging/src/main/java/org/springframework/messaging/simp/config/AbstractMessageBrokerConfiguration.java
View File

@@ -17,7 +17,13 @@
package org.springframework.messaging.simp.config;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.AbstractExecutorService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.BeansException;
@@ -93,17 +99,15 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
@Bean
public AbstractSubscribableChannel clientInboundChannel() {
ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(clientInboundChannelExecutor());
ChannelRegistration r = getClientInboundChannelRegistration();
if (r.hasInterceptors()) {
channel.setInterceptors(r.getInterceptors());
}
ChannelRegistration reg = getClientInboundChannelRegistration();
channel.setInterceptors(reg.getInterceptors());
return channel;
}
@Bean
public ThreadPoolTaskExecutor clientInboundChannelExecutor() {
TaskExecutorRegistration r = getClientInboundChannelRegistration().getTaskExecutorRegistration();
ThreadPoolTaskExecutor executor = (r != null) ? r.getTaskExecutor() : new ThreadPoolTaskExecutor();
TaskExecutorRegistration reg = getClientInboundChannelRegistration().getOrCreateTaskExecRegistration();
ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
executor.setThreadNamePrefix("clientInboundChannel-");
return executor;
}
@@ -129,17 +133,15 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
@Bean
public AbstractSubscribableChannel clientOutboundChannel() {
ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(clientOutboundChannelExecutor());
ChannelRegistration r = getClientOutboundChannelRegistration();
if (r.hasInterceptors()) {
channel.setInterceptors(r.getInterceptors());
}
ChannelRegistration reg = getClientOutboundChannelRegistration();
channel.setInterceptors(reg.getInterceptors());
return channel;
}
@Bean
public ThreadPoolTaskExecutor clientOutboundChannelExecutor() {
TaskExecutorRegistration r = getClientOutboundChannelRegistration().getTaskExecutorRegistration();
ThreadPoolTaskExecutor executor = (r != null) ? r.getTaskExecutor() : new ThreadPoolTaskExecutor();
TaskExecutorRegistration reg = getClientOutboundChannelRegistration().getOrCreateTaskExecRegistration();
ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
executor.setThreadNamePrefix("clientOutboundChannel-");
return executor;
}
@@ -162,24 +164,27 @@ public abstract class AbstractMessageBrokerConfiguration implements ApplicationC
@Bean
public AbstractSubscribableChannel brokerChannel() {
ChannelRegistration r = getBrokerRegistry().getBrokerChannelRegistration();
ExecutorSubscribableChannel channel;
if (r.hasTaskExecutor()) {
channel = new ExecutorSubscribableChannel(); // synchronous by default
}
else {
channel = new ExecutorSubscribableChannel(brokerChannelExecutor());
}
if (r.hasInterceptors()) {
channel.setInterceptors(r.getInterceptors());
}
ChannelRegistration reg = getBrokerRegistry().getBrokerChannelRegistration();
ExecutorSubscribableChannel channel = reg.hasTaskExecutor() ?
new ExecutorSubscribableChannel(brokerChannelExecutor()) : new ExecutorSubscribableChannel();
channel.setInterceptors(reg.getInterceptors());
return channel;
}
@Bean
public ThreadPoolTaskExecutor brokerChannelExecutor() {
TaskExecutorRegistration r = getBrokerRegistry().getBrokerChannelRegistration().getTaskExecutorRegistration();
ThreadPoolTaskExecutor executor = (r != null) ? r.getTaskExecutor() : new ThreadPoolTaskExecutor();
ChannelRegistration reg = getBrokerRegistry().getBrokerChannelRegistration();
ThreadPoolTaskExecutor executor;
if (reg.hasTaskExecutor()) {
executor = reg.taskExecutor().getTaskExecutor();
}
else {
// Should never be used
executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(0);
executor.setMaxPoolSize(1);
executor.setQueueCapacity(0);
}
executor.setThreadNamePrefix("brokerChannel-");
return executor;
}

24
spring-messaging/src/main/java/org/springframework/messaging/simp/config/ChannelRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,17 +31,19 @@ import org.springframework.messaging.support.ChannelInterceptor;
*/
public class ChannelRegistration {
private TaskExecutorRegistration taskExecutorRegistration;
private TaskExecutorRegistration registration;
private List<ChannelInterceptor> interceptors = new ArrayList<ChannelInterceptor>();
private final List<ChannelInterceptor> interceptors = new ArrayList<ChannelInterceptor>();
/**
* Configure properties of the ThreadPoolTaskExecutor backing the message channel.
* Configure the thread pool backing this message channel.
*/
public TaskExecutorRegistration taskExecutor() {
this.taskExecutorRegistration = new TaskExecutorRegistration();
return this.taskExecutorRegistration;
if (this.registration == null) {
this.registration = new TaskExecutorRegistration();
}
return this.registration;
}
/**
@@ -56,11 +58,15 @@ public class ChannelRegistration {
protected boolean hasTaskExecutor() {
return (this.taskExecutorRegistration != null);
return (this.registration != null);
}
protected TaskExecutorRegistration getTaskExecutorRegistration() {
return this.taskExecutorRegistration;
protected TaskExecutorRegistration getTaskExecRegistration() {
return this.registration;
}
protected TaskExecutorRegistration getOrCreateTaskExecRegistration() {
return taskExecutor();
}
protected boolean hasInterceptors() {

66
spring-messaging/src/main/java/org/springframework/messaging/simp/config/TaskExecutorRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.messaging.simp.config;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
/**
* A registration class for customizing the properties of {@link ThreadPoolTaskExecutor}.
*
@@ -26,18 +27,28 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
*/
public class TaskExecutorRegistration {
private int corePoolSize = 1;
private int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
private int maxPoolSize = Integer.MAX_VALUE;
private int keepAliveSeconds = 60;
private int queueCapacity = Integer.MAX_VALUE;
private int keepAliveSeconds = 60;
/**
* Set the ThreadPoolExecutor's core pool size.
* Default is 1.
* Set the core pool size of the ThreadPoolExecutor.
*
* <p><strong>NOTE:</strong> the core pool size is effectively the max pool size
* when an unbounded {@link #queueCapacity(int) queueCapacity} is configured
* (the default). This is essentially the "Unbounded queues" strategy as explained
* in {@link java.util.concurrent.ThreadPoolExecutor ThreadPoolExecutor}. When
* this strategy is used, the {@link #maxPoolSize(int) maxPoolSize} is ignored.
*
* <p>By default this is set to twice the value of
* {@link Runtime#availableProcessors()}. In an an application where tasks do not
* block frequently, the number should be closer to or equal to the number of
* available CPUs/cores.
*/
public TaskExecutorRegistration corePoolSize(int corePoolSize) {
this.corePoolSize = corePoolSize;
@@ -45,8 +56,15 @@ public class TaskExecutorRegistration {
}
/**
* Set the ThreadPoolExecutor's maximum pool size.
* Default is {@code Integer.MAX_VALUE}.
* Set the max pool size of the ThreadPoolExecutor.
*
* <p><strong>NOTE:</strong> when an unbounded
* {@link #queueCapacity(int) queueCapacity} is configured (the default), the
* max pool size is effectively ignored. See the "Unbounded queues" strategy
* in {@link java.util.concurrent.ThreadPoolExecutor ThreadPoolExecutor} for
* more details.
*
* <p>By default this is set to {@code Integer.MAX_VALUE}.
*/
public TaskExecutorRegistration maxPoolSize(int maxPoolSize) {
this.maxPoolSize = maxPoolSize;
@@ -54,24 +72,32 @@ public class TaskExecutorRegistration {
}
/**
* Set the ThreadPoolExecutor's keep-alive seconds.
* Default is 60.
* Set the queue capacity for the ThreadPoolExecutor.
*
* <p><strong>NOTE:</strong> when an unbounded
* {@link #queueCapacity(int) queueCapacity} is configured (the default) the
* core pool size is effectively the max pool size. This is essentially the
* "Unbounded queues" strategy as explained in
* {@link java.util.concurrent.ThreadPoolExecutor ThreadPoolExecutor}. When
* this strategy is used, the {@link #maxPoolSize(int) maxPoolSize} is ignored.
*
* <p>By default this is set to {@code Integer.MAX_VALUE}.
*/
public TaskExecutorRegistration keepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
public TaskExecutorRegistration queueCapacity(int queueCapacity) {
this.queueCapacity = queueCapacity;
return this;
}
/**
* Set the capacity for the ThreadPoolExecutor's BlockingQueue.
* Default is {@code Integer.MAX_VALUE}.
* <p>Any positive value will lead to a LinkedBlockingQueue instance;
* any other value will lead to a SynchronousQueue instance.
* @see java.util.concurrent.LinkedBlockingQueue
* @see java.util.concurrent.SynchronousQueue
* Set the time limit for which threads may remain idle before being terminated.
* If there are more than the core number of threads currently in the pool,
* after waiting this amount of time without processing a task, excess threads
* will be terminated. This overrides any value set in the constructor.
*
* <p>By default this is set to 60.
*/
public TaskExecutorRegistration queueCapacity(int queueCapacity) {
this.queueCapacity = queueCapacity;
public TaskExecutorRegistration keepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
return this;
}