feat(订单实时): 新增订单事件广播 Hub —— SSE 桶 + 心跳 + 单线程串行

java/src/main/java/com/payment/platform/module/payment/stream/OrderEventHub.java

@@ -0,0 +1,192 @@
+package com.payment.platform.module.payment.stream;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 订单实时信号的进程内广播器（部署形态是单实例，见 spec）。
+ *
+ * <p>桶模型：超管一桶 + 租户各一桶。超管**只进超管桶** —— 同一条连接收两份帧既是浪费、
+ * 又会让页面同一秒重拉两次。非超管且 tenantId 为 null：不进任何桶（保守，宁可少推）。
+ *
+ * <p>线程模型：对 emitter 的**每一次写**（注册时的 retry 帧、广播帧、心跳 ping）都提交到
+ * 同一条名为 {@code order-sse} 的守护线程上串行执行 —— {@code SseEmitter} 不允许多线程
+ * 并发 send。这条线程是自建的，**不是** Spring 的调度池：{@code spring.task.scheduling.pool.size}
+ * 默认 1、支付对账调度器就挂在上面，占用它会把对账挤停。
+ */
+@Slf4j
+@Component
+public class OrderEventHub {
+
+    /** emitter 超时：显式 1 小时。不赌 0L 的语义；到点自然断，前端重连 + 补拉接住。 */
+    public static final long TIMEOUT_MILLIS = TimeUnit.HOURS.toMillis(1);
+    /** 心跳间隔 20s：nginx proxy_read_timeout 默认 60s，留 3 倍余量。 */
+    public static final long PING_INTERVAL_SECONDS = 20;
+    /** 重连提示：浏览器断线后 3s 再来（首次连接不受影响）。 */
+    public static final long RECONNECT_TIME_MILLIS = 3000L;
+
+    private final List<SseEmitter> superAdmins = new CopyOnWriteArrayList<>();
+    private final Map<Long, List<SseEmitter>> byTenant = new ConcurrentHashMap<>();
+
+    private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(r -> {
+        Thread t = new Thread(r, "order-sse");
+        t.setDaemon(true);
+        return t;
+    });
+    private ScheduledFuture<?> heartbeat;
+
+    /**
+     * 注册一条连接。返回的 emitter 交给 Spring MVC 接管响应。
+     *
+     * @param tenantId   登录用户的租户；可为 null
+     * @param superAdmin is_superuser 开关（与 TenantInnerInterceptor 用的是同一个）
+     */
+    public SseEmitter register(Long tenantId, boolean superAdmin) {
+        SseEmitter emitter = new SseEmitter(TIMEOUT_MILLIS);
+        List<SseEmitter> bucket = superAdmin ? superAdmins
+                : (tenantId == null ? null : byTenant.computeIfAbsent(tenantId, k -> new CopyOnWriteArrayList<>()));
+        if (bucket == null) {
+            // 无租户的非超管：收不到任何信号，立刻正常结束，别让客户端挂满 1 小时
+            log.info("[订单SSE] 无租户的非超管登录，不建连接");
+            emitter.complete();
+            return emitter;
+        }
+        // ★ spec 的「实现前必须先验证的小项」里有一条：非超管 + tenantId=null 在
+        //   TenantInnerInterceptor 下到底看不看得见那些行。**这一条没有做**，如实记在这里：
+        //   它改不了本通道的行为 —— 不入桶就是不推，而这位用户取数一直走列表接口，
+        //   加通道前后完全一致（不回归、也不泄露）。哪天真要给他推，改的是上面这个分支 + 一条用例。
+        //   别把它写成「已验证」。
+        emitter.onCompletion(() -> remove(bucket, emitter));
+        emitter.onTimeout(() -> remove(bucket, emitter));
+        emitter.onError(e -> remove(bucket, emitter));
+        bucket.add(emitter);
+        // retry 帧在 emitter 初始化前 send 会进容器的 earlySendAttempts 队列、由 initialize 时补发；
+        // 仍排到同一条线程上，保持「所有写都在 order-sse」这条不变式
+        executor.execute(() -> sendRetry(emitter));
+        return emitter;
+    }
+
+    /** 同步广播：调用线程直接发。生产别用（见 {@link #submitBroadcast}），测试用。 */
+    public void broadcast(String channel, Long tenantId) {
+        deliver(superAdmins, channel);
+        if (tenantId != null) {
+            List<SseEmitter> bucket = byTenant.get(tenantId);
+            if (bucket != null) {
+                deliver(bucket, channel);
+            }
+        }
+    }
+
+    /** 生产路径：提交到 order-sse 单线程上执行，与心跳串行。 */
+    public void submitBroadcast(String channel, Long tenantId) {
+        executor.execute(() -> broadcast(channel, tenantId));
+    }
+
+    private void deliver(List<SseEmitter> bucket, String channel) {
+        for (SseEmitter emitter : bucket) {
+            if (!sendFrame(emitter, channel)) {
+                remove(bucket, emitter);
+            }
+        }
+    }
+
+    private void remove(List<SseEmitter> bucket, SseEmitter emitter) {
+        bucket.remove(emitter);
+        if (bucket != superAdmins && bucket.isEmpty()) {
+            // 空桶删 key：否则每个登录过的租户都留一个空 List 到进程结束
+            byTenant.values().remove(bucket);
+        }
+    }
+
+    /** 帧格式：有名事件 + 一个 JSON 对象。**不带数据** —— 页面拿信号去重拉它自己的列表接口。 */
+    protected boolean sendFrame(SseEmitter emitter, String channel) {
+        try {
+            emitter.send(SseEmitter.event().name("changed").data(Map.of("channel", channel)));
+            return true;
+        } catch (Exception e) {
+            log.debug("[订单SSE] 发送失败，摘除连接: channel={}, err={}", channel, e.getMessage());
+            return false;
+        }
+    }
+
+    protected void sendRetry(SseEmitter emitter) {
+        try {
+            emitter.send(SseEmitter.event().reconnectTime(RECONNECT_TIME_MILLIS));
+        } catch (Exception e) {
+            // 注册瞬间客户端可能已经走了：忽略，onError/onCompletion 会清
+            log.debug("[订单SSE] 重连提示发送失败: {}", e.getMessage());
+        }
+    }
+
+    /** 心跳：有名事件 ping（注释帧 JS 看不见，前端看门狗就废了）。 */
+    protected boolean sendPing(SseEmitter emitter) {
+        try {
+            emitter.send(SseEmitter.event().name("ping").data("{}"));
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
+
+    @PostConstruct
+    void startHeartbeat() {
+        heartbeat = executor.scheduleAtFixedRate(this::pingAll,
+                PING_INTERVAL_SECONDS, PING_INTERVAL_SECONDS, TimeUnit.SECONDS);
+        log.info("[订单SSE] 广播器就绪（单线程 order-sse，心跳 {}s）", PING_INTERVAL_SECONDS);
+    }
+
+    private void pingAll() {
+        pingBucket(superAdmins);
+        for (List<SseEmitter> bucket : byTenant.values()) {
+            pingBucket(bucket);
+        }
+    }
+
+    private void pingBucket(List<SseEmitter> bucket) {
+        for (SseEmitter emitter : bucket) {
+            if (!sendPing(emitter)) {
+                remove(bucket, emitter);
+            }
+        }
+    }
+
+    @PreDestroy
+    void shutdown() {
+        if (heartbeat != null) {
+            heartbeat.cancel(false);
+        }
+        executor.shutdownNow();
+    }
+
+    /** 仅测试用：当前连接总数。 */
+    int connectionCount() {
+        int n = superAdmins.size();
+        for (List<SseEmitter> bucket : byTenant.values()) {
+            n += bucket.size();
+        }
+        return n;
+    }
+
+    /** 仅测试用。 */
+    int tenantConnectionCount(long tenantId) {
+        List<SseEmitter> bucket = byTenant.get(tenantId);
+        return bucket == null ? 0 : bucket.size();
+    }
+
+    /** 仅测试用：还留着几个租户桶（空桶应被删掉）。 */
+    int tenantBucketCount() {
+        return byTenant.size();
+    }
+}

java/src/test/java/com/payment/platform/module/payment/stream/OrderEventHubTest.java

@@ -0,0 +1,142 @@
+package com.payment.platform.module.payment.stream;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Hub 的桶路由与线程模型。测试里**不碰真 emitter**（没有 HTTP 响应可写）：
+ * RecordingHub 覆写全部 send 出口记账，SseEmitter 只当身份令牌用。
+ */
+class OrderEventHubTest {
+
+    private final RecordingHub hub = new RecordingHub();
+
+    @AfterEach
+    void tearDown() {
+        hub.shutdown();      // 每个用例都要收：Hub 构造时起了执行器与心跳，不收会攒线程
+    }
+
+    @Test
+    void broadcast_routesOnlyToTheSameTenant() {
+        SseEmitter t1 = hub.register(1L, false);
+        SseEmitter t2 = hub.register(2L, false);
+
+        hub.broadcast("lsy_order", 1L);
+
+        assertEquals(List.of("lsy_order"), hub.framesOf(t1));
+        assertTrue(hub.framesOf(t2).isEmpty(), "别的租户一帧都不该收到");
+    }
+
+    @Test
+    void broadcast_reachesSuperAdminForAnyTenant() {
+        SseEmitter admin = hub.register(null, true);
+
+        hub.broadcast("har_order", 2L);
+
+        assertEquals(List.of("har_order"), hub.framesOf(admin));
+    }
+
+    @Test
+    void register_superAdminWithTenantId_doesNotDoubleReceive() {
+        // 超管也可能带 tenantId（UserDetailsServiceImpl 把 sys_user.tenant_id 原样搬进 LoginUser）。
+        // 两条桶都收他 = 一条连接同帧收两次 = 页面同秒重拉两次。
+        SseEmitter admin = hub.register(2L, true);
+
+        hub.broadcast("lsy_order", 2L);
+
+        assertEquals(List.of("lsy_order"), hub.framesOf(admin));
+    }
+
+    @Test
+    void register_nonSuperAdminWithoutTenant_getsNothing() {
+        SseEmitter nobody = hub.register(null, false);
+
+        hub.broadcast("lsy_order", 1L);
+
+        assertTrue(hub.framesOf(nobody).isEmpty());
+        assertEquals(0, hub.connectionCount(), "无租户的非超管不进任何桶（保守：宁可少推）");
+    }
+
+    @Test
+    void broadcast_sendFailure_removesConnectionAndEmptyBucket() {
+        SseEmitter dead = hub.register(3L, false);
+        hub.failFor.put(dead, true);
+
+        hub.broadcast("lsy_refund", 3L);
+
+        assertEquals(0, hub.connectionCount(), "发不出去的连接要立刻摘掉");
+        assertEquals(0, hub.tenantBucketCount(), "空桶要删 key，否则每个登录过的租户永远留一个空 List");
+    }
+
+    @Test
+    void submitBroadcast_everyFrameGoesOutOnTheOrderSseThread() throws Exception {
+        SseEmitter t1 = hub.register(1L, false);
+        hub.expectFrames(8);
+        ExecutorService callers = Executors.newFixedThreadPool(8);
+        try {
+            for (int i = 0; i < 8; i++) {
+                callers.execute(() -> hub.submitBroadcast("lsy_order", 1L));
+            }
+            assertTrue(hub.awaitFrames(5, TimeUnit.SECONDS), "8 次提交应在 5 秒内出完 8 帧");
+        } finally {
+            callers.shutdownNow();
+        }
+        assertEquals(8, hub.framesOf(t1).size());
+        assertEquals(List.of("order-sse"), hub.frameThreads.stream().distinct().toList(),
+                "所有 send 必须落在同一条 order-sse 线程上 —— SseEmitter 禁并发 send，"
+                        + "这条红了说明有人在调用线程上直接发了");
+    }
+
+    /** 覆写全部 send 出口：只记账，不碰真 emitter。 */
+    static class RecordingHub extends OrderEventHub {
+        final Map<SseEmitter, List<String>> sent = new ConcurrentHashMap<>();
+        final Map<SseEmitter, Boolean> failFor = new ConcurrentHashMap<>();
+        final List<String> frameThreads = new CopyOnWriteArrayList<>();
+        private volatile CountDownLatch latch = new CountDownLatch(0);
+
+        void expectFrames(int n) {
+            latch = new CountDownLatch(n);
+        }
+
+        boolean awaitFrames(long timeout, TimeUnit unit) throws InterruptedException {
+            return latch.await(timeout, unit);
+        }
+
+        List<String> framesOf(SseEmitter emitter) {
+            return sent.getOrDefault(emitter, List.of());
+        }
+
+        @Override
+        protected boolean sendFrame(SseEmitter emitter, String channel) {
+            if (Boolean.TRUE.equals(failFor.get(emitter))) {
+                return false;
+            }
+            sent.computeIfAbsent(emitter, k -> new CopyOnWriteArrayList<>()).add(channel);
+            frameThreads.add(Thread.currentThread().getName());
+            latch.countDown();
+            return true;
+        }
+
+        @Override
+        protected void sendRetry(SseEmitter emitter) {
+            // 注册时的 retry 帧不记账：它不属于任何通道，计进去会打乱帧计数
+        }
+
+        @Override
+        protected boolean sendPing(SseEmitter emitter) {
+            return true;   // 心跳在单测里是噪声（20s 才来一次，用例早结束了）
+        }
+    }
+}