如何计算Mono.fromRunnable订阅到Runnable实际执行的时间差

有两种常用的实现方式，可根据你的业务场景选择：

方案1：单独统计指定Mono的等待时间 #

利用Reactor的doOnSubscribe钩子记录订阅触发时间，在Runnable执行开头计算时间差即可，示例代码如下：
首先封装通用工具方法：

import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import reactor.core.publisher.Mono;

public class MonoTimedUtils {
    /**
     * 创建带排队时间统计的Mono<Runnable>
     * @param businessRunnable 原有业务Runnable
     * @param waitTimeConsumer 排队时间回调，入参为等待毫秒数
     * @return 包装后的Mono实例
     */
    public static Mono<Void> createTimedRunnableMono(Runnable businessRunnable, Consumer<Long> waitTimeConsumer) {
        AtomicLong subscribeTimeHolder = new AtomicLong();
        return Mono.fromRunnable(() -> {
                    // 任务拿到线程开始执行时计算等待时长
                    long waitMs = System.currentTimeMillis() - subscribeTimeHolder.get();
                    waitTimeConsumer.accept(waitMs);
                    // 执行业务逻辑
                    businessRunnable.run();
                })
                // 订阅触发时立刻记录时间
                .doOnSubscribe(subscription -> subscribeTimeHolder.set(System.currentTimeMillis()));
    }
}

替换原有代码即可使用：

// 原有逻辑直接替换为工具方法生成Mono
MonoTimedUtils.createTimedRunnableMono(new Runnable() {
    //Some business logic
}, waitMs -> {
    // 可在此处做日志打印、指标上报等统计逻辑
    System.out.printf("任务排队等待时间：%dms%n", waitMs);
})
.subscribeOn(scheduler)
.subscribe();

方案2：全局统计该线程池所有任务的等待时间 #

如果不需要单独统计特定Mono，想零侵入统计所有提交到该ExecutorService的任务等待时间，可以直接自定义线程池包装类，在任务提交和执行时埋点：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

public class TimedThreadPoolExecutor extends ThreadPoolExecutor {
    private final Consumer<Long> waitTimeConsumer;

    public TimedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Consumer<Long> waitTimeConsumer) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.waitTimeConsumer = waitTimeConsumer;
    }

    @Override
    public void execute(Runnable command) {
        // 任务提交时记录时间
        long submitTime = System.currentTimeMillis();
        super.execute(() -> {
            // 任务执行前计算等待时长
            long waitMs = System.currentTimeMillis() - submitTime;
            waitTimeConsumer.accept(waitMs);
            command.run();
        });
    }
}

后续用这个自定义线程池构造Scheduler即可，原有业务代码无需任何修改，就能自动统计所有任务的排队等待时间。

注意事项 #

• 如果你的Scheduler使用了带延迟的调度能力，统计出的时间差会包含设置的延迟时长，符合从订阅触发到实际执行的时间差统计需求。
• 若需要更高精度的统计，可以将System.currentTimeMillis()替换为System.nanoTime()，最终结果转换为毫秒即可。

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