虚拟线程

• 使用虚拟线程并发执行数学与安全判定
  • 背景
  • 设计目标
  • 关键点说明
  • 代码实现
  • 返回结果语义
  • 性能收益预期
  • 注意事项与可选优化
• 其他代码
  • TioVirtualThreadUtils
  • MvThreadUtils

使用虚拟线程并发执行数学与安全判定

本文介绍如何在 ApiMathController#determine 接口中，使用 MvThreadUtils（基于虚拟线程）并发执行两个独立的判定逻辑：determineTextMath 与 determineTextSafe。目标是在保持原有“判定顺序与返回语义不变”的前提下，减少总体等待时间。

背景

/api/v1/math/determine 接口需要对用户输入的 question 做两类校验：

1. 是否为数学问题：determineTextMath(question)
2. 是否为安全问题：determineTextSafe(question)

原始实现为串行调用：先执行 math 判定，再执行 safe 判定。若两个判定内部均包含网络调用、模型推理、IO 等耗时操作，则串行执行会把耗时累加，导致接口整体响应变慢。

由于这两个判定逻辑相互独立，可以并发执行，从而将总耗时从“二者之和”降低到“二者的最大值”（忽略少量线程调度开销）。

设计目标

• 并发执行两次判定（两条虚拟线程同时开始）

• 仍然保持原本的业务判定顺序：

  • 先判断是否为数学问题（失败优先返回 10001）
  • 再判断是否安全（失败返回 10002）

• 保持返回结构不变：最终 result 字段沿用 safe 判定结果

• 正确处理中断与执行异常，避免吞掉异常或破坏线程中断语义

关键点说明

1. 为什么并发但仍“串行消费结果”

做法是：

• 提交两个任务立即开始执行（并发开始）

• 按照原逻辑顺序调用 Future#get()：

  • 先取 mathFuture.get()，保持 math 失败时优先返回
  • 再取 safeFuture.get()，继续安全校验

这样既能利用并发缩短总耗时，又不改变原来的业务优先级与错误码语义。

2. 虚拟线程的意义

虚拟线程适合大量阻塞型任务（IO、远程调用、等待模型响应等），可以显著降低平台线程占用压力，让并发更轻量、更经济。

3. 异常处理原则

• InterruptedException：

  • 必须恢复中断标记：Thread.currentThread().interrupt();
  • 返回一个可识别的错误响应

• ExecutionException：

  • 代表任务内部抛出了异常
  • 记录日志并返回内部错误响应

代码实现

下面是修改后的完整控制器实现（使用虚拟线程并发执行两个判定任务）：

package nexus.io.manim.controller;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import com.jfinal.kit.Kv;
import nexus.io.annotation.Get;
import nexus.io.annotation.RequestPath;
import nexus.io.jfinal.aop.Aop;
import nexus.io.manim.services.MathQuesitonService;
import nexus.io.manim.utils.MvThreadUtils;
import nexus.io.model.body.RespBodyVo;

import lombok.extern.slf4j.Slf4j;

@RequestPath("/api/v1/math")
@Slf4j
public class ApiMathController {
  private MathQuesitonService mathQuesitonService = Aop.get(MathQuesitonService.class);

  @Get
  public RespBodyVo determine(String question) {
    if (question == null) {
      return RespBodyVo.fail("question can not be null");
    }

    // 两个虚拟线程同时执行
    Future<Boolean> mathFuture = MvThreadUtils.submit(() -> mathQuesitonService.determineTextMath(question));
    Future<Boolean> safeFuture = MvThreadUtils.submit(() -> mathQuesitonService.determineTextSafe(question));

    try {
      // 按原来的串行判定顺序消费结果：先 math，后 safe
      Boolean isMath = mathFuture.get();
      if (isMath != null && !isMath) {
        return RespBodyVo.fail(10001, "only support math question");
      }

      Boolean isSafe = safeFuture.get();
      if (isSafe != null && !isSafe) {
        return RespBodyVo.fail(10002, "only support safe question");
      }

      // 保持与原实现一致：最终返回 safe 的判定结果
      Kv kv = Kv.by("result", isSafe);
      return RespBodyVo.ok(kv);

    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      log.error("determine interrupted, question={}", question, e);
      return RespBodyVo.fail(10003, "request interrupted");
    } catch (ExecutionException e) {
      log.error("determine execute failed, question={}", question, e);
      return RespBodyVo.fail(10004, "internal error");
    }
  }
}

返回结果语义

• question == null

  • 返回失败：question can not be null

• determineTextMath(question) 返回 false（且不为 null）

  • 返回失败：10001 only support math question

• determineTextSafe(question) 返回 false（且不为 null）

  • 返回失败：10002 only support safe question

• 两者均通过（或返回 null）

  • 返回成功：{"result": isSafe}（保持与原逻辑一致）

性能收益预期

在两次判定耗时分别为 T_math、T_safe 的情况下：

• 串行：T_total ≈ T_math + T_safe
• 并发：T_total ≈ max(T_math, T_safe)（外加少量调度开销）

当两个判定都存在明显 IO 或等待时间时，并发通常能带来直观收益。

注意事项与可选优化

1. 提前返回时的取消

   • 当前实现如果 math 判定失败会直接返回，但 safe 任务可能仍在运行。
   • 若希望进一步节省资源，可以在 math 失败时尝试 safeFuture.cancel(true)（是否能中断取决于任务内部是否响应中断）。

2. 超时控制

   • 如果担心外部调用卡死，可使用带超时的 get(timeout, unit)，并返回特定错误码。

3. 日志与监控

   • 可增加耗时统计，便于验证并发带来的收益与定位慢点。

---

如果希望我再补一版“带取消与超时”的增强实现（同时保持原错误码优先级不变），我也可以直接给出完整代码。

其他代码

TioVirtualThreadUtils

package nexus.io.manim.utils;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Virtual thread utils (JDK 21+).
 * 
 * @author Tong Li
 */
public class TioVirtualThreadUtils {

  private static volatile ExecutorService virtualThreadExecutor;

  static {
    start();
  }

  private static ThreadFactory namedVirtualThreadFactory(String baseName) {
    AtomicInteger threadNumber = new AtomicInteger(1);
    return r -> Thread.ofVirtual().name(baseName + "-" + threadNumber.getAndIncrement()).unstarted(r);
  }

  public static ExecutorService getExecutor() {
    return virtualThreadExecutor;
  }

  public static <T> Future<T> submit(Callable<T> task) {
    return virtualThreadExecutor.submit(task);
  }

  public static <T> Future<T> submit(Runnable task, T result) {
    return virtualThreadExecutor.submit(task, result);
  }

  public static Future<?> submit(Runnable task) {
    return virtualThreadExecutor.submit(task);
  }

  public static void execute(Runnable runnable) {
    virtualThreadExecutor.execute(runnable);
  }

  /**
   * Start executor if not started. Uses a per-task virtual thread executor,
   * suitable for IO-bound workloads.
   */
  public static void start() {
    if (virtualThreadExecutor == null) {
      // Two equivalent options:
      // 1) with custom name factory:
      virtualThreadExecutor = Executors.newThreadPerTaskExecutor(namedVirtualThreadFactory("tio-vt"));

      // 2) simplest:
      // virtualThreadExecutor = Executors.newVirtualThreadPerTaskExecutor();
    }
  }

  public static void restart() {
    stop();
    start();
  }

  /**
   * Stop executor immediately. Note: shutdownNow() will attempt to interrupt
   * running tasks.
   */
  public static void stop() {
    if (virtualThreadExecutor != null) {
      virtualThreadExecutor.shutdownNow();
      virtualThreadExecutor = null;
    }
  }
}

MvThreadUtils

package nexus.io.manim.utils;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;

public class MvThreadUtils {

  public static void execute(Runnable runable) {
    TioVirtualThreadUtils.execute(runable);
  }
  
  public static <T> Future<T> submit(Callable<T> task) {
    return TioVirtualThreadUtils.submit(task);
  }

  public static <T> Future<T> submit(Runnable task, T result) {
    return TioVirtualThreadUtils.submit(task, result);
  }

  public static Future<?> submit(Runnable task) {
    return TioVirtualThreadUtils.submit(task);
  }
}