Implement SimpleAsyncTaskExecutorBuilder

The SimpleAsyncTaskExecutorBuilder can be used to create
SimpleAsyncTaskExecutor. It will be auto-configured into the context.
SimpleAsyncTaskExecutorCustomizer can be used to customize the built
SimpleAsyncTaskExecutor.

If virtual threads are enabled:
- SimpleAsyncTaskExecutor will use virtual threads
- SimpleAsyncTaskExecutorBuilder will be used as the application task
  executor

A new property 'spring.task.execution.simple.concurrency-limit' has been
added to control the concurrency limit of the SimpleAsyncTaskExecutor

Closes gh-35711

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/task/TaskExecutionAutoConfiguration.java

@@ -37,6 +37,7 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 @EnableConfigurationProperties(TaskExecutionProperties.class)
 @Import({ TaskExecutorConfigurations.ThreadPoolTaskExecutorBuilderConfiguration.class,
 		TaskExecutorConfigurations.TaskExecutorBuilderConfiguration.class,
+		TaskExecutorConfigurations.SimpleAsyncTaskExecutorBuilderConfiguration.class,
 		TaskExecutorConfigurations.VirtualThreadTaskExecutorConfiguration.class,
 		TaskExecutorConfigurations.ThreadPoolTaskExecutorConfiguration.class })
 public class TaskExecutionAutoConfiguration {

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/task/TaskExecutionProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2012-2019 the original author or authors.
+ * Copyright 2012-2023 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.
@@ -32,6 +32,8 @@ public class TaskExecutionProperties {
 
 	private final Pool pool = new Pool();
 
+	private final Simple simple = new Simple();
+
 	private final Shutdown shutdown = new Shutdown();
 
 	/**
@@ -39,6 +41,10 @@ public class TaskExecutionProperties {
 	 */
 	private String threadNamePrefix = "task-";
 
+	public Simple getSimple() {
+		return this.simple;
+	}
+
 	public Pool getPool() {
 		return this.pool;
 	}
@@ -55,6 +61,24 @@ public class TaskExecutionProperties {
 		this.threadNamePrefix = threadNamePrefix;
 	}
 
+	public static class Simple {
+
+		/**
+		 * Set the maximum number of parallel accesses allowed. -1 indicates no
+		 * concurrency limit at all.
+		 */
+		private Integer concurrencyLimit;
+
+		public Integer getConcurrencyLimit() {
+			return this.concurrencyLimit;
+		}
+
+		public void setConcurrencyLimit(Integer concurrencyLimit) {
+			this.concurrencyLimit = concurrencyLimit;
+		}
+
+	}
+
 	public static class Pool {
 
 		/**

spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/task/TaskExecutorConfigurations.java

@@ -23,6 +23,8 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnThreading;
 import org.springframework.boot.autoconfigure.task.TaskExecutionProperties.Shutdown;
 import org.springframework.boot.autoconfigure.thread.Threading;
+import org.springframework.boot.task.SimpleAsyncTaskExecutorBuilder;
+import org.springframework.boot.task.SimpleAsyncTaskExecutorCustomizer;
 import org.springframework.boot.task.TaskExecutorBuilder;
 import org.springframework.boot.task.TaskExecutorCustomizer;
 import org.springframework.boot.task.ThreadPoolTaskExecutorBuilder;
@@ -52,12 +54,8 @@ class TaskExecutorConfigurations {
 
 		@Bean(name = { TaskExecutionAutoConfiguration.APPLICATION_TASK_EXECUTOR_BEAN_NAME,
 				AsyncAnnotationBeanPostProcessor.DEFAULT_TASK_EXECUTOR_BEAN_NAME })
-		SimpleAsyncTaskExecutor applicationTaskExecutor(TaskExecutionProperties properties,
-				ObjectProvider<TaskDecorator> taskDecorator) {
-			SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(properties.getThreadNamePrefix());
-			executor.setVirtualThreads(true);
-			taskDecorator.ifUnique(executor::setTaskDecorator);
-			return executor;
+		SimpleAsyncTaskExecutor applicationTaskExecutor(SimpleAsyncTaskExecutorBuilder builder) {
+			return builder.build();
 		}
 
 	}
@@ -144,4 +142,49 @@ class TaskExecutorConfigurations {
 
 	}
 
+	@Configuration(proxyBeanMethods = false)
+	static class SimpleAsyncTaskExecutorBuilderConfiguration {
+
+		private final TaskExecutionProperties properties;
+
+		private final ObjectProvider<SimpleAsyncTaskExecutorCustomizer> taskExecutorCustomizers;
+
+		private final ObjectProvider<TaskDecorator> taskDecorator;
+
+		SimpleAsyncTaskExecutorBuilderConfiguration(TaskExecutionProperties properties,
+				ObjectProvider<SimpleAsyncTaskExecutorCustomizer> taskExecutorCustomizers,
+				ObjectProvider<TaskDecorator> taskDecorator) {
+			this.properties = properties;
+			this.taskExecutorCustomizers = taskExecutorCustomizers;
+			this.taskDecorator = taskDecorator;
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		@ConditionalOnThreading(Threading.PLATFORM)
+		SimpleAsyncTaskExecutorBuilder simpleAsyncTaskExecutorBuilder() {
+			return builder();
+		}
+
+		@Bean(name = "simpleAsyncTaskExecutorBuilder")
+		@ConditionalOnMissingBean
+		@ConditionalOnThreading(Threading.VIRTUAL)
+		SimpleAsyncTaskExecutorBuilder simpleAsyncTaskExecutorBuilderVirtualThreads() {
+			SimpleAsyncTaskExecutorBuilder builder = builder();
+			builder = builder.virtualThreads(true);
+			return builder;
+		}
+
+		private SimpleAsyncTaskExecutorBuilder builder() {
+			SimpleAsyncTaskExecutorBuilder builder = new SimpleAsyncTaskExecutorBuilder();
+			builder = builder.threadNamePrefix(this.properties.getThreadNamePrefix());
+			builder = builder.customizers(this.taskExecutorCustomizers.orderedStream()::iterator);
+			builder = builder.taskDecorator(this.taskDecorator.getIfUnique());
+			TaskExecutionProperties.Simple simple = this.properties.getSimple();
+			builder = builder.concurrencyLimit(simple.getConcurrencyLimit());
+			return builder;
+		}
+
+	}
+
 }

spring-boot-project/spring-boot-autoconfigure/src/test/java/org/springframework/boot/autoconfigure/task/TaskExecutionAutoConfigurationTests.java

@@ -31,6 +31,7 @@ import org.junit.jupiter.api.extension.ExtendWith;
 
 import org.springframework.beans.factory.config.BeanDefinition;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.task.SimpleAsyncTaskExecutorBuilder;
 import org.springframework.boot.task.TaskExecutorBuilder;
 import org.springframework.boot.task.TaskExecutorCustomizer;
 import org.springframework.boot.task.ThreadPoolTaskExecutorBuilder;
@@ -50,6 +51,7 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.Mockito.mock;
 
@@ -73,6 +75,7 @@ class TaskExecutionAutoConfigurationTests {
 			assertThat(context).hasSingleBean(TaskExecutorBuilder.class);
 			assertThat(context).hasSingleBean(ThreadPoolTaskExecutorBuilder.class);
 			assertThat(context).hasSingleBean(ThreadPoolTaskExecutor.class);
+			assertThat(context).hasSingleBean(SimpleAsyncTaskExecutorBuilder.class);
 		});
 	}
 
@@ -106,6 +109,17 @@ class TaskExecutionAutoConfigurationTests {
 			}));
 	}
 
+	@Test
+	void simpleAsyncTaskExecutorBuilderShouldReadProperties() {
+		this.contextRunner
+			.withPropertyValues("spring.task.execution.thread-name-prefix=mytest-",
+					"spring.task.execution.simple.concurrency-limit=1")
+			.run(assertSimpleAsyncTaskExecutor((taskExecutor) -> {
+				assertThat(taskExecutor.getConcurrencyLimit()).isEqualTo(1);
+				assertThat(taskExecutor.getThreadNamePrefix()).isEqualTo("mytest-");
+			}));
+	}
+
 	@Test
 	void threadPoolTaskExecutorBuilderShouldApplyCustomSettings() {
 		this.contextRunner
@@ -220,6 +234,23 @@ class TaskExecutionAutoConfigurationTests {
 			});
 	}
 
+	@Test
+	void simpleAsyncTaskExecutorBuilderUsesPlatformThreadsByDefault() {
+		this.contextRunner.run((context) -> {
+			SimpleAsyncTaskExecutorBuilder builder = context.getBean(SimpleAsyncTaskExecutorBuilder.class);
+			assertThat(builder).hasFieldOrPropertyWithValue("virtualThreads", null);
+		});
+	}
+
+	@Test
+	@EnabledForJreRange(min = JRE.JAVA_21)
+	void simpleAsyncTaskExecutorBuilderUsesVirtualThreadsWhenEnabled() {
+		this.contextRunner.withPropertyValues("spring.threads.virtual.enabled=true").run((context) -> {
+			SimpleAsyncTaskExecutorBuilder builder = context.getBean(SimpleAsyncTaskExecutorBuilder.class);
+			assertThat(builder).hasFieldOrPropertyWithValue("virtualThreads", true);
+		});
+	}
+
 	@Test
 	void taskExecutorWhenHasCustomTaskExecutorShouldBackOff() {
 		this.contextRunner.withUserConfiguration(CustomTaskExecutorConfig.class).run((context) -> {
@@ -318,6 +349,15 @@ class TaskExecutionAutoConfigurationTests {
 		};
 	}
 
+	private ContextConsumer<AssertableApplicationContext> assertSimpleAsyncTaskExecutor(
+			Consumer<SimpleAsyncTaskExecutor> taskExecutor) {
+		return (context) -> {
+			assertThat(context).hasSingleBean(SimpleAsyncTaskExecutorBuilder.class);
+			SimpleAsyncTaskExecutorBuilder builder = context.getBean(SimpleAsyncTaskExecutorBuilder.class);
+			taskExecutor.accept(builder.build());
+		};
+	}
+
 	private String virtualThreadName(SimpleAsyncTaskExecutor taskExecutor) throws InterruptedException {
 		AtomicReference<Thread> threadReference = new AtomicReference<>();
 		CountDownLatch latch = new CountDownLatch(1);
@@ -326,7 +366,9 @@ class TaskExecutionAutoConfigurationTests {
 			threadReference.set(currentThread);
 			latch.countDown();
 		});
-		latch.await(30, TimeUnit.SECONDS);
+		if (!latch.await(30, TimeUnit.SECONDS)) {
+			fail("Timeout while waiting for latch");
+		}
 		Thread thread = threadReference.get();
 		assertThat(thread).extracting("virtual").as("%s is virtual", thread).isEqualTo(true);
 		return thread.getName();