Smallrye Mutiny中如何为Multi添加超时控制逻辑

两类核心需求的落地方式 #

1. 连续时段无条目发射的空闲超时 #

Multi 未提供和 Uni 完全一致的ifNoItem().after()链式调用入口，但框架从早期版本就内置了相邻发射间隔检测的原生方法timeout(Duration)，语义完全匹配“连续一段时间无条目即触发逻辑”的要求：

• 方法行为：订阅流后，持续记录上一次收到上游条目的时间，只要任意两个相邻条目的发射间隔超过指定时长，就自动取消上游订阅，向下游抛出TimeoutException
• 灵活自定义触发动作：捕获框架默认抛出的TimeoutException后，可以替换为任意业务逻辑，不需要强制中断流。

示例代码：

// 配置5分钟空闲超时规则
multi
    .timeout(Duration.ofMinutes(5))
    // 捕获超时事件，自定义处理动作
    .onFailure(TimeoutException.class)
    .recoverWithMulti(() -> {
        // 自定义逻辑参考：上报监控指标、抛出业务自定义异常、发射超时提示事件等
        Metrics.counter("multi.idle.timeout").increment();
        return Multi.createFrom().failure(new BusinessException("5分钟未收到业务数据"));
    })
// 可传入自定义调度器替换框架默认线程池
// .timeout(Duration.ofMinutes(5), customScheduledExecutor)

常用的自定义超时动作参考：

• 收到超时后直接正常完成流，不中断业务链路：调用recoverWithCompletion()
• 收到超时后发射一个内置的超时标记条目，后续再正常结束流
• 触发告警、上报链路追踪埋点

2. 订阅后指定时长未完成的总流程超时 #

这个场景以订阅时刻为起点计算总耗时，和中间条目的发射间隔无关，只要超过指定时长还没收到流的完成信号就触发对应逻辑。由于框架内置的timeout()仅检测相邻条目间隔，该场景需要通过组合流的方式实现，核心思路是把原业务流和一个固定延时的触发流做竞速，谁先向下游传递信号就按谁的逻辑处理。

示例代码：

// 广播原Multi，支持多订阅者独立消费流
Multi<T> sharedMulti = multi.broadcast().toAllSubscribers();

// 构建30分钟固定延时的超时触发流
Multi<T> timeoutTrigger = Uni.createFrom()
    // 构建一个不主动发射条目的Uni，靠超时规则触发信号
    .<T>nothing()
    .ifNoItem().after(Duration.ofMinutes(30))
    .failWith(new BusinessException("Multi总流程超过30分钟未完成"))
    .toMulti();

// 合并原业务流和超时触发流
Multi<T> resultMulti = Multi.createBy()
    .merging()
    .streams(sharedMulti, timeoutTrigger)
    // 下游取消订阅时，自动取消两个上游的资源占用，避免泄漏
    .onCancellation().invoke(() -> log.debug("Multi流程主动终止"));

实现优势：

• 完全复用框架已有的成熟超时实现，不需要手动管理定时任务的创建、取消逻辑
• 超时触发时会自动取消原业务流的上游订阅，不会造成资源泄漏
• 超时触发后的自定义逻辑和空闲超时完全一致，支持替换异常、正常完成、上报指标等动作

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