Add the async module
Every @Async behaviour that surprises people, asserted by a test and captured to docs/output/: the self-invocation trap, what CGLIB cannot override, the IllegalArgumentException a plain return type throws, the unbounded queue that makes max-size decoration, spring.task.execution.propagate-context (new in Boot 4.1.0), the two Executor beans that leave @Async on an unpooled SimpleAsyncTaskExecutor, and JEP 491 measured on JDK 21 against JDK 25.
This commit is contained in:
@@ -0,0 +1,22 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.scheduling.annotation.EnableAsync;
|
||||
|
||||
/**
|
||||
* Entry point for the @Async demonstrations.
|
||||
*
|
||||
* <p>{@code @EnableAsync} is what installs {@code AsyncAnnotationBeanPostProcessor}. Spring Boot
|
||||
* does <em>not</em> turn it on for you: without this annotation every {@code @Async} method in
|
||||
* this project runs on the caller's thread and nothing warns you. That is the first and most
|
||||
* boring way to lose a day — see docs/01-what-async-actually-does.md.
|
||||
*/
|
||||
@SpringBootApplication
|
||||
@EnableAsync
|
||||
public class AsyncDemoApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(AsyncDemoApplication.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/** Reports what the async thread can see of the caller's {@link RequestId}. */
|
||||
@Service
|
||||
public class ContextService {
|
||||
|
||||
@Async
|
||||
public CompletableFuture<String> readRequestId() {
|
||||
return CompletableFuture.completedFuture(String.valueOf(RequestId.get()));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.core.task.TaskExecutor;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* Prints the executors that actually exist in this context, rather than the ones you believe
|
||||
* exist. Runtime state here is otherwise invisible: an {@code @Async} call that silently landed
|
||||
* on {@code SimpleAsyncTaskExecutor} looks exactly like one that landed on a tuned pool until
|
||||
* you read a thread name under load.
|
||||
*
|
||||
* <p>Delete this from anything you ship — it is a diagnostic, not a feature.
|
||||
*/
|
||||
@Component
|
||||
public class ExecutorDiagnostics {
|
||||
|
||||
private final ApplicationContext context;
|
||||
|
||||
public ExecutorDiagnostics(ApplicationContext context) {
|
||||
this.context = context;
|
||||
}
|
||||
|
||||
public String report() {
|
||||
StringBuilder out = new StringBuilder();
|
||||
Map<String, Executor> executors = this.context.getBeansOfType(Executor.class);
|
||||
out.append("Executor beans in this context: ").append(executors.size()).append('\n');
|
||||
executors.forEach((name, executor) -> {
|
||||
out.append(" ").append(name)
|
||||
.append(" -> ").append(executor.getClass().getName())
|
||||
.append(executor instanceof TaskExecutor ? " [TaskExecutor]" : "")
|
||||
.append('\n');
|
||||
if (executor instanceof ThreadPoolTaskExecutor pool) {
|
||||
out.append(" corePoolSize=").append(pool.getCorePoolSize())
|
||||
.append(" maxPoolSize=").append(pool.getMaxPoolSize())
|
||||
.append(" queueCapacity=").append(pool.getQueueCapacity())
|
||||
.append(" threadNamePrefix=").append(pool.getThreadNamePrefix())
|
||||
.append('\n');
|
||||
}
|
||||
});
|
||||
out.append("Bean named 'taskExecutor' present: ")
|
||||
.append(this.context.containsBean("taskExecutor")).append('\n');
|
||||
out.append("Bean named 'applicationTaskExecutor' present: ")
|
||||
.append(this.context.containsBean("applicationTaskExecutor")).append('\n');
|
||||
return out.toString();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
|
||||
|
||||
/**
|
||||
* Three ways to have an {@code Executor} bean, with three different outcomes for {@code @Async}.
|
||||
*
|
||||
* <p>Boot's {@code applicationTaskExecutor} is conditional on there being no {@code Executor}
|
||||
* bean already (see {@code TaskExecutorConfigurations.OnExecutorCondition}). Declaring one of
|
||||
* your own therefore removes Boot's, and what {@code @Async} does next depends entirely on how
|
||||
* many candidates are left.
|
||||
*
|
||||
* <p>See docs/07-which-executor-runs-it.md.
|
||||
*/
|
||||
@Configuration(proxyBeanMethods = false)
|
||||
public class ExecutorSelectionConfig {
|
||||
|
||||
/**
|
||||
* One custom executor. Boot backs off, this is the only candidate, {@code @Async} uses it.
|
||||
* Threads are named {@code mine-N}.
|
||||
*/
|
||||
@Bean
|
||||
@Profile("ownexecutor")
|
||||
public Executor myExecutor() {
|
||||
return build("mine-");
|
||||
}
|
||||
|
||||
/**
|
||||
* Two custom executors and neither is called {@code taskExecutor}. Boot still backs off, and
|
||||
* {@code AsyncExecutionAspectSupport} cannot pick between them: it catches the
|
||||
* {@code NoUniqueBeanDefinitionException}, looks for a bean literally named
|
||||
* {@code taskExecutor}, does not find one, and falls back to a plain
|
||||
* {@code SimpleAsyncTaskExecutor} — a new unpooled thread for every call, forever.
|
||||
*/
|
||||
@Bean
|
||||
@Profile("twoexecutors")
|
||||
public Executor reportsExecutor() {
|
||||
return build("reports-");
|
||||
}
|
||||
|
||||
@Bean
|
||||
@Profile("twoexecutors")
|
||||
public Executor emailsExecutor() {
|
||||
return build("emails-");
|
||||
}
|
||||
|
||||
private static ThreadPoolTaskExecutor build(String prefix) {
|
||||
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
|
||||
executor.setCorePoolSize(2);
|
||||
executor.setMaxPoolSize(2);
|
||||
executor.setThreadNamePrefix(prefix);
|
||||
executor.initialize();
|
||||
return executor;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,77 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Measures whether blocking inside a {@code synchronized} block pins the carrier thread.
|
||||
*
|
||||
* <p>Every guide written before JDK 24 tells you not to combine virtual threads with
|
||||
* {@code synchronized}, because a virtual thread that blocks while holding a monitor pins its
|
||||
* carrier and the carrier cannot run anything else. JEP 491 (JDK 24) removed that. This probe is
|
||||
* how you check rather than believe: it runs the same 32 tasks twice, once inside a
|
||||
* {@code synchronized} block on a private monitor and once without, with the scheduler limited
|
||||
* to two carrier threads.
|
||||
*
|
||||
* <p>If pinning happens, the synchronized run takes roughly {@code tasks / carriers * sleep}.
|
||||
* If it does not, both runs take roughly {@code sleep}.
|
||||
*
|
||||
* <p>Run it with scripts/pinning-probe.sh, which sets the carrier count. See
|
||||
* docs/08-virtual-threads-and-pinning.md.
|
||||
*/
|
||||
public final class PinningProbe {
|
||||
|
||||
private static final int TASKS = 32;
|
||||
|
||||
private static final Duration SLEEP = Duration.ofMillis(300);
|
||||
|
||||
public static void main(String[] args) throws Exception {
|
||||
System.out.println("java.version : " + System.getProperty("java.version"));
|
||||
System.out.println("jdk.virtualThreadScheduler.parallelism: "
|
||||
+ System.getProperty("jdk.virtualThreadScheduler.parallelism", "<unset>"));
|
||||
System.out.println("tasks=" + TASKS + " sleep=" + SLEEP.toMillis() + "ms");
|
||||
System.out.println();
|
||||
|
||||
long plain = run(false);
|
||||
long guarded = run(true);
|
||||
|
||||
System.out.printf("no monitor held : %5d ms%n", plain);
|
||||
System.out.printf("blocked inside synchronized: %5d ms%n", guarded);
|
||||
System.out.println();
|
||||
System.out.println("Pinned would be about " + (TASKS / 2) * SLEEP.toMillis()
|
||||
+ " ms for the synchronized run (32 tasks / 2 carriers x 300 ms).");
|
||||
System.out.println("Not pinned is about " + SLEEP.toMillis() + " ms for both.");
|
||||
}
|
||||
|
||||
private static long run(boolean holdMonitor) throws InterruptedException {
|
||||
List<Thread> threads = new ArrayList<>();
|
||||
long start = System.nanoTime();
|
||||
for (int i = 0; i < TASKS; i++) {
|
||||
threads.add(Thread.ofVirtual().start(() -> task(holdMonitor)));
|
||||
}
|
||||
for (Thread t : threads) {
|
||||
t.join();
|
||||
}
|
||||
return Duration.ofNanos(System.nanoTime() - start).toMillis();
|
||||
}
|
||||
|
||||
private static void task(boolean holdMonitor) {
|
||||
// A monitor per task, so nothing here is contended. The question is only whether the
|
||||
// carrier thread is released while this virtual thread sleeps.
|
||||
Object monitor = new Object();
|
||||
try {
|
||||
if (holdMonitor) {
|
||||
synchronized (monitor) {
|
||||
Thread.sleep(SLEEP);
|
||||
}
|
||||
}
|
||||
else {
|
||||
Thread.sleep(SLEEP);
|
||||
}
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,69 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import io.micrometer.context.ContextRegistry;
|
||||
import io.micrometer.context.ThreadLocalAccessor;
|
||||
|
||||
/**
|
||||
* A ThreadLocal that survives the hop onto an {@code @Async} thread — but only because it
|
||||
* is registered with micrometer's {@link ContextRegistry} and only when
|
||||
* {@code spring.task.execution.propagate-context} is on.
|
||||
*
|
||||
* <p>{@code propagate-context} is new in Spring Boot 4.1.0. It was not present in 4.0.8 (checked
|
||||
* by decompiling {@code TaskExecutionProperties} from both jars). It makes Boot decorate the
|
||||
* auto-configured executor with {@code ContextPropagatingTaskDecorator}, which captures whatever
|
||||
* is registered here at submission time and restores it around the task.
|
||||
*
|
||||
* <p>See docs/06-context-propagation.md.
|
||||
*/
|
||||
public final class RequestId {
|
||||
|
||||
public static final String KEY = "requestId";
|
||||
|
||||
private static final ThreadLocal<String> HOLDER = new ThreadLocal<>();
|
||||
|
||||
static {
|
||||
ContextRegistry.getInstance().registerThreadLocalAccessor(new Accessor());
|
||||
}
|
||||
|
||||
private RequestId() {
|
||||
}
|
||||
|
||||
public static void set(String value) {
|
||||
HOLDER.set(value);
|
||||
}
|
||||
|
||||
public static String get() {
|
||||
return HOLDER.get();
|
||||
}
|
||||
|
||||
public static void clear() {
|
||||
HOLDER.remove();
|
||||
}
|
||||
|
||||
/** Touching this class runs the static initialiser that registers the accessor. */
|
||||
public static void register() {
|
||||
}
|
||||
|
||||
private static final class Accessor implements ThreadLocalAccessor<String> {
|
||||
|
||||
@Override
|
||||
public Object key() {
|
||||
return KEY;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getValue() {
|
||||
return HOLDER.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setValue(String value) {
|
||||
HOLDER.set(value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setValue() {
|
||||
HOLDER.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* Return types {@code @Async} supports, and the one that quietly returns null.
|
||||
*
|
||||
* <p>{@code AsyncExecutionAspectSupport.doSubmit} branches on the declared return type.
|
||||
* {@code CompletableFuture} and {@code Future} are submitted and the handle is returned;
|
||||
* {@code void} is submitted and {@code null} returned. Anything else — including a plain
|
||||
* {@code String} — falls into the final {@code else} and throws
|
||||
* {@code IllegalArgumentException: Invalid return type for async method (only Future and void
|
||||
* supported)}. It is thrown to the caller, at call time, on every invocation. The widely
|
||||
* repeated claim that such a method “returns null” is wrong; the method body never
|
||||
* runs at all.
|
||||
*
|
||||
* <p>See docs/04-return-types-and-exceptions.md.
|
||||
*/
|
||||
@Service
|
||||
public class ReturnTypeService {
|
||||
|
||||
@Async
|
||||
public CompletableFuture<String> completableFuture() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/**
|
||||
* Compiles cleanly and throws {@code IllegalArgumentException} on every call. Nothing about
|
||||
* the signature is rejected at startup, so this is a runtime failure in whichever code path
|
||||
* happens to reach it first.
|
||||
*/
|
||||
@Async
|
||||
public String plainString() {
|
||||
return Threads.describe();
|
||||
}
|
||||
|
||||
/** Exception from a void method: the caller cannot see it. */
|
||||
@Async
|
||||
public void voidThatThrows() {
|
||||
throw new IllegalStateException("thrown from a void @Async method");
|
||||
}
|
||||
|
||||
/** Exception from a future-returning method: the caller can see it, if it looks. */
|
||||
@Async
|
||||
public CompletableFuture<String> futureThatThrows() {
|
||||
throw new IllegalStateException("thrown from a CompletableFuture @Async method");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentSkipListSet;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* Where the pool actually grows, and where it does not.
|
||||
*
|
||||
* <p>{@code ThreadPoolTaskExecutor} wraps a {@code ThreadPoolExecutor}, and {@code
|
||||
* ThreadPoolExecutor} only creates a thread beyond the core size when the queue refuses a task.
|
||||
* Spring Boot's default queue capacity is unbounded, so it never refuses, so the pool never grows
|
||||
* past {@code spring.task.execution.pool.core-size} — whatever {@code max-size} says.
|
||||
*
|
||||
* <p>Each method records the distinct thread names it ran on. Counting them is the measurement.
|
||||
*
|
||||
* <p>See docs/05-pool-sizing.md.
|
||||
*/
|
||||
@Service
|
||||
public class SaturationService {
|
||||
|
||||
private final Set<String> threadsSeen = new ConcurrentSkipListSet<>();
|
||||
|
||||
/**
|
||||
* Blocks until released, so every concurrently running task holds its thread. That is what
|
||||
* makes the number of distinct threads equal to the pool's real concurrency.
|
||||
*/
|
||||
@Async
|
||||
public void occupyAThread(CountDownLatch arrived, CountDownLatch release) {
|
||||
threadsSeen.add(Threads.name());
|
||||
arrived.countDown();
|
||||
try {
|
||||
release.await(30, TimeUnit.SECONDS);
|
||||
}
|
||||
catch (InterruptedException ex) {
|
||||
Thread.currentThread().interrupt();
|
||||
}
|
||||
}
|
||||
|
||||
public Set<String> threadsSeen() {
|
||||
return Set.copyOf(this.threadsSeen);
|
||||
}
|
||||
|
||||
public void reset() {
|
||||
this.threadsSeen.clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* The self-invocation trap, and the two ways out of it.
|
||||
*
|
||||
* <p>{@code @Async} is implemented by a proxy that wraps this bean. Every call that arrives from
|
||||
* outside goes through the proxy and gets handed to an executor. Every call that starts
|
||||
* <em>inside</em> this class is a plain {@code this.method()} invocation on the target object, so
|
||||
* the proxy is never consulted and the method runs on the caller's thread. Nothing is logged.
|
||||
*
|
||||
* <p>See docs/02-the-self-invocation-trap.md.
|
||||
*/
|
||||
@Service
|
||||
public class SelfInvocationService {
|
||||
|
||||
private final ObjectProvider<SelfInvocationService> self;
|
||||
|
||||
/**
|
||||
* An {@code ObjectProvider} rather than a constructor-injected {@code SelfInvocationService}:
|
||||
* the latter is a self-reference that Spring resolves, but it hands you the proxy only
|
||||
* because of a special case, and it makes the cycle explicit in a way reviewers argue about.
|
||||
* {@code ObjectProvider} is lazy, so there is no cycle to resolve at construction time.
|
||||
*/
|
||||
public SelfInvocationService(ObjectProvider<SelfInvocationService> self) {
|
||||
this.self = self;
|
||||
}
|
||||
|
||||
/**
|
||||
* The annotated method. Reached through the proxy, it runs on an executor thread.
|
||||
*/
|
||||
@Async
|
||||
public CompletableFuture<String> annotated() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/**
|
||||
* The trap. {@code annotated()} here is {@code this.annotated()}, which never touches the
|
||||
* proxy, so the work runs on whatever thread called {@code viaSelfInvocation()} and the
|
||||
* returned future is already complete.
|
||||
*/
|
||||
public CompletableFuture<String> viaSelfInvocation() {
|
||||
return annotated();
|
||||
}
|
||||
|
||||
/**
|
||||
* The fix that keeps the code in one class: go out through the proxy deliberately.
|
||||
*/
|
||||
public CompletableFuture<String> viaSelfReference() {
|
||||
return self.getObject().annotated();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
/**
|
||||
* Thread description used by every transcript in docs/output/.
|
||||
*
|
||||
* <p>The whole project is built on one idea: instead of asserting which executor Spring picked,
|
||||
* print the name of the thread the method actually ran on. Thread names are the only honest
|
||||
* evidence, because {@code @Async} failing silently and {@code @Async} working look identical
|
||||
* from the outside.
|
||||
*/
|
||||
public final class Threads {
|
||||
|
||||
private Threads() {
|
||||
}
|
||||
|
||||
/** e.g. {@code task-1 (virtual=false)} or {@code VirtualThread[#42]/runnable (virtual=true)}. */
|
||||
public static String describe() {
|
||||
Thread t = Thread.currentThread();
|
||||
return t.getName() + " (virtual=" + t.isVirtual() + ")";
|
||||
}
|
||||
|
||||
/** Just the name, for assertions that care about the executor's thread-name prefix. */
|
||||
public static String name() {
|
||||
return Thread.currentThread().getName();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,49 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
|
||||
import org.springframework.scheduling.annotation.Async;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* What CGLIB can and cannot override.
|
||||
*
|
||||
* <p>Spring Boot proxies with CGLIB by default ({@code spring.aop.proxy-target-class} is true),
|
||||
* which means the proxy is a generated <em>subclass</em>. A subclass can override a public or
|
||||
* protected method. It cannot override a {@code final} one, and it cannot see a private one.
|
||||
* Both of those {@code @Async} annotations are therefore inert, and neither produces a warning.
|
||||
*
|
||||
* <p>See docs/03-what-the-proxy-cannot-see.md.
|
||||
*/
|
||||
@Service
|
||||
public class VisibilityService {
|
||||
|
||||
@Async
|
||||
public CompletableFuture<String> publicMethod() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/** Cannot be overridden by the CGLIB subclass, so the advice never runs. */
|
||||
@Async
|
||||
public final CompletableFuture<String> finalMethod() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/** Protected methods <em>can</em> be overridden by a subclass, so this one is advised. */
|
||||
@Async
|
||||
protected CompletableFuture<String> protectedMethod() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/** Reachable only from inside the class, so it is a self-invocation as well as private. */
|
||||
@Async
|
||||
@SuppressWarnings("unused")
|
||||
private CompletableFuture<String> privateMethod() {
|
||||
return CompletableFuture.completedFuture(Threads.describe());
|
||||
}
|
||||
|
||||
/** Calls the protected one from outside its own {@code this}, via the caller's proxy. */
|
||||
public CompletableFuture<String> callProtectedInternally() {
|
||||
return protectedMethod();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
spring:
|
||||
application:
|
||||
name: async
|
||||
# Nothing is set here on purpose. Every number in docs/output/executor-report.txt is a Spring
|
||||
# Boot default, printed from the live bean rather than copied out of the reference guide.
|
||||
# The profiles that change behaviour are in src/test/resources/ and in scripts/.
|
||||
@@ -0,0 +1,43 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** The same pool with a bounded queue. One property changes; the concurrency triples. */
|
||||
@SpringBootTest(properties = {
|
||||
"spring.task.execution.pool.core-size=4",
|
||||
"spring.task.execution.pool.max-size=12",
|
||||
"spring.task.execution.pool.queue-capacity=4" })
|
||||
class BoundedQueueTest {
|
||||
|
||||
@Autowired
|
||||
private SaturationService service;
|
||||
|
||||
@Test
|
||||
void aBoundedQueueIsWhatLetsThePoolGrow() throws Exception {
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch arrived = new CountDownLatch(12);
|
||||
service.reset();
|
||||
for (int i = 0; i < 16; i++) {
|
||||
service.occupyAThread(arrived, release);
|
||||
}
|
||||
assertThat(arrived.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
Thread.sleep(500);
|
||||
int threads = service.threadsSeen().size();
|
||||
release.countDown();
|
||||
|
||||
assertThat(threads).isEqualTo(12);
|
||||
Capture.write("pool-bounded-queue.txt",
|
||||
"core-size=4, max-size=12, queue-capacity=4, 16 blocking tasks",
|
||||
"distinct threads that ran a task : " + threads + "\n"
|
||||
+ "thread names : " + service.threadsSeen() + "\n"
|
||||
+ "4 core threads, 4 tasks queued, 8 more threads created up to max-size.\n");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
|
||||
/** Writes a transcript under docs/output/ so every number in the article has a file behind it. */
|
||||
final class Capture {
|
||||
|
||||
private Capture() {
|
||||
}
|
||||
|
||||
static void write(String fileName, String heading, String body) {
|
||||
Path dir = Path.of(System.getProperty("user.dir"), "docs", "output");
|
||||
try {
|
||||
Files.createDirectories(dir);
|
||||
Files.writeString(dir.resolve(fileName), "== " + heading + " ==\n\n" + body + "\n");
|
||||
}
|
||||
catch (IOException ex) {
|
||||
throw new IllegalStateException("could not write " + fileName, ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,27 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** The default: a ThreadLocal set by the caller is not visible to the @Async thread. */
|
||||
@SpringBootTest
|
||||
class ContextNotPropagatedTest {
|
||||
|
||||
@Autowired
|
||||
private ContextService service;
|
||||
|
||||
@Test
|
||||
void theAsyncThreadSeesNothing() throws Exception {
|
||||
RequestId.set("req-4711");
|
||||
try {
|
||||
assertThat(service.readRequestId().get()).isEqualTo("null");
|
||||
}
|
||||
finally {
|
||||
RequestId.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* {@code spring.task.execution.propagate-context=true} — new in Spring Boot 4.1.0 —
|
||||
* decorates the auto-configured executor so registered ThreadLocals survive the hop.
|
||||
*/
|
||||
@SpringBootTest(properties = "spring.task.execution.propagate-context=true")
|
||||
class ContextPropagatedTest {
|
||||
|
||||
@Autowired
|
||||
private ContextService service;
|
||||
|
||||
@Test
|
||||
void onePropertyCarriesTheThreadLocalAcross() throws Exception {
|
||||
RequestId.register();
|
||||
RequestId.set("req-4711");
|
||||
try {
|
||||
String seen = service.readRequestId().get();
|
||||
assertThat(seen).isEqualTo("req-4711");
|
||||
Capture.write("context-propagation.txt",
|
||||
"spring.task.execution.propagate-context (new in Boot 4.1.0)",
|
||||
"caller thread : " + Threads.describe() + "\n"
|
||||
+ "RequestId set : req-4711\n"
|
||||
+ "@Async thread saw: " + seen + "\n");
|
||||
}
|
||||
finally {
|
||||
RequestId.clear();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
/** Prints the stock context so the article's "defaults" table has a file behind it. */
|
||||
@SpringBootTest
|
||||
class ExecutorReportTest {
|
||||
|
||||
@Autowired
|
||||
private ExecutorDiagnostics diagnostics;
|
||||
|
||||
@Test
|
||||
void printStockContext() {
|
||||
Capture.write("executor-report.txt",
|
||||
"Stock Spring Boot 4.1.1 context, nothing configured", this.diagnostics.report());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* The same two custom executors, plus {@code spring.task.execution.mode=force}. Boot creates
|
||||
* {@code applicationTaskExecutor} anyway and marks it as the one @Async should use.
|
||||
*/
|
||||
@SpringBootTest(properties = "spring.task.execution.mode=force")
|
||||
@ActiveProfiles("twoexecutors")
|
||||
class ForceModeTest {
|
||||
|
||||
@Autowired
|
||||
private ReturnTypeService service;
|
||||
|
||||
@Autowired
|
||||
private ExecutorDiagnostics diagnostics;
|
||||
|
||||
@Test
|
||||
void forceRestoresTheApplicationTaskExecutor() throws Exception {
|
||||
String ranOn = service.completableFuture().get();
|
||||
assertThat(ranOn).startsWith("task-");
|
||||
Capture.write("executor-force-mode.txt",
|
||||
"Two custom Executor beans plus spring.task.execution.mode=force",
|
||||
this.diagnostics.report() + "\n@Async ran on : " + ranOn + "\n");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** One Executor bean of your own: Boot backs off and @Async uses yours. */
|
||||
@SpringBootTest
|
||||
@ActiveProfiles("ownexecutor")
|
||||
class OwnExecutorTest {
|
||||
|
||||
@Autowired
|
||||
private ReturnTypeService service;
|
||||
|
||||
@Autowired
|
||||
private ExecutorDiagnostics diagnostics;
|
||||
|
||||
@Test
|
||||
void yourExecutorWins() throws Exception {
|
||||
String ranOn = service.completableFuture().get();
|
||||
assertThat(ranOn).startsWith("mine-");
|
||||
Capture.write("executor-one-custom.txt", "A single custom Executor bean",
|
||||
this.diagnostics.report() + "\n@Async ran on : " + ranOn + "\n");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.assertj.core.api.Assertions.assertThatCode;
|
||||
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
||||
|
||||
/**
|
||||
* Return types, and where the exceptions go.
|
||||
*
|
||||
* <p>The interesting assertion is the first one. A very common claim is that an {@code @Async}
|
||||
* method with a plain return type quietly returns {@code null}; on Spring Framework 7.0.9 it
|
||||
* throws, at the call site, before the body has run.
|
||||
*/
|
||||
@SpringBootTest
|
||||
class ReturnTypeTest {
|
||||
|
||||
@Autowired
|
||||
private ReturnTypeService service;
|
||||
|
||||
@Test
|
||||
void aPlainReturnTypeThrowsAtTheCallSite() {
|
||||
assertThatThrownBy(() -> service.plainString())
|
||||
.isInstanceOf(IllegalArgumentException.class)
|
||||
.hasMessage("Invalid return type for async method (only Future and void supported): "
|
||||
+ "class java.lang.String");
|
||||
}
|
||||
|
||||
@Test
|
||||
void aFutureCarriesTheException() {
|
||||
CompletableFuture<String> future = service.futureThatThrows();
|
||||
assertThatThrownBy(() -> future.get(5, TimeUnit.SECONDS))
|
||||
.isInstanceOf(ExecutionException.class)
|
||||
.hasCauseInstanceOf(IllegalStateException.class);
|
||||
}
|
||||
|
||||
@Test
|
||||
void aVoidMethodThrowsNothingTheCallerCanSee() {
|
||||
assertThatCode(() -> service.voidThatThrows()).doesNotThrowAnyException();
|
||||
}
|
||||
|
||||
@Test
|
||||
void capture() throws Exception {
|
||||
String plain;
|
||||
try {
|
||||
plain = String.valueOf(service.plainString());
|
||||
}
|
||||
catch (IllegalArgumentException ex) {
|
||||
plain = ex.getClass().getName() + ": " + ex.getMessage();
|
||||
}
|
||||
String futureOutcome;
|
||||
try {
|
||||
service.futureThatThrows().get(5, TimeUnit.SECONDS);
|
||||
futureOutcome = "completed normally";
|
||||
}
|
||||
catch (ExecutionException ex) {
|
||||
futureOutcome = ex.getCause().getClass().getSimpleName() + ": " + ex.getCause().getMessage();
|
||||
}
|
||||
service.voidThatThrows();
|
||||
String body = """
|
||||
completableFuture().get() : %s
|
||||
plainString() : %s
|
||||
futureThatThrows().get() : %s
|
||||
voidThatThrows() : returned normally. The exception went to
|
||||
SimpleAsyncUncaughtExceptionHandler, which logs it at
|
||||
ERROR under the logger
|
||||
o.s.a.i.SimpleAsyncUncaughtExceptionHandler and
|
||||
discards it. The caller is never told.
|
||||
""".formatted(service.completableFuture().get(), plain, futureOutcome);
|
||||
Capture.write("return-types.txt", "What each @Async return type hands back", body);
|
||||
assertThat(plain).contains("IllegalArgumentException");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Proves the self-invocation trap by thread name rather than by timing, because a task that runs
|
||||
* on the caller's thread and a task that runs quickly on a pool thread are indistinguishable by
|
||||
* clock.
|
||||
*/
|
||||
@SpringBootTest
|
||||
class SelfInvocationTest {
|
||||
|
||||
@Autowired
|
||||
private SelfInvocationService service;
|
||||
|
||||
@Test
|
||||
void throughTheProxyItIsAsynchronous() throws Exception {
|
||||
assertThat(service.annotated().get()).startsWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void selfInvocationSilentlyRunsOnTheCallerThread() throws ExecutionException, InterruptedException {
|
||||
String callerThread = Threads.name();
|
||||
String ranOn = service.viaSelfInvocation().get();
|
||||
|
||||
assertThat(ranOn).startsWith(callerThread);
|
||||
assertThat(ranOn).doesNotStartWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void goingBackOutThroughTheProxyRestoresIt() throws Exception {
|
||||
assertThat(service.viaSelfReference().get()).startsWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void capture() throws Exception {
|
||||
String body = """
|
||||
caller thread : %s
|
||||
service.annotated() : %s
|
||||
service.viaSelfInvocation() : %s
|
||||
service.viaSelfReference() : %s
|
||||
""".formatted(Threads.describe(), service.annotated().get(),
|
||||
service.viaSelfInvocation().get(), service.viaSelfReference().get());
|
||||
Capture.write("self-invocation.txt",
|
||||
"Which thread each call actually ran on (Boot 4.1.1, JDK 25)", body);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.test.context.ActiveProfiles;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* Two Executor beans and no bean named {@code taskExecutor}. Boot's applicationTaskExecutor is
|
||||
* gone, neither of yours can be chosen, and @Async falls back to an unpooled
|
||||
* SimpleAsyncTaskExecutor without saying so at any log level.
|
||||
*/
|
||||
@SpringBootTest
|
||||
@ActiveProfiles("twoexecutors")
|
||||
class TwoExecutorsTest {
|
||||
|
||||
@Autowired
|
||||
private ReturnTypeService service;
|
||||
|
||||
@Autowired
|
||||
private ExecutorDiagnostics diagnostics;
|
||||
|
||||
@Test
|
||||
void neitherOfYoursIsUsed() throws Exception {
|
||||
String ranOn = service.completableFuture().get();
|
||||
assertThat(ranOn).doesNotStartWith("reports-");
|
||||
assertThat(ranOn).doesNotStartWith("emails-");
|
||||
assertThat(ranOn).doesNotStartWith("task-");
|
||||
Capture.write("executor-two-custom.txt", "Two custom Executor beans, no 'taskExecutor'",
|
||||
this.diagnostics.report() + "\n@Async ran on : " + ranOn + "\n");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* core-size 4, max-size 12, and the Boot default unbounded queue. Sixteen blocking tasks are
|
||||
* submitted; four threads run them, because an unbounded queue never refuses a task and
|
||||
* {@code ThreadPoolExecutor} only creates a non-core thread when the queue refuses one.
|
||||
*/
|
||||
@SpringBootTest(properties = {
|
||||
"spring.task.execution.pool.core-size=4",
|
||||
"spring.task.execution.pool.max-size=12" })
|
||||
class UnboundedQueueTest {
|
||||
|
||||
@Autowired
|
||||
private SaturationService service;
|
||||
|
||||
@Test
|
||||
void maxSizeIsIgnoredWhenTheQueueIsUnbounded() throws Exception {
|
||||
CountDownLatch release = new CountDownLatch(1);
|
||||
CountDownLatch arrived = new CountDownLatch(4);
|
||||
service.reset();
|
||||
for (int i = 0; i < 16; i++) {
|
||||
service.occupyAThread(arrived, release);
|
||||
}
|
||||
assertThat(arrived.await(10, TimeUnit.SECONDS)).isTrue();
|
||||
Thread.sleep(500);
|
||||
int threads = service.threadsSeen().size();
|
||||
release.countDown();
|
||||
|
||||
assertThat(threads).isEqualTo(4);
|
||||
Capture.write("pool-unbounded-queue.txt",
|
||||
"core-size=4, max-size=12, queue-capacity=<unbounded default>, 16 blocking tasks",
|
||||
"distinct threads that ran a task : " + threads + "\n"
|
||||
+ "thread names : " + service.threadsSeen() + "\n"
|
||||
+ "max-size had no effect: the queue never refused a task.\n");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
import org.springframework.core.task.SimpleAsyncTaskExecutor;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* With virtual threads on, {@code applicationTaskExecutor} is a {@code SimpleAsyncTaskExecutor}
|
||||
* over virtual threads. Note what that costs: the pool properties are still accepted and now
|
||||
* mean nothing, and a {@code SimpleAsyncTaskExecutor} has no queue and, by default, no
|
||||
* concurrency limit at all.
|
||||
*/
|
||||
@SpringBootTest(properties = {
|
||||
"spring.threads.virtual.enabled=true",
|
||||
"spring.task.execution.pool.core-size=4",
|
||||
"spring.task.execution.pool.max-size=12" })
|
||||
class VirtualThreadsTest {
|
||||
|
||||
@Autowired
|
||||
private ReturnTypeService service;
|
||||
|
||||
@Autowired
|
||||
@Qualifier("applicationTaskExecutor")
|
||||
private Executor applicationTaskExecutor;
|
||||
|
||||
@Autowired
|
||||
private ExecutorDiagnostics diagnostics;
|
||||
|
||||
@Test
|
||||
void asyncRunsOnAVirtualThread() throws Exception {
|
||||
String ranOn = service.completableFuture().get();
|
||||
assertThat(ranOn).contains("virtual=true");
|
||||
assertThat(this.applicationTaskExecutor).isInstanceOf(SimpleAsyncTaskExecutor.class);
|
||||
|
||||
Capture.write("virtual-threads.txt",
|
||||
"spring.threads.virtual.enabled=true, with pool properties still set",
|
||||
"applicationTaskExecutor : " + this.applicationTaskExecutor.getClass().getName() + "\n"
|
||||
+ "@Async ran on : " + ranOn + "\n\n"
|
||||
+ this.diagnostics.report());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
package com.ankurm.async;
|
||||
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import org.springframework.aop.support.AopUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.context.SpringBootTest;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/** What the CGLIB subclass can and cannot override, measured. */
|
||||
@SpringBootTest
|
||||
class VisibilityTest {
|
||||
|
||||
@Autowired
|
||||
private VisibilityService service;
|
||||
|
||||
@Test
|
||||
void theBeanIsACglibProxy() {
|
||||
assertThat(AopUtils.isCglibProxy(this.service)).isTrue();
|
||||
}
|
||||
|
||||
@Test
|
||||
void publicMethodIsAdvised() throws Exception {
|
||||
assertThat(service.publicMethod().get()).startsWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void finalMethodIsSilentlySynchronous() throws Exception {
|
||||
assertThat(service.finalMethod().get()).doesNotStartWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void protectedMethodIsAdvisedWhenCalledThroughTheProxy() throws Exception {
|
||||
// Same package as the service, so the test can reach a protected method on the proxy.
|
||||
assertThat(service.protectedMethod().get()).startsWith("task-");
|
||||
}
|
||||
|
||||
@Test
|
||||
void capture() throws Exception {
|
||||
String body = """
|
||||
proxy class : %s
|
||||
isCglibProxy : %s
|
||||
publicMethod() : %s
|
||||
finalMethod() : %s
|
||||
protectedMethod() via the proxy : %s
|
||||
callProtectedInternally() : %s
|
||||
""".formatted(service.getClass().getName(), AopUtils.isCglibProxy(this.service),
|
||||
service.publicMethod().get(), service.finalMethod().get(),
|
||||
service.protectedMethod().get(), service.callProtectedInternally().get());
|
||||
Capture.write("visibility.txt", "@Async against final, protected and private methods", body);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user