10.2 事件通知、状态行为与集中协调

Observer、State 与 Mediator 分别解决：一个事实怎样通知多个关注者、对象行为怎样随生命周期变化，以及一组同级对象怎样避免互相形成网状依赖。

Observer：发布事实，不遥控订阅者

Observer 建立一对多通知。发布者只知道观察者契约，不知道每个观察者为何响应。下面使用 Java 21，项目需提供 MatchId、PlayerId；每个 public 类型单独存放。

public record MatchFinished(
        MatchId matchId,
        PlayerId winnerId,
        java.time.Instant finishedAt) {}

@FunctionalInterface
public interface MatchFinishedListener {
    void on(MatchFinished event);
}

public final class LocalMatchEvents {
    private final java.util.List<MatchFinishedListener> listeners;

    public LocalMatchEvents(java.util.List<MatchFinishedListener> listeners) {
        this.listeners = java.util.List.copyOf(listeners);
    }

    public void publish(MatchFinished event) {
        java.util.Objects.requireNonNull(event, "event");
        for (MatchFinishedListener listener : listeners) {
            listener.on(event);
        }
    }
}

事件名称使用已经发生的事实 MatchFinished，而不是命令式 UpdateLeaderboard。后者把发布者重新耦合到订阅者职责。

同步 Observer 的失败语义

上面的实现同步、按注册顺序在发布线程调用；第一个异常会传播并中止后续通知，先前 listener 的副作用不会回滚。订阅列表是固定快照，没有动态增删；递归发布目前也没有保护。比如 A 记日志、B 抛异常、C 发邮件，轨迹只有 A、B。若需要别的行为，必须明确修改并测试：

• 一个 listener 失败，后续 listener 是否继续；
• listener 是否参与发布者事务；
• 是否允许递归发布；
• listener 能否修改订阅列表；
• 处理时限和观测方式是什么。

若“保存比赛”和“更新积分”必须原子完成，不应仅靠无事务保证的 Observer 假装解耦。若允许最终一致，可以在事务中写 Outbox，再由消息系统异步投递；那已经进入消息可靠性问题，需要幂等消费者、重试、顺序和死信策略。

进程内 Observer 与消息代理不是一回事：前者通常共享进程和故障，后者跨进程并引入投递语义。

Java Flow 还规定 request(n) 需求量和取消协议；上面的简单 Observer 没有背压能力，换个接口名也不会自动得到它。

订阅生命周期

动态注册时，Subject 持有 listener 强引用可能延长其生命周期。比“全部换成弱引用”更可靠的做法，是让订阅返回显式句柄：

interface Subscription extends AutoCloseable {
    @Override void close();
}

这个接口只是动态订阅 API 的契约草图，前面的固定列表发布者尚未实现注册/注销。实现应规定重复 close() 是否幂等，以及并发注销时已在执行的回调是否仍会完成。拥有者在生命周期结束时关闭订阅。弱引用可能让 listener 在无人持有强引用时悄悄消失，也不替代明确所有权。

State：把状态特有行为与转换放在一起

当一个对象在不同状态下允许的操作和结果差异很大，散落的 if (status == ...) 会让转换规则难以维护。下面给出四个完整状态：截止时刻不包含在可确认区间内；按可信支付完成时间判断，取消后的到账一律待退款。PaymentReceipt 是本例的最小值类型，调用层必须先核验支付身份、金额、币种和报名归属。

public record PaymentReceipt(String paymentId, java.time.Instant paidAt) {
    public PaymentReceipt {
        if (paymentId == null || paymentId.isBlank()) {
            throw new IllegalArgumentException("blank payment id");
        }
        java.util.Objects.requireNonNull(paidAt, "paidAt");
    }
}

public sealed interface RegistrationState permits PendingRegistration,
        ConfirmedRegistration, CancelledRegistration, RefundPendingRegistration {
    RegistrationState paymentConfirmed(PaymentReceipt receipt);
    RegistrationState cancel(java.time.Instant now);
}

public record PendingRegistration(java.time.Instant deadline)
        implements RegistrationState {
    public PendingRegistration {
        java.util.Objects.requireNonNull(deadline, "deadline");
    }
    @Override
    public RegistrationState paymentConfirmed(PaymentReceipt receipt) {
        java.util.Objects.requireNonNull(receipt, "receipt");
        return receipt.paidAt().isBefore(deadline)
                ? new ConfirmedRegistration(receipt)
                : new RefundPendingRegistration(receipt);
    }
    @Override
    public RegistrationState cancel(java.time.Instant now) {
        return new CancelledRegistration(now);
    }
}

public record ConfirmedRegistration(PaymentReceipt receipt)
        implements RegistrationState {
    public ConfirmedRegistration {
        java.util.Objects.requireNonNull(receipt, "receipt");
    }
    @Override
    public RegistrationState paymentConfirmed(PaymentReceipt repeated) {
        if (!receipt.equals(repeated)) {
            throw new IllegalArgumentException("conflicting payment");
        }
        return this;
    }
    @Override
    public RegistrationState cancel(java.time.Instant now) {
        java.util.Objects.requireNonNull(now, "now");
        return new RefundPendingRegistration(receipt);
    }
}

public record CancelledRegistration(java.time.Instant cancelledAt)
        implements RegistrationState {
    public CancelledRegistration {
        java.util.Objects.requireNonNull(cancelledAt, "cancelledAt");
    }
    @Override
    public RegistrationState paymentConfirmed(PaymentReceipt receipt) {
        return new RefundPendingRegistration(receipt);
    }
    @Override
    public RegistrationState cancel(java.time.Instant now) {
        java.util.Objects.requireNonNull(now, "now");
        return this;
    }
}

public record RefundPendingRegistration(PaymentReceipt receipt)
        implements RegistrationState {
    public RefundPendingRegistration {
        java.util.Objects.requireNonNull(receipt, "receipt");
    }
    @Override
    public RegistrationState paymentConfirmed(PaymentReceipt repeated) {
        if (!receipt.equals(repeated)) {
            throw new IllegalArgumentException("conflicting payment");
        }
        return this;
    }
    @Override
    public RegistrationState cancel(java.time.Instant now) {
        java.util.Objects.requireNonNull(now, "now");
        return this;
    }
}

每个状态方法返回下一状态，不修改旧对象；上下文必须接住返回值并持久化。相同回执重复到达才返回原状态，不同付款 ID 或时间会报告冲突，不能把第二笔扣款藏进“幂等”两个字里。取消后收到付款返回待退款；待退款只是记录意图，尚未调用退款网关，也未展示退款完成状态。

转换必须与并发一致

两个进程同时处理“支付成功”和“取消”时，内存中的 State 模式不能防止丢失更新。数据库层仍需版本号/OCC、条件更新或锁：

UPDATE registration
   SET state = :next_state, version = version + 1
 WHERE id = :id AND version = :expected_version;

这些 :name 是访问层绑定参数，不是直接可在 psql 中执行的占位符。受影响行数为 0 只表示没有更新到行：可能版本已变，也可能记录不存在或被过滤；不能直接认定命令已经成功。重新读取后区分不存在、冲突和重复请求，按最新状态重新计算转换。状态与待发布事件若要求一致，应在同一事务落盘。

状态少、转换封闭时，枚举加一个集中 switch 可能比类层次更清楚。State 适合状态特有行为较多、分支在多处重复的情况。

Mediator：集中一组对象的协作协议

一个比赛房间中，倒计时、参赛者连接、裁判面板和直播状态若互相直接调用，会形成网状依赖。Mediator 让它们只与协调者通信：

public final class MatchRoomCoordinator {
    private final PlayerConnections players;
    private final MatchClock clock;
    private final BroadcastPort broadcast;
    private boolean started;

    public MatchRoomCoordinator(PlayerConnections players, MatchClock clock,
                                BroadcastPort broadcast) {
        this.players = java.util.Objects.requireNonNull(players);
        this.clock = java.util.Objects.requireNonNull(clock);
        this.broadcast = java.util.Objects.requireNonNull(broadcast);
    }

    public void playerReady(PlayerId playerId) {
        if (started) return;
        players.markReady(playerId);
        if (players.allReady()) {
            started = true;
            clock.start();
            broadcast.publish(MatchRoomEvent.started());
        }
    }
}

这个项目片段限定在单线程房间事件循环中，players 负责核验玩家属于当前房间。started 在外部调用前置位，阻止重复或同步重入的 ready 通知再次启动。它不解决线程间竞争；若启动时钟或广播失败，状态仍是已启动，生产实现需要可恢复的状态与可靠事件发布，不能简单重置标志后赌外部操作没发生。

协调协议集中后，参与者变简单，也更容易测试事件顺序。代价是 Mediator 可能成长为新的 God Object。应按用例或协作上下文拆分，不要把全系统所有通信塞进一个 SystemMediator。

Mediator 与 Facade 的方向不同：Facade 为外部调用者提供简单入口；Mediator 协调内部同级对象，使它们不直接互相依赖。一个应用服务可能同时承担两种角色，但要清楚它维护的是哪套协议。

三种模式的选择问题

问题	模式
一个事实需要通知未知数量关注者	Observer
同一操作随对象状态改变行为	State
多个同级对象之间协作形成网状依赖	Mediator

参考资料

• Java SE 21：Flow Reactive Streams Interfaces

• Java SE 21：AutoCloseable

• PostgreSQL：Explicit Locking

• PostgreSQL: UPDATE result count