Skip to content
Permalink
Browse files

TaskExecutorRegistration does not apply its default settings to a use…

…r-provided executor

Also, ChannelRegistration.setInterceptors is deprecated now: in favor of a fluently named interceptors(...) method which is documented to add the given interceptors to the channel's current list.

Issue: SPR-15962
Issue: SPR-15976

(cherry picked from commit ac9cfef)
  • Loading branch information...
jhoeller committed Sep 19, 2017
1 parent dedecb9 commit d11bd64e0d9bc4388d462dfd7cc42f537cb7caf4
@@ -120,13 +120,15 @@ public ApplicationContext getApplicationContext() {
public AbstractSubscribableChannel clientInboundChannel() {
ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(clientInboundChannelExecutor());
ChannelRegistration reg = getClientInboundChannelRegistration();
channel.setInterceptors(reg.getInterceptors());
if (reg.hasInterceptors()) {
channel.setInterceptors(reg.getInterceptors());
}
return channel;
}

@Bean
public ThreadPoolTaskExecutor clientInboundChannelExecutor() {
TaskExecutorRegistration reg = getClientInboundChannelRegistration().getOrCreateTaskExecRegistration();
TaskExecutorRegistration reg = getClientInboundChannelRegistration().taskExecutor();
ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
executor.setThreadNamePrefix("clientInboundChannel-");
return executor;
@@ -136,7 +138,7 @@ protected final ChannelRegistration getClientInboundChannelRegistration() {
if (this.clientInboundChannelRegistration == null) {
ChannelRegistration registration = new ChannelRegistration();
configureClientInboundChannel(registration);
registration.setInterceptors(new ImmutableMessageChannelInterceptor());
registration.interceptors(new ImmutableMessageChannelInterceptor());
this.clientInboundChannelRegistration = registration;
}
return this.clientInboundChannelRegistration;
@@ -153,13 +155,15 @@ protected void configureClientInboundChannel(ChannelRegistration registration) {
public AbstractSubscribableChannel clientOutboundChannel() {
ExecutorSubscribableChannel channel = new ExecutorSubscribableChannel(clientOutboundChannelExecutor());
ChannelRegistration reg = getClientOutboundChannelRegistration();
channel.setInterceptors(reg.getInterceptors());
if (reg.hasInterceptors()) {
channel.setInterceptors(reg.getInterceptors());
}
return channel;
}

@Bean
public ThreadPoolTaskExecutor clientOutboundChannelExecutor() {
TaskExecutorRegistration reg = getClientOutboundChannelRegistration().getOrCreateTaskExecRegistration();
TaskExecutorRegistration reg = getClientOutboundChannelRegistration().taskExecutor();
ThreadPoolTaskExecutor executor = reg.getTaskExecutor();
executor.setThreadNamePrefix("clientOutboundChannel-");
return executor;
@@ -169,7 +173,7 @@ protected final ChannelRegistration getClientOutboundChannelRegistration() {
if (this.clientOutboundChannelRegistration == null) {
ChannelRegistration registration = new ChannelRegistration();
configureClientOutboundChannel(registration);
registration.setInterceptors(new ImmutableMessageChannelInterceptor());
registration.interceptors(new ImmutableMessageChannelInterceptor());
this.clientOutboundChannelRegistration = registration;
}
return this.clientOutboundChannelRegistration;
@@ -185,9 +189,9 @@ protected void configureClientOutboundChannel(ChannelRegistration registration)
@Bean
public AbstractSubscribableChannel brokerChannel() {
ChannelRegistration reg = getBrokerRegistry().getBrokerChannelRegistration();
ExecutorSubscribableChannel channel = reg.hasTaskExecutor() ?
new ExecutorSubscribableChannel(brokerChannelExecutor()) : new ExecutorSubscribableChannel();
reg.setInterceptors(new ImmutableMessageChannelInterceptor());
ExecutorSubscribableChannel channel = (reg.hasTaskExecutor() ?
new ExecutorSubscribableChannel(brokerChannelExecutor()) : new ExecutorSubscribableChannel());
reg.interceptors(new ImmutableMessageChannelInterceptor());
channel.setInterceptors(reg.getInterceptors());
return channel;
}
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2017 the original author or authors.7
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -28,6 +28,7 @@
* {@link org.springframework.messaging.MessageChannel}.
*
* @author Rossen Stoyanchev
* @author Juergen Hoeller
* @since 4.0
*/
public class ChannelRegistration {
@@ -41,26 +42,37 @@
* Configure the thread pool backing this message channel.
*/
public TaskExecutorRegistration taskExecutor() {
if (this.registration == null) {
this.registration = new TaskExecutorRegistration();
}
return this.registration;
return taskExecutor(null);
}

/**
* Configure the thread pool backing this message channel using a custom
* ThreadPoolTaskExecutor.
* @param taskExecutor the executor to use (or {@code null} for a default executor)
*/
public TaskExecutorRegistration taskExecutor(ThreadPoolTaskExecutor taskExecutor) {
if (this.registration == null) {
this.registration = new TaskExecutorRegistration(taskExecutor);
this.registration = (taskExecutor != null ? new TaskExecutorRegistration(taskExecutor) :
new TaskExecutorRegistration());
}
return this.registration;
}

/**
* Configure the given interceptors for this message channel,
* adding them to the channel's current list of interceptors.
* @since 4.3.12
*/
public ChannelRegistration interceptors(ChannelInterceptor... interceptors) {
this.interceptors.addAll(Arrays.asList(interceptors));
return this;
}

/**
* Configure interceptors for the message channel.
* @deprecated as of 4.3.12, in favor of {@link #interceptors(ChannelInterceptor...)}
*/
@Deprecated
public ChannelRegistration setInterceptors(ChannelInterceptor... interceptors) {
if (interceptors != null) {
this.interceptors.addAll(Arrays.asList(interceptors));
@@ -73,10 +85,18 @@ protected boolean hasTaskExecutor() {
return (this.registration != null);
}

/**
* @deprecated as of 4.3.12 since it's not used anymore
*/
@Deprecated
protected TaskExecutorRegistration getTaskExecRegistration() {
return this.registration;
}

/**
* @deprecated as of 4.3.12 since it's not used anymore
*/
@Deprecated
protected TaskExecutorRegistration getOrCreateTaskExecRegistration() {
return taskExecutor();
}
@@ -88,4 +108,5 @@ protected boolean hasInterceptors() {
protected List<ChannelInterceptor> getInterceptors() {
return this.interceptors;
}

}
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2017 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.
@@ -17,33 +17,49 @@
package org.springframework.messaging.simp.config;

import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.Assert;

/**
* A registration class for customizing the properties of {@link ThreadPoolTaskExecutor}.
*
* @author Rossen Stoyanchev
* @author Juergen Hoeller
* @since 4.0
*/
public class TaskExecutorRegistration {

private ThreadPoolTaskExecutor taskExecutor;
private final ThreadPoolTaskExecutor taskExecutor;

private int corePoolSize = Runtime.getRuntime().availableProcessors() * 2;
private Integer corePoolSize;

private int maxPoolSize = Integer.MAX_VALUE;
private Integer maxPoolSize;

private int queueCapacity = Integer.MAX_VALUE;
private Integer keepAliveSeconds;

private int keepAliveSeconds = 60;
private Integer queueCapacity;


/**
* Create a new {@code TaskExecutorRegistration} for a default
* {@link ThreadPoolTaskExecutor}.
*/
public TaskExecutorRegistration() {
this.taskExecutor = new ThreadPoolTaskExecutor();
this.taskExecutor.setCorePoolSize(Runtime.getRuntime().availableProcessors() * 2);
this.taskExecutor.setAllowCoreThreadTimeOut(true);
}

/**
* Create a new {@code TaskExecutorRegistration} for a given
* {@link ThreadPoolTaskExecutor}.
* @param taskExecutor the executor to use
*/
public TaskExecutorRegistration(ThreadPoolTaskExecutor taskExecutor) {
Assert.notNull(taskExecutor, "ThreadPoolTaskExecutor must not be null");
this.taskExecutor = taskExecutor;
}


/**
* Set the core pool size of the ThreadPoolExecutor.
* <p><strong>NOTE:</strong> The core pool size is effectively the max pool size
@@ -75,6 +91,18 @@ public TaskExecutorRegistration maxPoolSize(int maxPoolSize) {
return this;
}

/**
* 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 keepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
return this;
}

/**
* Set the queue capacity for the ThreadPoolExecutor.
* <p><strong>NOTE:</strong> when an unbounded {@code queueCapacity} is configured
@@ -89,26 +117,21 @@ public TaskExecutorRegistration queueCapacity(int queueCapacity) {
return this;
}

/**
* 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 keepAliveSeconds(int keepAliveSeconds) {
this.keepAliveSeconds = keepAliveSeconds;
return this;
}

protected ThreadPoolTaskExecutor getTaskExecutor() {
ThreadPoolTaskExecutor executor = (this.taskExecutor != null ? this.taskExecutor : new ThreadPoolTaskExecutor());
executor.setCorePoolSize(this.corePoolSize);
executor.setMaxPoolSize(this.maxPoolSize);
executor.setKeepAliveSeconds(this.keepAliveSeconds);
executor.setQueueCapacity(this.queueCapacity);
executor.setAllowCoreThreadTimeOut(true);
return executor;
if (this.corePoolSize != null) {
this.taskExecutor.setCorePoolSize(this.corePoolSize);
}
if (this.maxPoolSize != null) {
this.taskExecutor.setMaxPoolSize(this.maxPoolSize);
}
if (this.keepAliveSeconds != null) {
this.taskExecutor.setKeepAliveSeconds(this.keepAliveSeconds);
}
if (this.queueCapacity != null) {
this.taskExecutor.setQueueCapacity(this.queueCapacity);
}
return this.taskExecutor;
}

}
@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2017 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.
@@ -511,14 +511,14 @@ public void configureMessageBroker(MessageBrokerRegistry registry) {

@Override
protected void configureClientInboundChannel(ChannelRegistration registration) {
registration.setInterceptors(this.interceptor);
registration.interceptors(this.interceptor);
registration.taskExecutor(new CustomThreadPoolTaskExecutor())
.corePoolSize(11).maxPoolSize(12).keepAliveSeconds(13).queueCapacity(14);
}

@Override
protected void configureClientOutboundChannel(ChannelRegistration registration) {
registration.setInterceptors(this.interceptor, this.interceptor);
registration.interceptors(this.interceptor, this.interceptor);
registration.taskExecutor().corePoolSize(21).maxPoolSize(22).keepAliveSeconds(23).queueCapacity(24);
}

@@ -534,7 +534,7 @@ protected void addReturnValueHandlers(List<HandlerMethodReturnValueHandler> retu

@Override
protected void configureMessageBroker(MessageBrokerRegistry registry) {
registry.configureBrokerChannel().setInterceptors(this.interceptor, this.interceptor, this.interceptor);
registry.configureBrokerChannel().interceptors(this.interceptor, this.interceptor, this.interceptor);
registry.configureBrokerChannel().taskExecutor().corePoolSize(31).maxPoolSize(32).keepAliveSeconds(33).queueCapacity(34);
registry.setPathMatcher(new AntPathMatcher(".")).enableSimpleBroker("/topic", "/queue");
registry.setCacheLimit(8192);

0 comments on commit d11bd64

Please sign in to comment.
You can’t perform that action at this time.