《Spring Boot 高级》5.3 阻塞代码的隔离

parallel distinct threads = 10
parallel thread name      = parallel-1
boundedElastic thread name = boundedElastic-1

public Mono<RiskResult> assess(String memberId) {
    return Mono.fromCallable(() -> legacyRiskClient.assess(memberId))  // 同步阻塞调用
            .subscribeOn(Schedulers.boundedElastic());                 // 换到弹性线程池执行
}

// 误用一：阻塞写在 fromCallable 之外 —— 调用发生在构造期，与调度无关
Mono<RiskResult> bad1 = Mono.just(legacyRiskClient.assess(memberId))
        .subscribeOn(Schedulers.boundedElastic());   // 太晚了，早就阻塞完了

// 误用二：subscribeOn 写在阻塞之后，指望它把阻塞挪走
Mono<RiskResult> bad2 = Mono.fromCallable(() -> legacyRiskClient.assess(memberId))
        .map(this::enrich)                  // 这里的执行线程由 subscribeOn 决定，没错
        .subscribeOn(Schedulers.parallel()); // 但换到 parallel 等于没换，还是事件循环

// 误用三：publishOn 写在阻塞之后 —— 只影响下游，阻塞仍在原线程
Mono<RiskResult> bad3 = Mono.fromCallable(() -> legacyRiskClient.assess(memberId))
        .publishOn(Schedulers.boundedElastic());   // 阻塞已经在上游线程跑完了

public class reactor.blockhound.BlockHound {
  public static reactor.blockhound.BlockHound$Builder builder();
  public static void install(reactor.blockhound.integration.BlockHoundIntegration... integrations);
  public static void premain(java.lang.String, java.lang.instrument.Instrumentation);
}

@BeforeAll
static void installBlockHound() {
    BlockHound.builder()
            .allowBlockingCallsInside("java.io.PrintStream", "write")   // 放行日志
            .install();
}

reactor.blockhound.BlockingOperationError: Blocking call! java.io.FileInputStream#readBytes
	at java.base/java.io.FileInputStream.readBytes(FileInputStream.java)
	at com.example.borrowing.LegacyRiskClient.assess(LegacyRiskClient.java:42)
	at com.example.borrowing.BorrowingService.assess(BorrowingService.java:31)
	...

default void configureBlockingExecution(org.springframework.web.reactive.config.BlockingExecutionConfigurer configurer);

@Configuration
public class WebFluxBlockingConfig implements WebFluxConfigurer {

    @Override
    public void configureBlockingExecution(BlockingExecutionConfigurer configurer) {
        configurer.setControllerMethodPredicate(handlerMethod ->
                handlerMethod.hasMethodAnnotation(Blocking.class));
        // 默认执行器即 boundedElastic；如需自定义可 setExecutor(...)
    }
}

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Blocking {
}

export JAVA_HOME=/tmp/springboot_book/jdk-21.0.12.1+1/Contents/Home
J=/tmp/springboot_book/jars

# 1) 调度器常量与工厂方法
"$JAVA_HOME/bin/javap" -cp "$J/reactor-core-3.8.7.jar" reactor.core.scheduler.Schedulers

# 2) 打印默认常量值（本机实测：SIZE=100、QUEUESIZE=100000、ON_VIRTUAL_THREADS=false）
cat > /tmp/Probe.java <<'EOF'
import reactor.core.scheduler.Schedulers;
public class Probe {
  public static void main(String[] a) {
    System.out.println("SIZE=" + Schedulers.DEFAULT_BOUNDED_ELASTIC_SIZE);
    System.out.println("QUEUESIZE=" + Schedulers.DEFAULT_BOUNDED_ELASTIC_QUEUESIZE);
    System.out.println("ON_VT=" + Schedulers.DEFAULT_BOUNDED_ELASTIC_ON_VIRTUAL_THREADS);
  }
}
EOF
"$JAVA_HOME/bin/javac" -cp "$J/reactor-core-3.8.7.jar" -d /tmp /tmp/Probe.java
"$JAVA_HOME/bin/java" -cp "$J/reactor-core-3.8.7.jar:/tmp" Probe

# 3) BlockHound 的类与配置项
"$JAVA_HOME/bin/javap" -cp "$J/blockhound-1.0.13.RELEASE.jar" reactor.blockhound.BlockHound
"$JAVA_HOME/bin/javap" -cp "$J/blockhound-1.0.13.RELEASE.jar" 'reactor.blockhound.BlockHound$Builder'

# 4) WebFlux 的阻塞执行配置点
"$JAVA_HOME/bin/javap" -cp "$J/spring-webflux-7.0.9.jar" \
  org.springframework.web.reactive.config.BlockingExecutionConfigurer