java

13.1 eventd v1 架构与模块划分

按仓库当前真实状态介绍 eventd v1：父 POM + eventd-core + eventd-app 三个模块，eventd-core 目前只有 SyncSinglePublisher 与 LoggingSubscriber 两个类。给出本机 mvn clean compile 的真实结果和 dependency:tree 原始输出，并解释一个容易踩的坑：这两个类实现的是 java.util.concurrent.Flow 接口，而 RS TCK 验证的是 org.reactivestreams 接口，两者形状相似但不能直接互传，需要适配。

Leeting Yan 6 分钟阅读 2937 字

先说清楚：工程现在有什么

很多章节喜欢先画一棵完整的类树，再声称「编译通过」。这一节按仓库的实际状态写，避免把计划当成既成事实。

eventd-v1 当前只有：

eventd-v1/
├── pom.xml
├── eventd-core/
│   ├── pom.xml
│   └── src/main/java/com/plume/eventd/core/
│       ├── SyncSinglePublisher.java
│       └── LoggingSubscriber.java
└── eventd-app/
    ├── pom.xml
    └── src/                      # 尚无源码

也就是说，本卷后面出现的 AsyncRangePublisher、MapProcessor、各类 Operator、SensorSimulatorPublisher、BackpressureDashboard，是各章节逐步给出的示例代码，不是仓库里已经存在的文件。阅读时把它们当作「按章节组装」的零件，而不是一个开箱即用的成品。

模块职责

模块打包职责当前内容
父 POMpom版本统一、模块聚合依赖管理
eventd-corejarRS 规范实现、操作符、订阅者2 个类
eventd-appjar演示应用、main 入口空

父 POM 用 dependencyManagement 导入两个 BOM：

<properties>
  <maven.compiler.release>25</maven.compiler.release>
  <reactor.bom.version>2025.0.8</reactor.bom.version>
  <rxjava.version>3.1.12</rxjava.version>
  <rs.tck.version>1.0.4</rs.tck.version>
  <junit.jupiter.version>5.14.4</junit.jupiter.version>
</properties>

子模块只写 groupId、artifactId，版本由父 POM 决定。eventd-app 依赖 eventd-core，因此会自动拿到 reactor-core；它自己又显式声明了一次 reactor-core，这是冗余但无害的，删掉也不影响编译。

真实依赖树

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
mvn dependency:tree -pl eventd-core

本机实测（Reactor BOM 2025.0.8）：

com.plume.eventd:eventd-core:jar:1.0.0-SNAPSHOT
+- org.reactivestreams:reactive-streams:jar:1.0.4:compile
+- io.projectreactor:reactor-core:jar:3.8.8:compile
|  \- org.jspecify:jspecify:jar:1.0.1:compile
+- io.reactivex.rxjava3:rxjava:jar:3.1.12:compile
+- org.reactivestreams:reactive-streams-tck:jar:1.0.4:test
|  +- org.testng:testng:jar:7.3.0:test
|  |  +- com.beust:jcommander:jar:1.78:test
|  |  +- org.apache.ant:ant:jar:1.10.3:test
|  |  +- junit:junit:jar:4.12:test
|  |  \- com.google.inject:guice:jar:no_aop:4.2.2:test
|  \- org.reactivestreams:reactive-streams-examples:jar:1.0.4:test
\- org.junit.jupiter:junit-jupiter:jar:5.14.4:test
   +- org.junit.jupiter:junit-jupiter-api:jar:5.14.4:test
   +- org.junit.jupiter:junit-jupiter-params:jar:5.14.4:test
   \- org.junit.jupiter:junit-jupiter-engine:jar:5.14.4:test

几个容易误判的点：

  1. BOM 的版本号不是 Reactor 的版本号。2025.0.8 是发布列车版本，解析出来的 reactor-core 是 3.8.8。写文章时不要把 BOM 版本当成组件版本。
  2. reactor-core 3.8.8 的 POM 确实声明了 reactive-streams:1.0.4（compile 作用域，非 optional）。树上没显示它，是因为项目自己直接声明了同一个坐标，Maven 把 Reactor 传来的那份按重复依赖省略了。也就是说：删掉项目里的直接依赖仍能编译，但版本会变成由 Reactor 传递决定；如果想锁定版本，保留直接声明更稳妥。
  3. TCK 1.0.4 是 TestNG 体系，会传递 testng:7.3.0、junit:4.12 和 reactive-streams-examples:1.0.4。JUnit Jupiter 5.14.4 是另外一条独立的测试栈，两者可以共存，但测试类不能混用注解。
  4. 除了这些，没有 Spring、Netty、Kafka、Flink。

一个关键选择：Flow 还是 org.reactivestreams

eventd-core 的两个现有类实现的是 JDK 自带的 java.util.concurrent.Flow：

import java.util.concurrent.Flow.Publisher;
import java.util.concurrent.Flow.Subscriber;
import java.util.concurrent.Flow.Subscription;

public final class SyncSinglePublisher<T> implements Publisher<T> {
    @Override
    public void subscribe(Subscriber<? super T> subscriber) {
        subscriber.onSubscribe(new SingleSubscription<>(subscriber, value));
    }
}

而 Reactive Streams TCK 的 PublisherVerification 接受的是 org.reactivestreams.Publisher。两套接口的方法形状几乎一样：

接口方法
Flow.Publishersubscribe(Flow.Subscriber<? super T>)
Publisher（RS）subscribe(Subscriber<? super T>)
Flow.Subscriptionrequest(long)、cancel()
Subscription（RS）request(long)、cancel()

但它们是不同的类型，没有继承关系，不能直接互换。把 Flow.Publisher 直接传给 PublisherVerification 会编译失败。

有两条官方支持的路线，不需要自己写适配器。

路线一：用 FlowAdapters 转成 RS 类型

reactive-streams:1.0.4 自带 org.reactivestreams.FlowAdapters：

import java.util.concurrent.Flow;
import org.reactivestreams.FlowAdapters;
import org.reactivestreams.Publisher;

Flow.Publisher<String> flowPublisher = new SyncSinglePublisher<>("x");
Publisher<String> rsPublisher = FlowAdapters.toPublisher(flowPublisher);

反向转换用 FlowAdapters.toFlowPublisher(...)。适配器会做一层对象包装并转发 request/cancel；如果原实现依赖对象身份做判断，适配后身份就变了。

路线二：直接使用 Flow TCK

org.reactivestreams:reactive-streams-tck-flow:1.0.4 提供了 FlowPublisherVerification、FlowSubscriberWhiteboxVerification、FlowSubscriberBlackboxVerification、IdentityFlowProcessorVerification。测试类直接继承，不需要转换：

import java.util.concurrent.Flow;
import org.reactivestreams.tck.TestEnvironment;
import org.reactivestreams.tck.flow.FlowPublisherVerification;

public class SyncSinglePublisherFlowTckTest
        extends FlowPublisherVerification<String> {

    public SyncSinglePublisherFlowTckTest() {
        super(new TestEnvironment(5_000));
    }

    @Override
    public Flow.Publisher<String> createFlowPublisher(long elements) {
        return new SyncSinglePublisher<>("x");
    }

    @Override
    public Flow.Publisher<String> createFailedFlowPublisher() {
        return null;   // 返回 null 表示不测试失败路径
    }
}

这条路线更直接，但多一个测试依赖；路线一少一个依赖，多一层包装。两者都比「自己手写适配器」可靠。

仓库目前选择了 Flow，并在 eventd-core/pom.xml 里保留了 reactive-streams 依赖，注释写的是「JDK 25 已有 Flow，但 RS TCK 需要这个」。跑 TCK 时，用上面的任一条路线即可。

现有两个类

SyncSinglePublisher

只持有一个值，request 一次就发完：

public void request(long n) {
    if (cancelled || completed) return;
    if (n <= 0) {
        cancelled = true;
        subscriber.onError(new IllegalArgumentException(
            "§3.9: non-positive request not allowed: " + n));
        return;
    }
    completed = true;
    if (value != null) subscriber.onNext(value);
    if (!cancelled) subscriber.onComplete();
}

行为要点：

  • 订阅时同步调用 onSubscribe，不启动线程。
  • request(n > 0) 最多发一个元素，随后 onComplete。
  • 第二次 request 是空操作，因为 completed 已置位。
  • cancel() 后再 request 不发任何信号。
  • request(0) 走 onError(IllegalArgumentException)。

它有两个明确边界：

  1. 不是线程安全的。cancelled、completed 都是普通 boolean。多线程并发 request 可能两次都通过检查，导致重复发射。示例假设单线程调用；真正的并发实现要么加锁，要么用原子变量和 wip 模式（13.2 会给出）。
  2. value == null 时静默完成。构造器没有拒绝 null，request 里用 if (value != null) 跳过元素。RS 规范禁止 onNext(null)，这里用「不发」规避，但调用方可能期待一个值。更严谨的写法是在构造器抛 NullPointerException。

LoggingSubscriber

一个调试用的 Subscriber：onSubscribe 里保存 Subscription，立刻 request(Long.MAX_VALUE)，所有信号打印线程名和名字。它适合观察信号顺序，不适合生产：无界请求会关掉背压，高频流会把 stdout 打满。

真实编译结果

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
cd .local/java_reactive/eventd-v1
mvn -q -DskipTests clean compile

本机实测：

mvn -q -DskipTests clean compile  3.25s user 0.26s system 116% cpu 3.004 total
exit=0

生成的 class 文件：

eventd-core/target/classes/com/plume/eventd/core/LoggingSubscriber.class
eventd-core/target/classes/com/plume/eventd/core/SyncSinglePublisher.class
eventd-core/target/classes/com/plume/eventd/core/SyncSinglePublisher$SingleSubscription.class

SingleSubscription 是私有静态内部类，所以单独有一个 $ class 文件。eventd-app 没有源码，编译阶段不会产生 class——mvn compile 成功不代表 app 模块有可运行的入口。

后续章节与模块的对应关系

内容出现位置是否已在仓库
SyncSinglePublisher、LoggingSubscriber第 2 章是
AsyncRangePublisher2.2否
MapProcessor3.1否
MapOperator、FilterOperator、TakeOperator3.2否
SubscribeOnOperator、PublishOnOperator3.3否
SensorSimulatorPublisher13.2否
BackpressureDashboard13.3否
Reactor 重写版本14.1否

这种「边写边加」的组织方式有一个好处：每个类都能在对应章节里单独编译、单独运行，不需要先凑齐整个工程。代价是仓库快照永远不完整，读者不能只 git clone 就得到书中所有示例。

演进路线（计划，不是现状）

eventd v1（卷一）
├── RS 规范自实现
├── Project Reactor 重写
└── RS TCK 验证

eventd v2（卷二）
├── Reactor Netty HTTP 网关
├── WebFlux 路由与处理
└── R2DBC 数据管道

eventd v3（卷三）
├── Kafka Streams 拓扑
├── Flink DataStream 窗口
└── 实时指标面板

本机验证口径

Apple M1 Pro，10 核，32 GB，macOS 26.3 (25D125)
Temurin 25.0.5+8
Apache Maven 3.9.12
Project Reactor BOM 2025.0.8（reactor-core 3.8.8）
RxJava 3.1.12
Reactive Streams TCK 1.0.4（TestNG 7.3.0）
JUnit Jupiter 5.14.4

小结

问题结论
仓库里有几个类2 个：SyncSinglePublisher、LoggingSubscriber
mvn clean compile 能过吗能，约 3 秒，exit 0
Reactor 实际版本3.8.8（由 BOM 2025.0.8 解析）
用的是哪套接口java.util.concurrent.Flow
能直接跑 RS TCK 吗不能直接，需 FlowAdapters 或 reactive-streams-tck-flow
编译通过等于能运行吗不等于，eventd-app 还没有源码

阅读导航

上一篇：12.3 装配时 vs 订阅时

下一篇：13.2 传感器模拟器：带背压的随机数据流

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南