☰
Reactor线程池切换publishOn与subscribeOn
2026/10/27 8:52:24 网站建设 项目流程

在 Reactor(Project Reactor)中,publishOn和subscribeOn是用于切换执行线程(调度器)的两个操作符,但它们的行为和用途截然不同。下面是它们的详细区别与使用说明。


🧠 一句话理解

  • **subscribeOn**:指定源(Publisher)创建 & 订阅逻辑运行在哪个线程。
  • **publishOn**:指定后续操作符运行在哪个线程(从当前开始切换)。

🔁 subscribeOn:影响订阅动作及其“上游”

Mono.fromCallable(()->{log.info("Generating value");return"Hello";}).subscribeOn(Schedulers.boundedElastic())// 👈 切换线程.map(v->{log.info("Mapping: {}",v);returnv.toUpperCase();});

📌 说明:

  • subscribeOn会影响:
    • 源数据的创建(如fromCallable、Mono.just)
    • 源的订阅行为
    • 上游所有逻辑(在链条中靠前)

🧪 输出结果中的 log 会显示Generating value与Mapping: xxx都在boundedElastic线程中执行。


🔀 publishOn:影响后续操作符

Mono.fromCallable(()->{log.info("Generating value");return"Hello";}).publishOn(Schedulers.parallel())// 👈 从这里切换线程.map(v->{log.info("Mapping: {}",v);returnv.toUpperCase();});

📌 说明:

  • publishOn只影响后续操作符的执行线程
  • 当前位置之后的操作符都在新的线程中执行
  • 当前位置之前的执行线程不会受影响

🧪 输出 log:

[main] Generating value [parallel-1] Mapping: Hello

⚠️ 两者配合使用

Mono.fromCallable(()->{log.info("Generating value");return"Hello";}).subscribeOn(Schedulers.boundedElastic()).publishOn(Schedulers.parallel()).map(v->{log.info("Mapping: {}",v);returnv.toUpperCase();});

🧪 输出 log:

[boundedElastic-1] Generating value [parallel-1] Mapping: Hello

🔍 总结对比表

特性subscribeOnpublishOn
生效位置整个流的最上游(首次调用生效)只影响之后的操作符
常用于指定数据源的线程(IO、慢操作)指定数据处理的线程
多次调用只有第一次调用有效(对源)可以多次调用,影响每一处之后
应用场景创建慢资源(数据库、文件等)控制不同阶段逻辑在不同线程
替代线程切换逻辑可以代替传统线程池包装异步逻辑用于流中间切线程做不同的隔离或调度

✅ 实践建议

  • 网络/IO/阻塞调用 ➜ 放在subscribeOn(Schedulers.boundedElastic())
  • 业务计算阶段 ➜ 用publishOn(Schedulers.parallel())来提高并发度
  • 不要滥用这两个操作符,频繁切线程反而会增加上下文切换成本

如你有具体的使用场景(如 Flux + WebClient、文件读取、Kafka 等),我可以针对场景详细分析。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询