task-75: 虚拟线程任务等待队列上限与拒绝/延迟指标

2026-08-30 20:39:31 +08:00
parent ba163aa4bd
commit 6a4a115260
4 changed files with 340 additions and 3 deletions
@@ -0,0 +1,114 @@
+package com.nanri.aiimage.config;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.core.task.TaskRejectedException;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Task 75：虚拟线程任务排队闸门。Coze 执行池的信号量只限制"正在执行"的
+ * 并发度，提交侧仍会在虚拟线程里无限排队。此闸门在提交时统计"已受理未启动"
+ * 的等待数，达到上限立即拒绝并记录指标，防止等待队列无界堆积：
+ * <ul>
+ *   <li>等待数：提交时 +1，任务执行（或执行器拒绝）时 -1；上限钳制到 [1, +∞)；</li>
+ *   <li>指标：等待耗时、执行耗时、拒绝次数（queue-full / delegate-rejected / invalid-input）；</li>
+ *   <li>执行器不可用（provider 为空）时拒绝新提交，不产生死等任务。</li>
+ * </ul>
+ */
+@Slf4j
+public class CozeTaskQueueGate implements TaskExecutor {
+
+    private final TaskExecutor delegate;
+    private final int maxWaiting;
+    private final ObjectProvider<MeterRegistry> meterRegistryProvider;
+    private final AtomicInteger waiting = new AtomicInteger();
+
+    public CozeTaskQueueGate(TaskExecutor delegate, int maxWaiting,
+                             ObjectProvider<MeterRegistry> meterRegistryProvider) {
+        this.delegate = delegate;
+        this.maxWaiting = Math.max(1, maxWaiting);
+        this.meterRegistryProvider = meterRegistryProvider;
+    }
+
+    public int waiting() {
+        return waiting.get();
+    }
+
+    @Override
+    public void execute(Runnable command) {
+        if (command == null) {
+            recordRejected("invalid-input");
+            throw new IllegalArgumentException("coze 任务不能为 null");
+        }
+        if (waiting.get() >= maxWaiting) {
+            recordRejected("queue-full");
+            log.warn("[coze-task][gate] waiting queue full, reject submit waiting={} limit={}",
+                    waiting.get(), maxWaiting);
+            throw new TaskRejectedException("coze 等待队列已满，limit=" + maxWaiting
+                    + ", waiting=" + waiting.get());
+        }
+        waiting.incrementAndGet();
+        long queuedAt = System.nanoTime();
+        try {
+            delegate.execute(() -> {
+                long waitNanos = System.nanoTime() - queuedAt;
+                try {
+                    run(command);
+                } finally {
+                    waiting.decrementAndGet();
+                    recordQueueWait(waitNanos);
+                }
+            });
+        } catch (RuntimeException ex) {
+            waiting.decrementAndGet();
+            recordQueueWait(System.nanoTime() - queuedAt);
+            recordRejected("delegate-rejected");
+            log.warn("[coze-task][gate] delegate rejected submit waiting={} limit={} msg={}",
+                    waiting.get(), maxWaiting, ex.getMessage(), ex);
+            throw ex;
+        }
+    }
+
+    private void run(Runnable command) {
+        long startedAt = System.nanoTime();
+        try {
+            command.run();
+        } finally {
+            recordExecution(System.nanoTime() - startedAt);
+        }
+    }
+
+    private void recordRejected(String reason) {
+        MeterRegistry registry = meterRegistry();
+        if (registry != null) {
+            registry.counter("aiimage.coze-task.submit.rejected.total", "reason", reason).increment();
+        }
+    }
+
+    private void recordQueueWait(long waitNanos) {
+        MeterRegistry registry = meterRegistry();
+        if (registry != null && waitNanos >= 0L) {
+            Timer.builder("aiimage.coze-task.queue.wait.duration")
+                    .register(registry)
+                    .record(waitNanos, TimeUnit.NANOSECONDS);
+        }
+    }
+
+    private void recordExecution(long durationNanos) {
+        MeterRegistry registry = meterRegistry();
+        if (registry != null && durationNanos >= 0L) {
+            Timer.builder("aiimage.coze-task.execution.duration")
+                    .register(registry)
+                    .record(durationNanos, TimeUnit.NANOSECONDS);
+        }
+    }
+
+    private MeterRegistry meterRegistry() {
+        return meterRegistryProvider == null ? null : meterRegistryProvider.getIfAvailable();
+    }
+}
@@ -1,5 +1,7 @@
 package com.nanri.aiimage.config;

+import io.micrometer.core.instrument.MeterRegistry;
+import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -40,9 +42,11 @@ public class TaskFileJobConfig {
    @Bean("cozeTaskExecutor")
    public TaskExecutor cozeTaskExecutor(
            ExecutorService cozeVirtualThreadExecutor,
-            @Value("${aiimage.coze-task.max-concurrent:12}") int maxConcurrent) {
+            @Value("${aiimage.coze-task.max-concurrent:12}") int maxConcurrent,
+            @Value("${aiimage.coze-task.max-waiting:1000}") int maxWaiting,
+            ObjectProvider<MeterRegistry> meterRegistryProvider) {
        Semaphore semaphore = new Semaphore(Math.max(1, maxConcurrent));
-        return new ConcurrentTaskExecutor(command -> {
+        TaskExecutor semaphoreLimited = new ConcurrentTaskExecutor(command -> {
            if (command == null) {
                throw new IllegalArgumentException("coze 任务不能为 null");
            }
@@ -61,5 +65,6 @@ public class TaskFileJobConfig {
                }
            });
        });
+        return new CozeTaskQueueGate(semaphoreLimited, maxWaiting, meterRegistryProvider);
    }
 }
@@ -0,0 +1,218 @@
+package com.nanri.aiimage.config;
+
+import io.micrometer.core.instrument.Counter;
+import io.micrometer.core.instrument.MeterRegistry;
+import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.core.task.TaskExecutor;
+import org.springframework.core.task.TaskRejectedException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Task 75：为虚拟线程任务增加等待队列上限与拒绝/延迟指标。
+ * CozeTaskQueueGate 在信号量限流（并发上限）之外增加排队闸门：
+ * 已提交未启动（含等待信号量）的任务数量达到 max-waiting 上限时
+ * 立即拒绝新提交并记录拒绝指标；每次提交/执行记录等待耗时与执行耗时
+ * 指标；任务执行完成、失败或执行器拒绝后排队名额必须释放。
+ */
+class CozeTaskQueueGateTest {
+
+    private SimpleMeterRegistry registry;
+    private ObjectProvider<MeterRegistry> meterRegistryProvider;
+    private List<Runnable> captured;
+    private TaskExecutor capturingDelegate;
+
+    @BeforeEach
+    @SuppressWarnings("unchecked")
+    void setUp() {
+        registry = new SimpleMeterRegistry();
+        meterRegistryProvider = mock(ObjectProvider.class);
+        when(meterRegistryProvider.getIfAvailable()).thenReturn(registry);
+        captured = new ArrayList<>();
+        capturingDelegate = captured::add;
+    }
+
+    private CozeTaskQueueGate gate(int maxWaiting) {
+        return new CozeTaskQueueGate(capturingDelegate, maxWaiting, meterRegistryProvider);
+    }
+
+    private long rejectedCount(String reason) {
+        Counter counter = registry.find("aiimage.coze-task.submit.rejected.total")
+                .tag("reason", reason).counter();
+        return counter == null ? 0 : (long) counter.count();
+    }
+
+    private long executionCount() {
+        Timer timer = registry.find("aiimage.coze-task.execution.duration").timer();
+        return timer == null ? 0 : timer.count();
+    }
+
+    private long waitCount() {
+        Timer timer = registry.find("aiimage.coze-task.queue.wait.duration").timer();
+        return timer == null ? 0 : timer.count();
+    }
+
+    @Test
+    void test_task_075_metrics_normal_default_path() {
+        // 默认路径：任务正常受理并执行，等待/执行指标各记一次，排队名额释放。
+        CozeTaskQueueGate gate = gate(4);
+        AtomicInteger ran = new AtomicInteger();
+        gate.execute(ran::incrementAndGet);
+
+        assertEquals(1, captured.size(), "任务入队");
+        assertEquals(1, gate.waiting(), "未启动任务计数为 1");
+        captured.get(0).run();
+        assertEquals(1, ran.get(), "任务已执行");
+        assertEquals(0, gate.waiting(), "执行后名额释放");
+        assertEquals(1, executionCount(), "执行耗时指标记录一次");
+        assertEquals(1, waitCount(), "等待耗时指标记录一次");
+        assertEquals(0, rejectedCount("queue-full"));
+    }
+
+    @Test
+    void test_task_075_metrics_normal_multiple_items() {
+        // 批量场景：多个任务依次受理，执行顺序稳定不丢失，指标逐条记录。
+        CozeTaskQueueGate gate = gate(8);
+        List<Integer> order = new ArrayList<>();
+        for (int i = 1; i <= 3; i++) {
+            int id = i;
+            gate.execute(() -> order.add(id));
+        }
+        assertEquals(3, captured.size());
+        for (Runnable runnable : captured) {
+            runnable.run();
+        }
+        assertEquals(List.of(1, 2, 3), order, "执行顺序与提交顺序一致");
+        assertEquals(0, gate.waiting());
+        assertEquals(3, executionCount());
+        assertEquals(3, waitCount());
+    }
+
+    @Test
+    void test_task_075_metrics_normal_repeated_operation_is_idempotent() {
+        // 幂等：同一任务对象重复提交各自独立执行，不合并、不丢失。
+        CozeTaskQueueGate gate = gate(4);
+        AtomicInteger ran = new AtomicInteger();
+        Runnable task = ran::incrementAndGet;
+        gate.execute(task);
+        gate.execute(task);
+        assertEquals(2, captured.size(), "同一任务重复提交各入队一次");
+        captured.forEach(Runnable::run);
+        assertEquals(2, ran.get());
+        assertEquals(2, executionCount());
+    }
+
+    @Test
+    void test_task_075_metrics_boundary_empty_input() {
+        // 空输入：null 任务直接拒绝，不计数、不产生指标。
+        CozeTaskQueueGate gate = gate(4);
+        IllegalArgumentException ex = assertThrows(IllegalArgumentException.class,
+                () -> gate.execute(null));
+        assertTrue(ex.getMessage().contains("不能为 null"), "可识别错误消息");
+        assertTrue(captured.isEmpty());
+        assertEquals(0, gate.waiting());
+        assertEquals(0, rejectedCount("queue-full"));
+        assertEquals(0, executionCount());
+    }
+
+    @Test
+    void test_task_075_metrics_boundary_single_item() {
+        // 单元素：单任务直接受理执行，不依赖批量路径。
+        CozeTaskQueueGate gate = gate(1);
+        AtomicInteger ran = new AtomicInteger();
+        gate.execute(ran::incrementAndGet);
+        captured.get(0).run();
+        assertEquals(1, ran.get());
+        assertEquals(0, gate.waiting());
+        assertEquals(1, executionCount());
+    }
+
+    @Test
+    void test_task_075_metrics_boundary_limit_and_overflow() {
+        // 上限/超限：等待队列达到上限后新提交被拒绝（记录拒绝指标），
+        // 排队名额释放后可再次受理，等待数不发生无界增长。
+        CozeTaskQueueGate gate = gate(1);
+        AtomicInteger ran = new AtomicInteger();
+
+        gate.execute(ran::incrementAndGet);
+        assertEquals(1, gate.waiting());
+
+        TaskRejectedException firstReject = assertThrows(TaskRejectedException.class,
+                () -> gate.execute(ran::incrementAndGet), "队列满拒绝新提交");
+        assertTrue(firstReject.getMessage().contains("limit=1"), "错误消息含上限值");
+        TaskRejectedException secondReject = assertThrows(TaskRejectedException.class,
+                () -> gate.execute(ran::incrementAndGet));
+        assertEquals(2, rejectedCount("queue-full"), "两次拒绝各记一次指标");
+        assertEquals(1, captured.size(), "被拒绝的任务不入队");
+
+        captured.get(0).run();
+        assertEquals(0, gate.waiting(), "执行后名额释放");
+        gate.execute(ran::incrementAndGet);
+        assertEquals(2, captured.size(), "超限后仍可继续受理");
+        assertEquals(1, gate.waiting(), "等待数不超过上限");
+        captured.get(1).run();
+        assertEquals(2, ran.get(), "再次受理的任务正常执行");
+        assertEquals(0, gate.waiting(), "执行后名额再次释放");
+    }
+
+    @Test
+    void test_task_075_metrics_invalid_input_rejected() {
+        // 非法参数：等待上限 0/负值统一钳制到 1（首个受理、第二个拒绝），
+        // null 任务拒绝且不产生指标。
+        CozeTaskQueueGate zeroLimit = gate(0);
+        zeroLimit.execute(() -> { });
+        assertThrows(TaskRejectedException.class, () -> zeroLimit.execute(() -> { }));
+        assertEquals(1, captured.size(), "上限 0 回退到 1");
+
+        captured.clear();
+        CozeTaskQueueGate negativeLimit = gate(-5);
+        negativeLimit.execute(() -> { });
+        assertThrows(TaskRejectedException.class, () -> negativeLimit.execute(() -> { }));
+        assertEquals(1, captured.size(), "负值回退到 1");
+
+        assertThrows(IllegalArgumentException.class, () -> negativeLimit.execute(null));
+        assertEquals(0, executionCount(), "非法参数不产生执行指标");
+    }
+
+    @Test
+    void test_task_075_metrics_dependency_failure_releases_resources() {
+        // 依赖失败：执行器拒绝时名额释放、记录拒绝指标，恢复后再次提交成功；
+        // 任务执行抛异常时名额同样释放，后续任务不受影响。
+        CozeTaskQueueGate gate = new CozeTaskQueueGate(command -> {
+            throw new TaskRejectedException("executor full");
+        }, 4, meterRegistryProvider);
+        assertThrows(TaskRejectedException.class, () -> gate.execute(() -> { }));
+        assertEquals(0, gate.waiting(), "拒绝后名额释放");
+        assertEquals(1, rejectedCount("delegate-rejected"), "执行器拒绝单独计数");
+
+        CozeTaskQueueGate recovered = gate(4);
+        AtomicInteger ran = new AtomicInteger();
+        recovered.execute(ran::incrementAndGet);
+        captured.get(0).run();
+        assertEquals(1, ran.get(), "恢复后提交成功");
+
+        captured.clear();
+        CozeTaskQueueGate failingTaskGate = gate(4);
+        failingTaskGate.execute(() -> {
+            throw new IllegalStateException("task boom");
+        });
+        assertThrows(IllegalStateException.class, () -> captured.get(0).run(),
+                "任务异常向上传播（生产环境由虚拟线程吞掉）");
+        assertEquals(0, failingTaskGate.waiting(), "任务抛异常后名额释放");
+        failingTaskGate.execute(ran::incrementAndGet);
+        captured.get(1).run();
+        assertEquals(2, ran.get(), "异常后新任务可受理执行");
+    }
+}
@@ -62,7 +62,7 @@ class ThreadPoolIsolationConfigTest {
    }

    private TaskExecutor newCoze(ExecutorService virtualExecutor, int maxConcurrent) {
-        return new TaskFileJobConfig().cozeTaskExecutor(virtualExecutor, maxConcurrent);
+        return new TaskFileJobConfig().cozeTaskExecutor(virtualExecutor, maxConcurrent, 4, null);
    }

    @Test