Spring Boot 3.0.2中Reactive Streams热发布者与GraphQL订阅失效排查

结合你描述的现象（冷发布者正常、实体变更后订阅直接结束、FluxSink标记为已取消），核心问题大概率出在自定义事件流的创建/管理逻辑上，以下是具体排查点和解决方案：

1. 检查是否误触发了FluxSink的终止操作 #

这是最常见的原因：如果你的EntityChangeNotifier在发布变更事件时，不小心调用了sink.complete()或sink.error()，会直接终止整个流，导致订阅断开。

错误示例（会导致流结束）：

public void notifyChange(MyEntity entity) {
    sink.next(entity);
    sink.complete(); // 这里会直接关闭流，订阅随即结束
}

正确做法：只调用next()发布事件，不要主动调用complete()或error()，除非你确实需要终止整个订阅流。

2. 确认使用热流而非冷流实现全局通知 #

你的需求是全局实体变更通知，需要用热流让多个订阅者共享同一事件流。如果每次订阅都创建新的Flux.create实例（冷流），mutation触发的事件会发送到错误的流实例，甚至当前订阅的Sink已被回收。

推荐使用Spring Reactor的Sinks.Many实现热流，替代直接使用Flux.create：

正确的EntityChangeNotifier实现： #

@Component
public class EntityChangeNotifier {
    // 创建支持多订阅者的热流，背压采用缓冲策略
    private final Sinks.Many<MyEntity> entityChangeSink = Sinks.many().multicast().onBackpressureBuffer();

    // 发布变更事件
    public void notifyChange(MyEntity entity) {
        // 安全发布事件，避免背压或订阅者已取消的异常
        entityChangeSink.emitNext(entity, Sinks.EmitFailureHandler.FAIL_FAST);
    }

    // 对外暴露订阅流
    public Flux<MyEntity> getEntityChanges() {
        return entityChangeSink.asFlux();
    }
}

GraphQL订阅映射类： #

@Controller
public class MyEntityGraphQLController {
    private final EntityChangeNotifier changeNotifier;

    public MyEntityGraphQLController(EntityChangeNotifier changeNotifier) {
        this.changeNotifier = changeNotifier;
    }

    @SubscriptionMapping
    public Flux<MyEntity> changed() {
        // 返回共享的热流，所有订阅者都会收到同一份事件
        return changeNotifier.getEntityChanges();
    }

    @MutationMapping
    public MyEntity create(@Argument MyEntity input) {
        // 业务逻辑：保存实体
        MyEntity savedEntity = myEntityService.save(input);
        // 触发变更通知
        changeNotifier.notifyChange(savedEntity);
        return savedEntity;
    }
}

3. 排查GraphQL Schema定义是否正确 #

确认订阅字段的SDL定义符合流式要求，不要错误定义为列表类型：

正确的Schema示例：

type Subscription {
    changed: MyEntity! # 单个实体类型，而非[MyEntity]，订阅会逐个推送事件
}

type MyEntity {
    id: ID!
    name: String!
    # 其他字段
}

type Mutation {
    create(input: MyEntityInput!): MyEntity!
}

input MyEntityInput {
    name: String!
}

4. 查看日志定位取消原因 #

开启Spring GraphQL的DEBUG级别日志，追踪订阅的生命周期：

• 在application.yml中添加：

logging:
  level:
    org.springframework.graphql: DEBUG

通过日志可以看到：

• 订阅何时建立
• FluxSink被取消的具体时机和原因（比如是否是客户端主动取消，还是服务端触发终止）

5. 检查线程安全问题 #

如果多个线程同时操作FluxSink，可能会导致Sink状态异常。使用Sinks.Many可以避免大部分线程安全问题，因为它本身是线程安全的。如果是自己维护Sink集合，一定要加同步锁。

内容的提问来源于stack exchange，提问作者Cosdix