@Async ve Thread Havuzu — Max Neden Hiç Dolmuyor?

1
@Configuration
2
@EnableAsync
3
class AsyncConfig implements AsyncConfigurer {
4

5
    private static final Logger log = LoggerFactory.getLogger(AsyncConfig.class);
6

7
    @Bean(name = "notificationExecutor")
8
    ThreadPoolTaskExecutor notificationExecutor() {
9
        var executor = new ThreadPoolTaskExecutor();
10
        executor.setCorePoolSize(4);
11
        executor.setMaxPoolSize(8);
12
        executor.setQueueCapacity(200);             // bounded: memory stays flat on salary day
13
        executor.setThreadNamePrefix("notify-");
14
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
15
        executor.setTaskDecorator(new MdcTaskDecorator());
16
        executor.setWaitForTasksToCompleteOnShutdown(true);
17
        executor.setAwaitTerminationSeconds(20);
18
        return executor;
19
    }
20

21
    // void @Async methods have no caller to throw to; log them with context.
22
    // Params may hold phone numbers: log the method only, the MDC carries the trace.
23
    @Override
24
    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
25
        return (ex, method, params) -> log.error("async {} failed", method.getName(), ex);
26
    }
27
}

1
class MdcTaskDecorator implements TaskDecorator {
2

3
    @Override
4
    public Runnable decorate(Runnable task) {
5
        Map<String, String> context = MDC.getCopyOfContextMap();  // captured on the caller
6
        return () -> {
7
            if (context != null) MDC.setContextMap(context);
8
            try {
9
                task.run();
10
            } finally {
11
                MDC.clear();                                        // pooled thread: clean up
12
            }
13
        };
14
    }
15
}

1
@Service
2
class TransferService {
3

4
    private final Ledger ledger;
5
    private final ApplicationEventPublisher events;
6

7
    TransferService(Ledger ledger, ApplicationEventPublisher events) {
8
        this.ledger = ledger;
9
        this.events = events;
10
    }
11

12
    @Transactional
13
    public long transfer(TransferCommand cmd) {
14
        long entryId = ledger.post(cmd.fromIban(), cmd.toIban(), cmd.amount());
15
        events.publishEvent(new TransferCompleted(entryId, cmd.customerId(), cmd.fromIban(), cmd.amount()));
16
        return entryId;
17
    }
18
}
19

20
record TransferCompleted(long entryId, long customerId, String fromIban, BigDecimal amount) {}

1
@Component
2
class TransferNotifier {
3

4
    private static final Logger log = LoggerFactory.getLogger(TransferNotifier.class);
5

6
    private final SmsGateway sms;
7
    private final CustomerContacts contacts;
8

9
    TransferNotifier(SmsGateway sms, CustomerContacts contacts) {
10
        this.sms = sms;
11
        this.contacts = contacts;
12
    }
13

14
    // AFTER_COMMIT: the money has really moved before the customer is told so.
15
    // @Async("notificationExecutor"): the request thread is free as soon as the event is queued.
16
    @Async("notificationExecutor")
17
    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
18
    public void onTransfer(TransferCompleted event) {
19
        log.info("notifying customer {} about entry {}", event.customerId(), event.entryId());   // MDC traceId is here
20
        String masked = "…" + event.fromIban().substring(event.fromIban().length() - 4);
21
        sms.send(contacts.phoneOf(event.customerId()),
22
                "Hesabınızdan (%s) %s TL transfer yapıldı.".formatted(masked, event.amount().toPlainString()));
23
    }
24
}

1
@RestController
2
@RequestMapping("/transfers")
3
class TransferController {
4

5
    private final TransferService transfers;
6

7
    TransferController(TransferService transfers) {
8
        this.transfers = transfers;
9
    }
10

11
    @PostMapping
12
    ResponseEntity<Void> transfer(@Valid @RequestBody TransferCommand cmd) {
13
        long id = transfers.transfer(cmd);
14
        return ResponseEntity.created(URI.create("/transfers/" + id)).build();   // SMS still on its way
15
    }
16
}

1
spring:
2
  task:
3
    execution:
4
      thread-name-prefix: app-
5
      pool:
6
        core-size: 8
7
        max-size: 16
8
        queue-capacity: 500      # default is unbounded
9
      shutdown:
10
        await-termination: true
11
        await-termination-period: 20s