fix: keep shadow work off legacy threads

2026-08-12 09:19:59 +08:00
parent 34c4d54669
commit bb3b93a79f
8 changed files with 274 additions and 169 deletions
--- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/InboundShadowCompletion.java
@@ -0,0 +1,49 @@
+package com.ruoyi.algorithm.shadow;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public final class InboundShadowCompletion {
+    private static final InboundShadowCompletion REJECTED =
+            new InboundShadowCompletion(false, 0L);
+
+    private final boolean accepted;
+    private final long deadlineNanos;
+    private final CompletableFuture<LegacyResult> legacyResult = new CompletableFuture<>();
+
+    static InboundShadowCompletion accepted(int timeoutSeconds) {
+        return new InboundShadowCompletion(
+                true, System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds));
+    }
+
+    public static InboundShadowCompletion rejected() {
+        return REJECTED;
+    }
+
+    private InboundShadowCompletion(boolean accepted, long deadlineNanos) {
+        this.accepted = accepted;
+        this.deadlineNanos = deadlineNanos;
+    }
+
+    public boolean accepted() {
+        return accepted;
+    }
+
+    public void complete(String resultCode, String spaceId) {
+        if (accepted) {
+            legacyResult.complete(new LegacyResult(resultCode, spaceId));
+        }
+    }
+
+    LegacyResult awaitLegacyResult() throws Exception {
+        long remainingNanos = deadlineNanos - System.nanoTime();
+        if (remainingNanos <= 0) {
+            throw new TimeoutException("Shadow inbound observation deadline expired");
+        }
+        return legacyResult.get(remainingNanos, TimeUnit.NANOSECONDS);
+    }
+
+    record LegacyResult(String resultCode, String spaceId) {
+    }
+}
--- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/OutboundShadowCompletion.java
@@ -0,0 +1,46 @@
+package com.ruoyi.algorithm.shadow;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public final class OutboundShadowCompletion {
+    private static final OutboundShadowCompletion REJECTED =
+            new OutboundShadowCompletion(false, 0L);
+
+    private final boolean accepted;
+    private final long deadlineNanos;
+    private final CompletableFuture<String> legacyResult = new CompletableFuture<>();
+
+    static OutboundShadowCompletion accepted(int timeoutSeconds) {
+        return new OutboundShadowCompletion(
+                true, System.nanoTime() + TimeUnit.SECONDS.toNanos(timeoutSeconds));
+    }
+
+    public static OutboundShadowCompletion rejected() {
+        return REJECTED;
+    }
+
+    private OutboundShadowCompletion(boolean accepted, long deadlineNanos) {
+        this.accepted = accepted;
+        this.deadlineNanos = deadlineNanos;
+    }
+
+    public boolean accepted() {
+        return accepted;
+    }
+
+    public void complete(String resultCode) {
+        if (accepted) {
+            legacyResult.complete(resultCode);
+        }
+    }
+
+    String awaitLegacyResult() throws Exception {
+        long remainingNanos = deadlineNanos - System.nanoTime();
+        if (remainingNanos <= 0) {
+            throw new TimeoutException("Shadow outbound observation deadline expired");
+        }
+        return legacyResult.get(remainingNanos, TimeUnit.NANOSECONDS);
+    }
+}
--- a/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/algorithm/shadow/ShadowObservationGateway.java
@@ -1,11 +1,15 @@
 package com.ruoyi.algorithm.shadow;

+import jakarta.annotation.PreDestroy;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;

 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;

@Component
 public class ShadowObservationGateway {
@@ -13,88 +17,95 @@ public class ShadowObservationGateway {

    private final ShadowObservationProperties properties;
    private final ShadowPlanningEngine engine;
-    private final AtomicBoolean planning = new AtomicBoolean();
+    private final ThreadPoolExecutor executor;

    public ShadowObservationGateway(ShadowObservationProperties properties, ShadowPlanningEngine engine) {
        this.properties = Objects.requireNonNull(properties, "properties");
        this.engine = engine;
+        this.executor = new ThreadPoolExecutor(
+                1, 1, 0L, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), runnable -> {
+                    Thread thread = new Thread(runnable, "ft546-shadow-observer");
+                    thread.setDaemon(true);
+                    return thread;
+                }, new ThreadPoolExecutor.AbortPolicy());
    }

-    public InboundShadowObservation planInbound(String taskNo, String trayNo) {
-        if (!properties.isEnabled() || engine == null) {
-            return InboundShadowObservation.empty();
-        }
-        if (!planning.compareAndSet(false, true)) {
-            return InboundShadowObservation.empty();
+    public InboundShadowCompletion observeInbound(String taskNo, String trayNo) {
+        if (!available()) {
+            return InboundShadowCompletion.rejected();
        }
+        InboundShadowCompletion completion =
+                InboundShadowCompletion.accepted(properties.getTimeoutSeconds());
        try {
-            return engine.planInbound(taskNo, trayNo);
-        } catch (RuntimeException exception) {
-            LOG.warn("Shadow inbound planning failed open for task {}: {}", taskNo, exception.getMessage());
-            return InboundShadowObservation.empty();
-        } finally {
-            planning.set(false);
+            executor.execute(() -> observeInbound(taskNo, trayNo, completion));
+            return completion;
+        } catch (RejectedExecutionException exception) {
+            LOG.debug("Shadow inbound observation skipped because worker is busy for task {}", taskNo);
+            return InboundShadowCompletion.rejected();
        }
    }

-    public OutboundShadowObservation planProjectOutbound(String stockoutDetailId) {
-        if (!properties.isEnabled() || engine == null) {
-            return OutboundShadowObservation.empty();
-        }
-        if (!planning.compareAndSet(false, true)) {
-            return OutboundShadowObservation.empty();
-        }
+    public OutboundShadowCompletion observeProjectOutbound(String stockoutDetailId) {
+        return observeOutbound(stockoutDetailId,
+                () -> engine.planProjectOutbound(stockoutDetailId));
+    }
+
+    public OutboundShadowCompletion observeExactStockOutbound(String stockoutDetailId, String stockId) {
+        return observeOutbound(stockoutDetailId,
+                () -> engine.planExactStockOutbound(stockoutDetailId, stockId));
+    }
+
+    private void observeInbound(String taskNo, String trayNo, InboundShadowCompletion completion) {
        try {
-            return engine.planProjectOutbound(stockoutDetailId);
-        } catch (RuntimeException exception) {
-            LOG.warn("Shadow outbound planning failed open for detail {}: {}",
-                    stockoutDetailId, exception.getMessage());
-            return OutboundShadowObservation.empty();
-        } finally {
-            planning.set(false);
+            InboundShadowObservation observation = engine.planInbound(taskNo, trayNo);
+            InboundShadowCompletion.LegacyResult legacy = completion.awaitLegacyResult();
+            engine.compareInbound(observation, legacy.resultCode(), legacy.spaceId());
+        } catch (InterruptedException exception) {
+            Thread.currentThread().interrupt();
+            LOG.warn("Shadow inbound observation interrupted for task {}", taskNo);
+        } catch (Exception exception) {
+            LOG.warn("Shadow inbound observation failed open for task {}: {}", taskNo, exception.getMessage());
        }
    }

-    public OutboundShadowObservation planExactStockOutbound(String stockoutDetailId, String stockId) {
-        if (!properties.isEnabled() || engine == null) {
-            return OutboundShadowObservation.empty();
-        }
-        if (!planning.compareAndSet(false, true)) {
-            return OutboundShadowObservation.empty();
+    private OutboundShadowCompletion observeOutbound(String stockoutDetailId,
+            OutboundPlanner planner) {
+        if (!available()) {
+            return OutboundShadowCompletion.rejected();
        }
+        OutboundShadowCompletion completion =
+                OutboundShadowCompletion.accepted(properties.getTimeoutSeconds());
        try {
-            return engine.planExactStockOutbound(stockoutDetailId, stockId);
-        } catch (RuntimeException exception) {
-            LOG.warn("Shadow exact-stock planning failed open for detail {} stock {}: {}",
-                    stockoutDetailId, stockId, exception.getMessage());
-            return OutboundShadowObservation.empty();
-        } finally {
-            planning.set(false);
+            executor.execute(() -> {
+                try {
+                    OutboundShadowObservation observation = planner.plan();
+                    engine.compareOutbound(observation, completion.awaitLegacyResult());
+                } catch (InterruptedException exception) {
+                    Thread.currentThread().interrupt();
+                    LOG.warn("Shadow outbound observation interrupted for detail {}", stockoutDetailId);
+                } catch (Exception exception) {
+                    LOG.warn("Shadow outbound observation failed open for detail {}: {}",
+                            stockoutDetailId, exception.getMessage());
+                }
+            });
+            return completion;
+        } catch (RejectedExecutionException exception) {
+            LOG.debug("Shadow outbound observation skipped because worker is busy for detail {}", stockoutDetailId);
+            return OutboundShadowCompletion.rejected();
        }
    }

-    public void compareInbound(InboundShadowObservation observation, String legacyResultCode,
-            String legacySpaceId) {
-        if (!properties.isEnabled() || engine == null || observation == null || !observation.observed()) {
-            return;
-        }
-        try {
-            engine.compareInbound(observation, legacyResultCode, legacySpaceId);
-        } catch (RuntimeException exception) {
-            LOG.warn("Shadow inbound comparison failed open for request {}: {}",
-                    observation.requestId(), exception.getMessage());
-        }
+    private boolean available() {
+        return properties.isEnabled() && engine != null && !executor.isShutdown();
    }

-    public void compareOutbound(OutboundShadowObservation observation, String legacyResultCode) {
-        if (!properties.isEnabled() || engine == null || observation == null || !observation.observed()) {
-            return;
-        }
-        try {
-            engine.compareOutbound(observation, legacyResultCode);
-        } catch (RuntimeException exception) {
-            LOG.warn("Shadow outbound comparison failed open for request {}: {}",
-                    observation.requestId(), exception.getMessage());
-        }
+    @PreDestroy
+    public void shutdown() {
+        executor.shutdownNow();
+    }
+
+    @FunctionalInterface
+    private interface OutboundPlanner {
+        OutboundShadowObservation plan();
    }
 }
--- a/ruoyi-admin/src/main/java/com/ruoyi/api/service/impl/ToWmsServiceImpl.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/api/service/impl/ToWmsServiceImpl.java
@@ -6,7 +6,7 @@ import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.ObjUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
-import com.ruoyi.algorithm.shadow.InboundShadowObservation;
+import com.ruoyi.algorithm.shadow.InboundShadowCompletion;
 import com.ruoyi.algorithm.shadow.ShadowObservationGateway;
 import com.ruoyi.api.domain.*;
 import com.ruoyi.api.service.IToErpService;
@@ -416,21 +416,21 @@ public class ToWmsServiceImpl implements IToWmsService {

                map.put("v_tray_code", inJob.getTrayCode());
                map.put("v_code", deviceCode);
-                InboundShadowObservation shadowObservation = InboundShadowObservation.empty();
+                InboundShadowCompletion shadowCompletion = InboundShadowCompletion.rejected();
                try {
-                    shadowObservation = shadowObservationGateway.planInbound(taskNo, inJob.getTrayCode());
+                    shadowCompletion = shadowObservationGateway.observeInbound(taskNo, inJob.getTrayCode());
                } catch (RuntimeException exception) {
-                    log.warn("Inbound shadow observation failed open for task {}: {}",
+                    log.warn("Inbound shadow submission failed open for task {}: {}",
                            taskNo, exception.getMessage());
                }
                mydoJobMapper.allotInJobSpace(map);
                String legacyResultCode = map.get("v_result").toString();
                if (!"10".equals(legacyResultCode)) {
-                    compareInboundShadow(shadowObservation, legacyResultCode, null);
+                    shadowCompletion.complete(legacyResultCode, null);
                    return AjaxResult.error(map.get("v_resultmsg").toString());
                }else{
                    MydoJob inJob2 = mydoJobMapper.selectInJobByTaskNo(taskNo);
-                    compareInboundShadow(shadowObservation, legacyResultCode,
+                    shadowCompletion.complete(legacyResultCode,
                            inJob2 == null ? null : inJob2.getCargoSpaceId());
                    return AjaxResult.success(inJob2);
                }
@@ -440,15 +440,4 @@ public class ToWmsServiceImpl implements IToWmsService {
        }
    }

-    private void compareInboundShadow(InboundShadowObservation observation,
-            String legacyResultCode, String legacySpaceId) {
-        try {
-            shadowObservationGateway.compareInbound(observation, legacyResultCode, legacySpaceId);
-        } catch (RuntimeException exception) {
-            log.warn("Inbound shadow comparison failed open for request {}: {}",
-                    observation == null ? null : observation.requestId(), exception.getMessage());
-        }
-    }
-
-
 }
--- a/ruoyi-admin/src/main/java/com/ruoyi/out/outStockout/service/impl/OutStockoutServiceImpl.java
+++ b/ruoyi-admin/src/main/java/com/ruoyi/out/outStockout/service/impl/OutStockoutServiceImpl.java
@@ -1,7 +1,7 @@
 package com.ruoyi.out.outStockout.service.impl;

 import cn.hutool.core.util.StrUtil;
-import com.ruoyi.algorithm.shadow.OutboundShadowObservation;
+import com.ruoyi.algorithm.shadow.OutboundShadowCompletion;
 import com.ruoyi.algorithm.shadow.ShadowObservationGateway;
 import com.ruoyi.api.service.IToWcsService;
 import com.ruoyi.bas.basJobId.service.IBasJobIdService;
@@ -188,10 +188,10 @@ public class OutStockoutServiceImpl implements IOutStockoutService
                map.put("v_stockout_detail_id",  detail.getStockoutDetailId());
                map.put("v_result", null);
                map.put("v_resultmsg", null);
-                OutboundShadowObservation shadowObservation =
-                        planProjectOutboundShadow(detail.getStockoutDetailId());
+                OutboundShadowCompletion shadowCompletion =
+                        observeProjectOutboundShadow(detail.getStockoutDetailId());
                outStockoutMapper.autoAllot(map);
-                compareOutboundShadow(shadowObservation, map.get("v_result"));
+                shadowCompletion.complete(map.get("v_result"));
                if (!"10".equals(map.get("v_result"))){
                    stringBuilder.append(",").append(detail.getStockoutDetailId()+map.get("v_resultmsg"));
                }
@@ -218,10 +218,10 @@ public class OutStockoutServiceImpl implements IOutStockoutService
        for (String stock : stockList) {
            map.put("v_stockout_detail_id", stockoutDetailId);
            map.put("v_stock_id", stock);
-            OutboundShadowObservation shadowObservation =
-                    planExactStockOutboundShadow(stockoutDetailId, stock);
+            OutboundShadowCompletion shadowCompletion =
+                    observeExactStockOutboundShadow(stockoutDetailId, stock);
            outStockoutMapper.manualAllot(map);
-            compareOutboundShadow(shadowObservation, map.get("v_result"));
+            shadowCompletion.complete(map.get("v_result"));
            //toWcsService.pushOutMydoJob();
            if (!"10".equals(map.get("v_result"))){
                stringBuilder.append(",").append(stockoutDetailId+map.get("v_resultString"));
@@ -233,32 +233,23 @@ public class OutStockoutServiceImpl implements IOutStockoutService
        return AjaxResult.success();
    }

-    private OutboundShadowObservation planProjectOutboundShadow(String stockoutDetailId) {
+    private OutboundShadowCompletion observeProjectOutboundShadow(String stockoutDetailId) {
        try {
-            return shadowObservationGateway.planProjectOutbound(stockoutDetailId);
+            return shadowObservationGateway.observeProjectOutbound(stockoutDetailId);
        } catch (RuntimeException exception) {
-            log.warn("Project outbound shadow observation failed open for detail {}: {}",
+            log.warn("Project outbound shadow submission failed open for detail {}: {}",
                    stockoutDetailId, exception.getMessage());
-            return OutboundShadowObservation.empty();
+            return OutboundShadowCompletion.rejected();
        }
    }

-    private OutboundShadowObservation planExactStockOutboundShadow(String stockoutDetailId, String stockId) {
+    private OutboundShadowCompletion observeExactStockOutboundShadow(String stockoutDetailId, String stockId) {
        try {
-            return shadowObservationGateway.planExactStockOutbound(stockoutDetailId, stockId);
+            return shadowObservationGateway.observeExactStockOutbound(stockoutDetailId, stockId);
        } catch (RuntimeException exception) {
-            log.warn("Exact-stock shadow observation failed open for detail {} stock {}: {}",
+            log.warn("Exact-stock shadow submission failed open for detail {} stock {}: {}",
                    stockoutDetailId, stockId, exception.getMessage());
-            return OutboundShadowObservation.empty();
-        }
-    }
-
-    private void compareOutboundShadow(OutboundShadowObservation observation, String legacyResultCode) {
-        try {
-            shadowObservationGateway.compareOutbound(observation, legacyResultCode);
-        } catch (RuntimeException exception) {
-            log.warn("Outbound shadow comparison failed open for request {}: {}",
-                    observation == null ? null : observation.requestId(), exception.getMessage());
+            return OutboundShadowCompletion.rejected();
        }
    }