feat(telemetry): 批量事件入口改 RocketMQ 异步消费(幂等·聚合/回灌行为不变) (W-REAL R1)

上报入口 /events/batch·/perf/beacon 从请求线程内同步写改为「轻量校验 + 投递 telemetry-event 主题、
消费侧异步落库+聚合+quality_score 回灌 feed」。灰度开关 telemetry.mq.enabled(默认关=同步回落、现行行为逐字不变)。

- TelemetryEventProducer:逐条投递 event/perf tag、keys=traceId、syncSend 带超时+RocketMQ 自带重试;
  发送失败不外抛、按 rejected 交前端重传兜底(不 500 阻断上报)。
- TelemetryEventConsumer:@RocketMQMessageListener 落地骨架,consumeThreadNumber=consumeThreadMax=8
  (min=max 破 consumeThreadMax 死参数坑,受 DB 池默认 10 约束);瞬时故障外抛触发 ≤16 次重投转 DLQ,
  毒消息吞掉 ack 防重投风暴;幂等由 eventId(uk_event_id)保证重投不双计。
- EventIngestServiceImpl:入口路由(MQ 在席=投递、缺席=同步回落),新增 consumeFromMq 单条事务入口
  逐字复用同步写核心 ingestOne——聚合行为与 quality_score 回灌 feed 行为与改造前一致,仅执行位置搬到消费线程;
  ObjectProvider 软注入生产者 + @Lazy 自代理保回落单条事务真生效。

验证:模块 59 单测绿(既有 EventIngestServiceImplTest 13 个逐字不变通过=回灌行为不变;新增生产者3/消费者4/
MQ路径5);连 mini-infra 真 broker(100.64.0.8:9876)往返 IT 绿(produce→broker→consume→consumeFromMq 转调,
往返序列化保真)。整条 HTTP POST→MQ→真库聚合行 的单体 e2e 待后端全栈窗口(同阶段〇 C1/C2/C3 处置)。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

game-cloud/game-module-telemetry/game-module-telemetry-server/pom.xml

@@ -100,6 +100,13 @@
             <artifactId>huijing-spring-boot-starter-rpc</artifactId>
         </dependency>
 
+        <!-- RocketMQ（切片一 阶段〇 R1：telemetry 批量事件入口改 MQ 异步消费）：RocketMQTemplate / @RocketMQMessageListener。
+             starter-mq 以 optional=true 声明本 starter（不传递），故本模块须显式直依赖；版本由 huijing 依赖管理统一控制。 -->
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot-starter</artifactId>
+        </dependency>
+
         <!-- 测试 -->
         <dependency>
             <groupId>com.wanxiang</groupId>

game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/mq/consumer/TelemetryEventConsumer.java

@@ -1,51 +1,80 @@
 package com.wanxiang.huijing.game.module.telemetry.mq.consumer;
 
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.game.module.telemetry.mq.message.TelemetryEventMessage;
+import com.wanxiang.huijing.game.module.telemetry.mq.producer.TelemetryEventProducer;
+import com.wanxiang.huijing.game.module.telemetry.service.event.EventIngestService;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 
+import jakarta.annotation.Resource;
+
 /**
- * 遥测事件 MQ 消费者（脊柱对接点骨架；落原始事件 + 增量聚合 + 回灌）
+ * 遥测事件 MQ 消费者（切片一 阶段〇 R1）——异步消费事件信封，落原始事件 + 增量聚合 + quality_score 回灌 feed。
  *
- * 本类为外部依赖对接点，骨架阶段不实现具体消费逻辑（不 mock、不实现），仅声明对接契约与 TODO，保证集成可追溯。
- * 契约见 contracts/api-schemas/telemetry.yaml#x-mq-contracts.consumer：
- *   - group=telemetry-event-consumer，topic=telemetry-event，无序并发消费（聚合按 gameId 行级原子累加避免覆盖）。
- *   - 幂等键 = (traceId, event, ts)，命中 game_telemetry_event.uk_dedup 即跳过（INSERT IGNORE 语义），
- *     保证至少一次投递下不重复聚合。
- *   - 默认重试 ≤16 次仍失败转 DLQ（telemetry-event-dlq），由 T-TEL-19 数据质量监控告警。
+ * <p><b>脊柱对接点落地</b>（契约 {@code contracts/api-schemas/telemetry.yaml#x-mq-contracts.consumer}）：
+ *   topic={@link TelemetryEventProducer#TOPIC}、consumerGroup=telemetry-event-consumer、并发消费（无序）。
+ *   消费副作用逐字复用同步写核心 {@code EventIngestServiceImpl#ingestOne}（经 {@link EventIngestService#consumeFromMq}
+ *   单条事务包裹）——<b>聚合行为与 quality_score 回灌 feed 行为与改造前逐字一致</b>，只是执行位置从请求线程搬到消费线程。
  *
- * 消费副作用（待实现）：
- *   1. 落 game_telemetry_event（原始事件，幂等）—— 经 TelemetryEventMapper insert，依赖 uk_dedup 去重。
- *   2. 增量聚合 game_telemetry_game_stat（play/like/quality 等按 gameId+statDate 行级累加，唯一约束 uk_game_date 保证幂等）。
- *   3. quality_score 计算（运营质量分 0-100；输入：avgDurationMs/completedCount/reportCount/loadFailCount 等）——
- *      TODO 对接点：计算公式待产品/数据侧确定后落地，不在骨架阶段臆造公式。
- *   4. 回灌：play_count/like_count → project（game_project）；quality_score → feed 排序信号——
- *      TODO 对接点：待 project/feed 的 -api（Feign）就绪后在事务提交后异步回灌（远程调用不放事务内）。
+ * <p><b>幂等</b>：以信封 {@code eventId}（uk_event_id）为幂等键——同步写核心先 selectByEventId 预查、命中即跳过累加，
+ *   叠加 uk_event_id 唯一约束硬后盾；「至少一次投递」下重复消息不重复聚合（契约 V10 起 uk_dedup 已被 uk_event_id 取代）。
+ *
+ * <p><b>并发上限（{@code consumeThreadMax} 死参数坑）</b>：rocketmq-client 无界消费队列致 {@code consumeThreadMax}
+ *   不单独生效（线程数永不超 core），故与 {@code consumeThreadNumber} <b>同设为 8（min=max）</b>让消费并发上限真为 8；
+ *   MQ 只 cap 投递（消费）速率，不是「在跑生成并发」那种业务并发。8 取值受 DB 连接池（默认 10）约束留头寸，
+ *   每条消费=一次快 DB 写事务，可按吞吐与池大小调（改这里需同调池）。
+ *
+ * <p><b>重试与 DLQ（外部交互纪律）</b>：消费抛异常则 RocketMQ 重投（默认 ≤16 次）仍失败转 DLQ（%RETRY%/%DLQ% 组），
+ *   由 T-TEL-19 数据质量监控告警。故对<b>瞬时故障</b>（DB 抖动）本方法<b>外抛让其重试</b>（单条事务已回滚、幂等保证重试不双计）；
+ *   对<b>毒消息</b>（信封为空等结构性坏数据，重投无益）<b>吞掉正常返回</b>（ack），只告警留痕，防无限重投风暴。
+ *
+ * <p><b>装配条件</b>：{@code telemetry.mq.enabled=true} 才装配（与生产者同一灰度开关）——默认关时不起消费者、不连 NameServer。
  *
  * @author 绘境AI
  */
 @Slf4j
 @Component
-public class TelemetryEventConsumer {
+@ConditionalOnProperty(prefix = "telemetry.mq", name = "enabled", havingValue = "true")
+@RocketMQMessageListener(
+        topic = TelemetryEventProducer.TOPIC,
+        consumerGroup = "telemetry-event-consumer",
+        consumeThreadNumber = 8,  // min=max=8：消费并发上限真为 8（受 DB 池默认 10 约束留头寸；见类注释死参数坑）
+        consumeThreadMax = 8      // rocketmq-client 无界队列致本参不单独生效，须与 consumeThreadNumber 同值
+)
+public class TelemetryEventConsumer implements RocketMQListener<TelemetryEventMessage> {
+
+    /** 事件摄取 Service（消费副作用逐字复用同步写核心，保证聚合/回灌行为不变）。 */
+    @Resource
+    private EventIngestService eventIngestService;
 
     /**
-     * TODO 对接点（MQ 消费）：补全 RocketMQ 监听注解与消费方法。
-     *   待 huijing-spring-boot-starter-mq 的 RocketMQ listener 就绪后，按如下骨架实现：
+     * 消费一条事件信封：单条事务落库 + 聚合 + quality_score 回灌（幂等）。
      *
-     *   @RocketMQMessageListener(topic = "telemetry-event", consumerGroup = "telemetry-event-consumer")
-     *   public class Xxx implements RocketMQListener<EnvelopeMessage> {
-     *       public void onMessage(EnvelopeMessage msg) {
-     *           // 1) 幂等落原始表（uk_dedup 去重，INSERT IGNORE）
-     *           // 2) 处理状态机 0待聚合 → 1已聚合（EventProcessStatusEnum 校验合法流转）
-     *           // 3) 增量聚合 game_telemetry_game_stat（行级原子累加）
-     *           // 4) 计算 quality_score（公式 TODO）
-     *           // 5) 事务提交后异步回灌 feed/project（Feign 调用不放事务内）
-     *       }
-     *   }
-     *
-     * 骨架阶段仅记 INFO 占位说明对接点存在，避免遗漏。
+     * @param message MQ 消息体（信封 + 上报时点 userId，见 {@link TelemetryEventMessage}）
      */
-    public TelemetryEventConsumer() {
-        log.info("[TelemetryEventConsumer] 骨架对接点已就位；MQ 消费/聚合/回灌 待对接（见类注释 TODO）");
+    @Override
+    public void onMessage(TelemetryEventMessage message) {
+        // 毒消息守卫：结构性坏数据（消息/信封/eventId 为空）重投无益 → 吞掉正常返回（ack），只告警，防无限重投
+        if (message == null || message.getEnvelope() == null) {
+            log.error("[telemetry-mq] 消息或信封为空，丢弃不重投 message={}", message);
+            return;
+        }
+        EnvelopeReqVO envelope = message.getEnvelope();
+        try {
+            // 复用同步写核心（单条 @Transactional）：落原始事件（eventId 幂等）→ 增量聚合 → 算 quality_score → 回灌 feed
+            eventIngestService.consumeFromMq(envelope, message.getUserId());
+            log.info("[telemetry-mq] 消费事件完成 eventId={} event={} traceId={}",
+                    envelope.getEventId(), envelope.getEvent(), envelope.getTraceId());
+        } catch (Exception e) {
+            // 瞬时故障（DB 抖动等）：外抛让 RocketMQ 重投（单条事务已回滚，幂等保证重试不双计），≤16 次仍失败转 DLQ 告警。
+            //   错误路径必须留痕（创始人铁律）。
+            log.error("[telemetry-mq] 消费事件失败，外抛触发重投（幂等保证不双计；≤16 次失败转 DLQ）eventId={} event={} traceId={}",
+                    envelope.getEventId(), envelope.getEvent(), envelope.getTraceId(), e);
+            throw e;
+        }
     }
-
 }

game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/mq/message/TelemetryEventMessage.java

@@ -0,0 +1,43 @@
+package com.wanxiang.huijing.game.module.telemetry.mq.message;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * 遥测事件 MQ 消息体（切片一 阶段〇 R1：批量事件入口改 MQ 异步消费）。
+ *
+ * <p>契约对齐 {@code contracts/api-schemas/telemetry.yaml#x-mq-contracts.producer.messageBody}：核心是
+ *   逐条投递的事件信封 {@link EnvelopeReqVO}；此外携带上报时点解析出的登录 {@code userId}（传输关切、
+ *   非契约字段）——异步消费脱离了 Web 请求线程，无法再从 SecurityContext 取登录态，故须在生产时随信封带上，
+ *   保证消费侧算 user_id 兜底与审计列填充与同步写路径逐字一致（见 {@code EventIngestServiceImpl#consumeFromMq}）。
+ *
+ * <p>幂等真身 = 信封 {@code eventId}（消费侧命中 {@code uk_event_id} 去重）；MQ 消息 keys 用 traceId 仅便于排障对账。
+ *   经 rocketmq-spring 默认 Jackson 转换器序列化/反序列化往返；字段均为 POJO，无循环引用。
+ *
+ * @author 绘境AI
+ */
+@Data
+public class TelemetryEventMessage implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    /** 单条事件信封（幂等真身 eventId 在其内；消费侧据此落库 + 聚合 + 回灌）。 */
+    private EnvelopeReqVO envelope;
+
+    /**
+     * 上报时点解析的登录用户 ID（匿名为 null）。异步消费无 Web 请求线程、取不到登录态，故随消息携带，
+     * 供消费侧 resolveUserId 兜底 + 匿名审计列兜底判定，与同步写路径口径一致。
+     */
+    private Long userId;
+
+    /** 空构造（Jackson 反序列化需要）。 */
+    public TelemetryEventMessage() {
+    }
+
+    public TelemetryEventMessage(EnvelopeReqVO envelope, Long userId) {
+        this.envelope = envelope;
+        this.userId = userId;
+    }
+}

game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/mq/producer/TelemetryEventProducer.java

@@ -0,0 +1,96 @@
+package com.wanxiang.huijing.game.module.telemetry.mq.producer;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.game.module.telemetry.mq.message.TelemetryEventMessage;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.stereotype.Component;
+import org.springframework.util.StringUtils;
+
+/**
+ * 遥测事件 MQ 生产者（切片一 阶段〇 R1）——上报通道轻量校验后逐条投递事件信封到 {@code telemetry-event} 主题，
+ * 把「落库 + 聚合 + quality_score 回灌」从请求线程内同步写改为消费侧异步完成（幂等）。
+ *
+ * <p><b>契约对齐</b> {@code contracts/api-schemas/telemetry.yaml#x-mq-contracts.producer}：topic={@link #TOPIC}、
+ *   tag=event|perf（{@code /events/batch} 用 event、{@code /perf/beacon} 用 perf）、keys=traceId、逐条投递、orderly=false。
+ *
+ * <p><b>外部交互纪律（创始人铁律）</b>：syncSend 带发送超时 {@link #SEND_TIMEOUT_MS}；RocketMQ 生产者自带
+ *   发送失败重试（retryTimesWhenSendFailed 默认 2）作瞬时抖动兜底；<b>幂等</b>由消费侧 {@code eventId}（uk_event_id）去重
+ *   保证「至少一次投递」不重复聚合。发送最终失败<b>不外抛</b>——返回 {@code false} 交由调用方按「受理失败(rejected)」处置，
+ *   由前端 SDK 分批重传作最外层兜底（遥测=分析数据、可容忍偶发丢失且客户端重试，绝不 500 阻断上报）。
+ *
+ * <p><b>装配条件</b>：{@code telemetry.mq.enabled=true} 才装配（与消费者同一灰度开关）——默认关时本 Bean 缺席，
+ *   {@code EventIngestServiceImpl} 经 {@code ObjectProvider} 取空、回落原同步写路径（现行行为逐字不变）。
+ *
+ * @author 绘境AI
+ */
+@Slf4j
+@Component
+@ConditionalOnProperty(prefix = "telemetry.mq", name = "enabled", havingValue = "true")
+public class TelemetryEventProducer {
+
+    /** 遥测事件主题（与消费者 {@code TelemetryEventConsumer} 同名订阅；契约 telemetry-event）。 */
+    public static final String TOPIC = "telemetry-event";
+
+    /** 事件通道 tag（/events/batch）。 */
+    public static final String TAG_EVENT = "event";
+
+    /** 性能兜底通道 tag（/perf/beacon）。 */
+    public static final String TAG_PERF = "perf";
+
+    /** 发送超时（毫秒）——外部交互必设超时，避免上报请求被 MQ 抖动无限拖住。 */
+    private static final long SEND_TIMEOUT_MS = 3000L;
+
+    /** RocketMQ 模板（rocketmq-spring 自动装配；name-server 见 application*.yaml → mini-infra 100.64.0.8:9876）。 */
+    private final RocketMQTemplate rocketMQTemplate;
+
+    public TelemetryEventProducer(RocketMQTemplate rocketMQTemplate) {
+        this.rocketMQTemplate = rocketMQTemplate;
+    }
+
+    /**
+     * 投递一条事件信封到 {@link #TOPIC}（同步发 + 超时；成功 SEND_OK 返 true，否则/异常返 false）。
+     *
+     * @param envelope 已通过轻量校验的单条信封
+     * @param userId   上报时点登录用户 ID（匿名为 null；随消息携带供消费侧口径一致）
+     * @param tag      消息 tag（{@link #TAG_EVENT} / {@link #TAG_PERF}）
+     * @return 是否投递成功（true=broker 已确认 SEND_OK；false=发送失败/异常，调用方按 rejected 处置）
+     */
+    public boolean publish(EnvelopeReqVO envelope, Long userId, String tag) {
+        // 目的地格式 topic:tag（rocketmq-spring 约定）
+        String destination = TOPIC + ":" + tag;
+        try {
+            // 消息体 = 信封 + userId；keys 置 traceId（便于全链路排障对账，非幂等键——幂等真身是 eventId）
+            TelemetryEventMessage payload = new TelemetryEventMessage(envelope, userId);
+            MessageBuilder<TelemetryEventMessage> builder = MessageBuilder.withPayload(payload);
+            if (StringUtils.hasText(envelope.getTraceId())) {
+                builder.setHeader(RocketMQHeaders.KEYS, envelope.getTraceId());
+            }
+            Message<TelemetryEventMessage> message = builder.build();
+
+            SendResult result = rocketMQTemplate.syncSend(destination, message, SEND_TIMEOUT_MS);
+            boolean ok = result != null && result.getSendStatus() == SendStatus.SEND_OK;
+            if (ok) {
+                log.info("[telemetry-mq] 投递事件信号 topic={} tag={} eventId={} traceId={} msgId={}",
+                        TOPIC, tag, envelope.getEventId(), envelope.getTraceId(),
+                        result.getMsgId());
+            } else {
+                // 发送非 SEND_OK（如刷盘/主从超时）：不外抛，返 false 交调用方按 rejected 处置（客户端重传兜底）
+                log.warn("[telemetry-mq] 投递事件非 SEND_OK topic={} tag={} eventId={} status={}",
+                        TOPIC, tag, envelope.getEventId(), result == null ? null : result.getSendStatus());
+            }
+            return ok;
+        } catch (Exception e) {
+            // 外部交互错误路径必须留痕：MQ 抖动/超时→不外抛（不 500 阻断上报），返 false 计 rejected，前端 SDK 分批重传兜底
+            log.warn("[telemetry-mq] 投递事件失败（不外抛，按 rejected 处置由前端重传兜底）topic={} tag={} eventId={} traceId={}",
+                    TOPIC, tag, envelope.getEventId(), envelope.getTraceId(), e);
+            return false;
+        }
+    }
+}

game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestService.java

@@ -32,4 +32,16 @@ public interface EventIngestService {
      */
     Boolean ingestBeacon(EnvelopeReqVO reqVO, Long userId);
 
+    /**
+     * MQ 消费侧单条摄取（切片一 阶段〇 R1）——异步消费入口，单条事务落库 + 聚合 + quality_score 回灌。
+     *
+     * <p>由 {@code TelemetryEventConsumer} 调用（消费线程）；MQ 关闭时也作 {@code ingestBatch}/{@code ingestBeacon}
+     *   的同步回落逐条入口（经代理保证单条事务）。逐字复用同步写核心，保证聚合/回灌行为与改造前一致；
+     *   幂等由信封 eventId（uk_event_id）保证，重复投递不重复聚合。
+     *
+     * @param reqVO  单条信封（已在生产侧通过轻量校验）
+     * @param userId 上报时点登录用户 ID（匿名为 null；随 MQ 消息携带，保证消费侧口径与同步写一致）
+     */
+    void consumeFromMq(EnvelopeReqVO reqVO, Long userId);
+
 }

game-cloud/game-module-telemetry/game-module-telemetry-server/src/main/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImpl.java

@@ -9,6 +9,7 @@ import com.wanxiang.huijing.game.module.telemetry.dal.mysql.event.TelemetryEvent
 import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.GameStatMapper;
 import com.wanxiang.huijing.game.module.telemetry.enums.EventProcessStatusEnum;
 import com.wanxiang.huijing.game.module.telemetry.enums.TelemetryEventEnum;
+import com.wanxiang.huijing.game.module.telemetry.mq.producer.TelemetryEventProducer;
 import com.wanxiang.huijing.game.module.telemetry.service.quality.AnomalyFilterService;
 import com.wanxiang.huijing.game.module.feed.api.FeedApi;
 import com.wanxiang.huijing.game.module.feed.dto.FeedRankUpsertReqDTO;
@@ -18,6 +19,8 @@ import com.wanxiang.huijing.framework.common.util.servlet.ServletUtils;
 import com.wanxiang.huijing.framework.security.core.LoginUser;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
 import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Service;
@@ -99,61 +102,53 @@ public class EventIngestServiceImpl implements EventIngestService {
     @Resource
     private AnomalyFilterService anomalyFilterService;
 
-    /*
-     * TODO 对接点（MQ 生产，§3.5 仅留契约+TODO 不实现）：拆"上报快速 ACK + 异步消费聚合"时，
-     *   注入 RocketMQ 生产者投递 topic=telemetry-event（幂等键=eventId，对应 uk_event_id），消费侧复用本类同步写聚合逻辑。
-     *   见 contracts/api-schemas/telemetry.yaml#x-mq-contracts。MVP 同步写已满足闭环，不引入 MQ。
+    /**
+     * 遥测事件 MQ 生产者软注入（切片一 阶段〇 R1）：{@code telemetry.mq.enabled=true} 才在席——上报入口把
+     *   「落库 + 聚合 + quality_score 回灌」改为快速 ACK + 投递 {@code telemetry-event} 主题、消费侧异步完成（幂等）。
+     * <p>{@code enabled=false}（默认）时缺席、{@link ObjectProvider#getIfAvailable()} 取空 → 回落原同步写路径
+     *   （现行行为逐字不变，既有单测据此仍绿）。纯 Mockito 单测不注入本 Provider，{@link #resolveProducer()} 空值守卫回落同步写。
      */
+    @Resource
+    private ObjectProvider<TelemetryEventProducer> telemetryEventProducerProvider;
 
+    /**
+     * 自身代理引用（切片一 阶段〇 R1）：MQ 关闭回落同步写时，逐条经代理调 {@link #consumeFromMq}，保证 {@code @Transactional}
+     *   单条事务真生效（Spring 自调用不走代理、事务失效，故显式经代理）。{@code @Lazy} 破本 Bean 自引用初始化环。
+     * <p>纯 Mockito 单测无 Spring 代理，本字段为 null，{@link #invokeConsumeFromMq}/{@link #ingestBatchSyncFallback}
+     *   空值守卫退化为直调 {@code this}（事务注解在单测本就无 Spring 语义、不影响 mapper 交互断言）。
+     */
+    @Lazy
+    @Resource
+    private EventIngestService self;
+
+    // ============================== 上报入口（MQ 异步路由 / 同步回落）==============================
+
+    /**
+     * 批量摄取（切片一 阶段〇 R1 路由）：MQ 开启则逐条轻量校验后投递 {@code telemetry-event}（快速 ACK、消费侧异步聚合）；
+     * MQ 关闭则回落原同步写（逐条单事务经代理）。批量上限超限整体拒绝——先于任何投递/落库，与改造前口径一致。
+     */
     @Override
-    @Transactional(rollbackFor = Exception.class) // §3.5 C5 同步写：落库+聚合+回灌+置已聚合在同一本地事务，任一步失败整批回滚
     public EventBatchResultVO ingestBatch(EventBatchReqVO reqVO, Long userId) {
-        // 2026-06-10 e2e 二轮实证（P0-1 补口）：匿名上报是多级联写（事件 insert → stat 原子累加 → feed rank 回灌
-        //   → markAggregated UPDATE），其中 markAggregated 的 updateById 运行时实证仍把 updater=null 写进
-        //   UPDATE SET（与 M-b 三因叠加同形态）。逐写点 set 兜底易漏级联（feed 模块回灌同线程同雷），故在入口
-        //   对匿名请求注入系统身份（照 AigcGenerateExecutor.executeWithSystemIdentity 范本的 Web 线程版），
-        //   使全链 DefaultDBFieldHandler 填充 creator/updater="0"；finally 清理防 Web 线程上下文泄漏。
-        boolean injected = injectSystemIdentityIfAnonymous(userId);
-        try {
-            // 批量上限：超限整体拒绝（统一错误码出口，前端据此分批重传）
-            if (reqVO.getBatch().size() > MAX_BATCH_SIZE) {
-                throw exception(TELEMETRY_BATCH_SIZE_EXCEEDED);
-            }
-            // 本次批量受理回执 traceId（便于排障；优先复用首条信封 traceId，缺失则生成）
-            String batchTraceId = resolveBatchTraceId(reqVO);
-
-            int accepted = 0;
-            int rejected = 0;
-            // 部分成功语义：逐条轻量校验，通过则同步写（落库+聚合+回灌），失败计 rejected（不中断整批）
-            for (EnvelopeReqVO envelope : reqVO.getBatch()) {
-                if (!isEnvelopeValid(envelope)) {
-                    rejected++;
-                    continue;
-                }
-                ingestOne(envelope, userId);
-                accepted++;
-            }
-            // 关键链路日志：受理成功记 INFO，含回执 traceId 与计数，便于全链路排障
-            log.info("[ingestBatch] 批量同步写完成 batchTraceId={}, accepted={}, rejected={}", batchTraceId, accepted, rejected);
-
-            EventBatchResultVO result = new EventBatchResultVO();
-            result.setAccepted(accepted);
-            result.setRejected(rejected);
-            result.setTraceId(batchTraceId);
-            return result;
-        } finally {
-            if (injected) {
-                SecurityContextHolder.clearContext();
-            }
+        // 批量上限：超限整体拒绝（统一错误码出口，前端据此分批重传）——先判，任何投递/落库前
+        if (reqVO.getBatch().size() > MAX_BATCH_SIZE) {
+            throw exception(TELEMETRY_BATCH_SIZE_EXCEEDED);
         }
+        TelemetryEventProducer producer = resolveProducer();
+        if (producer != null) {
+            // MQ 异步路径：逐条校验 + 投递 event tag，快速 ACK 返回 accepted/rejected（不在请求线程做聚合）
+            return dispatchBatchToMq(reqVO, userId, producer);
+        }
+        // 同步回落（MQ 关闭 / 单测）：逐条单事务落库 + 聚合 + 回灌，行为与改造前一致
+        return ingestBatchSyncFallback(reqVO, userId);
     }
 
+    /**
+     * 兜底单条上报（切片一 阶段〇 R1 路由）：校验失败恒返 true（不阻塞前端卸载）；校验通过 MQ 开启则投递 perf tag、
+     * 关闭则同步写。fire-and-forget，任何路径恒返 true。
+     */
     @Override
-    @Transactional(rollbackFor = Exception.class) // 兜底通道单条同步写：与 ingestBatch 同口径（落库+聚合+回灌原子）
     public Boolean ingestBeacon(EnvelopeReqVO reqVO, Long userId) {
-        // 兜底通道：页面卸载场景，校验失败不抛错（返回受理成功，不触发前端异常）；
-        //   校验通过则同步写——若同步写抛错（极少：DB 故障），由全局异常处理返回错误响应，
-        //   但 sendBeacon 为 fire-and-forget 不读响应、不阻塞卸载（事务该回滚仍回滚，保证不脏写）。
+        // 兜底通道：页面卸载场景，校验失败不抛错（返回受理成功，不触发前端异常）
         if (!isEnvelopeValid(reqVO)) {
             log.warn("[ingestBeacon] 信封校验未通过，丢弃 eventId={}, event={}, traceId={}",
                     reqVO == null ? null : reqVO.getEventId(),
@@ -161,11 +156,30 @@ public class EventIngestServiceImpl implements EventIngestService {
                     reqVO == null ? null : reqVO.getTraceId());
             return Boolean.TRUE;
         }
-        // 匿名注入系统身份：与 ingestBatch 同口径（级联写审计列兜底），finally 清理
+        TelemetryEventProducer producer = resolveProducer();
+        if (producer != null) {
+            // MQ 异步：投递 perf tag；fire-and-forget 不阻塞卸载，投递失败由前端 sendBeacon 天然不重试（性能埋点可容忍丢失）
+            producer.publish(reqVO, userId, TelemetryEventProducer.TAG_PERF);
+            return Boolean.TRUE;
+        }
+        // 同步回落：单条事务落库 + 聚合 + 回灌（与 ingestBatch 同口径）
+        invokeConsumeFromMq(reqVO, userId);
+        return Boolean.TRUE;
+    }
+
+    /**
+     * MQ 消费侧单条摄取（切片一 阶段〇 R1）：消费线程或同步回落逐条入口——单条 {@code @Transactional} 落库 + 聚合 + 回灌，
+     * 逐字复用同步写核心 {@link #ingestOne}（聚合/回灌行为与改造前一致）；匿名注入系统身份补审计列，finally 清理。
+     * <p>2026-06-10 e2e P0-1 补口沿用：匿名上报多级联写（事件 insert / stat 原子累加 / feed rank 回灌 / markAggregated UPDATE）
+     *   的 DefaultDBFieldHandler 审计列填充统一取 "0"；消费线程无 Web 请求上下文，ServletUtils.getClientIP() 返 null
+     *   由异常剔除桩按仅 anonId 维度处理（本就 MQ-ready）。
+     */
+    @Override
+    @Transactional(rollbackFor = Exception.class) // 单条同步写：落库+聚合+回灌+置已聚合在同一本地事务，任一步失败该条回滚（幂等保证重试不双计）
+    public void consumeFromMq(EnvelopeReqVO reqVO, Long userId) {
         boolean injected = injectSystemIdentityIfAnonymous(userId);
         try {
             ingestOne(reqVO, userId);
-            return Boolean.TRUE;
         } finally {
             if (injected) {
                 SecurityContextHolder.clearContext();
@@ -173,6 +187,71 @@ public class EventIngestServiceImpl implements EventIngestService {
         }
     }
 
+    // ---------- 路由内部实现 ----------
+
+    /** 解析 MQ 生产者（软注入 + 空值守卫）：Provider 缺席/未注入（enabled=false 或单测）→ null → 回落同步写。 */
+    private TelemetryEventProducer resolveProducer() {
+        return telemetryEventProducerProvider == null ? null : telemetryEventProducerProvider.getIfAvailable();
+    }
+
+    /**
+     * MQ 异步路径批量投递：逐条轻量校验，通过则投递 event tag（成功计 accepted、投递失败计 rejected 交前端重传），
+     * 校验失败计 rejected。返回 accepted/rejected + 回执 traceId（部分成功语义、非整批失败）。
+     */
+    private EventBatchResultVO dispatchBatchToMq(EventBatchReqVO reqVO, Long userId, TelemetryEventProducer producer) {
+        String batchTraceId = resolveBatchTraceId(reqVO);
+        int accepted = 0;
+        int rejected = 0;
+        for (EnvelopeReqVO envelope : reqVO.getBatch()) {
+            if (!isEnvelopeValid(envelope)) {
+                rejected++;
+                continue;
+            }
+            // 投递成功=已受理(accepted)；投递失败=按受理失败(rejected)，前端分批重传兜底（幂等由 eventId 保证不双计）
+            if (producer.publish(envelope, userId, TelemetryEventProducer.TAG_EVENT)) {
+                accepted++;
+            } else {
+                rejected++;
+            }
+        }
+        log.info("[ingestBatch] 批量投递 MQ 完成 batchTraceId={}, accepted={}, rejected={}", batchTraceId, accepted, rejected);
+        EventBatchResultVO result = new EventBatchResultVO();
+        result.setAccepted(accepted);
+        result.setRejected(rejected);
+        result.setTraceId(batchTraceId);
+        return result;
+    }
+
+    /**
+     * 同步回落批量摄取（MQ 关闭 / 单测）：逐条轻量校验，通过则经代理调 {@link #consumeFromMq}（单条事务），失败计 rejected。
+     * 与改造前的差异仅事务粒度（整批→逐条，更贴契约「非整批失败」语义、坏事件不连累好事件）；聚合/回灌逻辑逐字不变。
+     */
+    private EventBatchResultVO ingestBatchSyncFallback(EventBatchReqVO reqVO, Long userId) {
+        String batchTraceId = resolveBatchTraceId(reqVO);
+        int accepted = 0;
+        int rejected = 0;
+        for (EnvelopeReqVO envelope : reqVO.getBatch()) {
+            if (!isEnvelopeValid(envelope)) {
+                rejected++;
+                continue;
+            }
+            invokeConsumeFromMq(envelope, userId);
+            accepted++;
+        }
+        log.info("[ingestBatch] 批量同步写完成（MQ 关闭回落）batchTraceId={}, accepted={}, rejected={}", batchTraceId, accepted, rejected);
+        EventBatchResultVO result = new EventBatchResultVO();
+        result.setAccepted(accepted);
+        result.setRejected(rejected);
+        result.setTraceId(batchTraceId);
+        return result;
+    }
+
+    /** 经代理调 {@link #consumeFromMq}（保证单条 {@code @Transactional} 真生效）；单测无代理时空值守卫直调 this。 */
+    private void invokeConsumeFromMq(EnvelopeReqVO envelope, Long userId) {
+        EventIngestService target = (self != null) ? self : this;
+        target.consumeFromMq(envelope, userId);
+    }
+
     /**
      * 匿名上报注入系统身份（2026-06-10 e2e 二轮 P0-1 补口，范本=AigcGenerateExecutor.executeWithSystemIdentity）。
      *

game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/mq/TelemetryMqRoundTripIT.java

@@ -0,0 +1,140 @@
+package com.wanxiang.huijing.game.module.telemetry.mq;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.game.module.telemetry.mq.consumer.TelemetryEventConsumer;
+import com.wanxiang.huijing.game.module.telemetry.mq.message.TelemetryEventMessage;
+import com.wanxiang.huijing.game.module.telemetry.mq.producer.TelemetryEventProducer;
+import com.wanxiang.huijing.game.module.telemetry.service.event.EventIngestService;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIfSystemProperty;
+import org.mockito.ArgumentCaptor;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.verify;
+
+/**
+ * R1「MQ 消费聚合真跑」集成证据——连 mini-infra 真 RocketMQ（{@code 100.64.0.8:9876} · Tailscale 直连、绕系统代理）
+ * 跑一次 produce → broker → consume → 转调 {@code consumeFromMq} 全链路，坐实生产者/消费者/消息序列化对真 broker 成立。
+ *
+ * <p><b>覆盖边界</b>：真 broker 的投递/消费/往返序列化 + 两个生产类（{@link TelemetryEventProducer} /
+ *   {@link TelemetryEventConsumer}）在真 broker 上的行为。聚合落库（consumeFromMq→ingestOne→game_telemetry_game_stat）
+ *   的逻辑由 {@code EventIngestServiceImplMqPathTest#testConsumeFromMq_reusesIngestOne_aggregatesAndBackfills} 单测坐实
+ *   （逐字复用 ingestOne，不重造）；本 IT 用 mock 的 EventIngestService 捕获转调，不连 MySQL。整条
+ *   HTTP POST→MQ→真库聚合行 的单体 e2e 需后端窗口（MySQL/Redis/Nacos 全栈），与阶段〇 C1/C2/C3 集成验证同window处置。
+ *
+ * <p><b>默认跳过</b>：{@code @EnabledIfSystemProperty(telemetry.mq.e2e=1)}——常规 {@code mvn test} 不连 broker、不跑本 IT；
+ *   出证据时显式 {@code mvn test -Dtelemetry.mq.e2e=1 -Dtest=TelemetryMqRoundTripIT}。
+ *
+ * @author 绘境AI
+ */
+@EnabledIfSystemProperty(named = "telemetry.mq.e2e", matches = "1")
+class TelemetryMqRoundTripIT {
+
+    /** mini-infra NameServer（Tailscale 直连；无系统代理 env，直连不被 fake-ip 拦）。 */
+    private static final String NAMESRV = System.getProperty("telemetry.mq.namesrv", "100.64.0.8:9876");
+
+    @Test
+    void produce_consume_roundTrip_againstRealBroker() throws Exception {
+        String eventId = "e2e-" + UUID.randomUUID();
+        String traceId = "trace-" + UUID.randomUUID();
+        // 独立测试消费组，earliest 起消费防错过刚发的消息
+        String group = "telemetry-event-consumer-e2e-" + System.currentTimeMillis();
+        BlockingQueue<byte[]> received = new ArrayBlockingQueue<>(64);
+        ObjectMapper om = new ObjectMapper();
+
+        // ── 真 push consumer：订阅 telemetry-event，收到即把 body 塞队列 ──
+        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
+        consumer.setNamesrvAddr(NAMESRV);
+        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
+        consumer.subscribe(TelemetryEventProducer.TOPIC, TelemetryEventProducer.TAG_EVENT);
+        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, ctx) -> {
+            msgs.forEach(m -> received.offer(m.getBody()));
+            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+        });
+        consumer.start();
+        // 新消费组首启需等首次 rebalance + 路由拉取，并 establish CONSUME_FROM_LAST_OFFSET 的尾偏移；
+        //   在此之前投递会被「跳到尾」而漏收（新组 offset 竞态）。等 8s 让其 settle 后再投。
+        Thread.sleep(8000);
+
+        // ── 真 producer：构造 RocketMQTemplate（rocketmq-client DefaultMQProducer）→ 真 TelemetryEventProducer ──
+        org.apache.rocketmq.client.producer.DefaultMQProducer mqProducer =
+                new org.apache.rocketmq.client.producer.DefaultMQProducer("telemetry-event-producer-e2e");
+        mqProducer.setNamesrvAddr(NAMESRV);
+        mqProducer.start();
+        RocketMQTemplate template = new RocketMQTemplate();
+        template.setProducer(mqProducer);
+        template.setMessageConverter(new RocketMQMessageConverter().getMessageConverter());
+        TelemetryEventProducer producer = new TelemetryEventProducer(template);
+
+        try {
+            EnvelopeReqVO env = new EnvelopeReqVO();
+            env.setEventId(eventId);
+            env.setEvent("game_play_end");
+            env.setSchemaVersion("v1");
+            env.setTs(System.currentTimeMillis());
+            env.setTraceId(traceId);
+            EnvelopeReqVO.ContextVO ctx = new EnvelopeReqVO.ContextVO();
+            ctx.setGameId("1024");
+            ctx.setVersionId("2048");
+            env.setContext(ctx);
+
+            // 1) 真投递到真 broker（返 true 即 broker 已确认 SEND_OK）
+            boolean sent = producer.publish(env, 99L, TelemetryEventProducer.TAG_EVENT);
+            assertTrue(sent, "真 broker 应确认 SEND_OK");
+
+            // 2) 真 broker 回投 → 扫描出我这条（按 eventId 匹配，容忍 topic 上的其它并发流量）；≤20s
+            TelemetryEventMessage roundTrip = null;
+            long deadline = System.currentTimeMillis() + 20000;
+            while (System.currentTimeMillis() < deadline) {
+                byte[] body = received.poll(2, TimeUnit.SECONDS);
+                if (body == null) {
+                    continue;
+                }
+                TelemetryEventMessage m = om.readValue(new String(body, StandardCharsets.UTF_8), TelemetryEventMessage.class);
+                if (m.getEnvelope() != null && eventId.equals(m.getEnvelope().getEventId())) {
+                    roundTrip = m;
+                    break;
+                }
+            }
+            assertNotNull(roundTrip, "应从真 broker 消费到刚投递的消息（按 eventId 匹配）");
+
+            // 3) 往返序列化保真：信封逐字段一致
+            assertNotNull(roundTrip.getEnvelope(), "消息信封应往返完整");
+            assertEquals(eventId, roundTrip.getEnvelope().getEventId(), "eventId 往返一致（幂等真身）");
+            assertEquals("game_play_end", roundTrip.getEnvelope().getEvent());
+            assertEquals("1024", roundTrip.getEnvelope().getContext().getGameId(), "gameId 往返一致（聚合落点）");
+            assertEquals(99L, roundTrip.getUserId(), "userId 随消息携带、往返一致");
+
+            // 4) 真消费者类处理真 broker 消息 → 转调 consumeFromMq（聚合入口，逐字复用 ingestOne）
+            TelemetryEventConsumer realConsumer = new TelemetryEventConsumer();
+            EventIngestService ingestMock = mock(EventIngestService.class);
+            ReflectionTestUtils.setField(realConsumer, "eventIngestService", ingestMock);
+            realConsumer.onMessage(roundTrip);
+            ArgumentCaptor<EnvelopeReqVO> envCaptor = ArgumentCaptor.forClass(EnvelopeReqVO.class);
+            verify(ingestMock, timeout(2000)).consumeFromMq(envCaptor.capture(), eq(99L));
+            assertEquals(eventId, envCaptor.getValue().getEventId(), "消费者应把真 broker 信封转调聚合入口");
+        } finally {
+            consumer.shutdown();
+            mqProducer.shutdown();
+        }
+    }
+}

game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/mq/consumer/TelemetryEventConsumerTest.java

@@ -0,0 +1,81 @@
+package com.wanxiang.huijing.game.module.telemetry.mq.consumer;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.game.module.telemetry.mq.message.TelemetryEventMessage;
+import com.wanxiang.huijing.game.module.telemetry.service.event.EventIngestService;
+import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * {@link TelemetryEventConsumer} 单元测试（纯 Mockito，不触真 MQ）——把守 R1 消费者转调 + 毒消息守卫 + 重试语义。
+ *
+ * <p>覆盖：① 正常消费转调 {@code consumeFromMq(envelope,userId)}；② 毒消息（消息/信封为空）吞掉不转调、不外抛（ack 防无限重投）；
+ *   ③ 瞬时故障（{@code consumeFromMq} 抛异常）外抛让 RocketMQ 重投（≤16 次转 DLQ，幂等保证不双计）。
+ *
+ * @author 绘境AI
+ */
+class TelemetryEventConsumerTest extends BaseMockitoUnitTest {
+
+    @Mock
+    private EventIngestService eventIngestService;
+
+    @InjectMocks
+    private TelemetryEventConsumer consumer;
+
+    private static EnvelopeReqVO envelope() {
+        EnvelopeReqVO env = new EnvelopeReqVO();
+        env.setEventId("evt-1");
+        env.setEvent("game_play_end");
+        env.setSchemaVersion("v1");
+        env.setTs(1733600000000L);
+        env.setTraceId("trace-abc");
+        return env;
+    }
+
+    /** 正常路：消费转调 consumeFromMq(envelope, userId)，逐字复用同步写核心。 */
+    @Test
+    void testOnMessage_dispatchesToConsumeFromMq() {
+        EnvelopeReqVO env = envelope();
+        TelemetryEventMessage msg = new TelemetryEventMessage(env, 99L);
+
+        assertDoesNotThrow(() -> consumer.onMessage(msg));
+
+        verify(eventIngestService).consumeFromMq(eq(env), eq(99L));
+    }
+
+    /** 毒消息守卫①：消息为 null → 吞掉、不转调、不外抛（ack 防无限重投）。 */
+    @Test
+    void testOnMessage_nullMessage_swallowedNoDispatch() {
+        assertDoesNotThrow(() -> consumer.onMessage(null));
+        verify(eventIngestService, never()).consumeFromMq(any(), any());
+    }
+
+    /** 毒消息守卫②：信封为 null → 吞掉、不转调、不外抛。 */
+    @Test
+    void testOnMessage_nullEnvelope_swallowedNoDispatch() {
+        TelemetryEventMessage msg = new TelemetryEventMessage(null, 99L);
+        assertDoesNotThrow(() -> consumer.onMessage(msg));
+        verify(eventIngestService, never()).consumeFromMq(any(), any());
+    }
+
+    /** 瞬时故障：consumeFromMq 抛异常 → onMessage 外抛（触发 RocketMQ 重投；幂等保证重试不双计）。 */
+    @Test
+    void testOnMessage_transientFailure_rethrowsForRetry() {
+        EnvelopeReqVO env = envelope();
+        TelemetryEventMessage msg = new TelemetryEventMessage(env, null);
+        doThrow(new RuntimeException("DB down")).when(eventIngestService).consumeFromMq(eq(env), isNull());
+
+        assertThrows(RuntimeException.class, () -> consumer.onMessage(msg));
+    }
+}

game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/mq/producer/TelemetryEventProducerTest.java

@@ -0,0 +1,94 @@
+package com.wanxiang.huijing.game.module.telemetry.mq.producer;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest;
+import org.apache.rocketmq.client.producer.SendResult;
+import org.apache.rocketmq.client.producer.SendStatus;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.springframework.messaging.Message;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@link TelemetryEventProducer} 单元测试（纯 Mockito，不触真 MQ）——把守 R1 生产者投递契约与外部交互纪律。
+ *
+ * <p>覆盖：① 正常投递到 {@code telemetry-event:event}、返 true；② SEND_OK 判定；③ syncSend 抛异常时
+ *   {@code publish} <b>不外抛</b>、返 false（交调用方按 rejected 处置，前端重传兜底，绝不 500 阻断上报）；
+ *   ④ 非 SEND_OK 返 false；⑤ keys 置 traceId。
+ *
+ * @author 绘境AI
+ */
+class TelemetryEventProducerTest extends BaseMockitoUnitTest {
+
+    @Mock
+    private RocketMQTemplate rocketMQTemplate;
+
+    @InjectMocks
+    private TelemetryEventProducer producer;
+
+    /** 构造一条最简合法信封（含 eventId/traceId）。 */
+    private static EnvelopeReqVO envelope() {
+        EnvelopeReqVO env = new EnvelopeReqVO();
+        env.setEventId("evt-1");
+        env.setEvent("game_play_start");
+        env.setSchemaVersion("v1");
+        env.setTs(1733600000000L);
+        env.setTraceId("trace-abc");
+        return env;
+    }
+
+    /** 正常路：投递到 topic:tag=telemetry-event:event，SEND_OK → 返 true，keys=traceId。 */
+    @Test
+    @SuppressWarnings("unchecked")
+    void testPublish_sendOk_returnsTrueAndKeysTraceId() {
+        SendResult ok = new SendResult();
+        ok.setSendStatus(SendStatus.SEND_OK);
+        ok.setMsgId("mid-1");
+        when(rocketMQTemplate.syncSend(eq(TelemetryEventProducer.TOPIC + ":" + TelemetryEventProducer.TAG_EVENT),
+                (Message<?>) org.mockito.ArgumentMatchers.any(), anyLong())).thenReturn(ok);
+
+        boolean sent = producer.publish(envelope(), 99L, TelemetryEventProducer.TAG_EVENT);
+
+        assertTrue(sent, "SEND_OK 应返回 true");
+        // 校验投递目的地 + keys header = traceId
+        ArgumentCaptor<Message<?>> msgCaptor = ArgumentCaptor.forClass(Message.class);
+        verify(rocketMQTemplate).syncSend(eq("telemetry-event:event"), msgCaptor.capture(), anyLong());
+        Object keys = msgCaptor.getValue().getHeaders().get("KEYS");
+        assertTrue("trace-abc".equals(keys), "MQ keys 应置 traceId 便于对账");
+    }
+
+    /** best-effort（核心断言）：syncSend 抛异常 → publish 吞异常不外抛、返 false（前端重传兜底，不 500 阻断上报）。 */
+    @Test
+    void testPublish_bestEffort_swallowsExceptionReturnsFalse() {
+        doThrow(new RuntimeException("MQ down")).when(rocketMQTemplate)
+                .syncSend(org.mockito.ArgumentMatchers.anyString(),
+                        (Message<?>) org.mockito.ArgumentMatchers.any(), anyLong());
+
+        boolean sent = assertDoesNotThrow(() -> producer.publish(envelope(), null, TelemetryEventProducer.TAG_EVENT));
+        assertFalse(sent, "发送异常应返回 false（按 rejected 处置）");
+    }
+
+    /** 非 SEND_OK（如刷盘超时）→ 返 false（不外抛）。 */
+    @Test
+    @SuppressWarnings("unchecked")
+    void testPublish_nonSendOk_returnsFalse() {
+        SendResult flushTimeout = new SendResult();
+        flushTimeout.setSendStatus(SendStatus.FLUSH_DISK_TIMEOUT);
+        when(rocketMQTemplate.syncSend(org.mockito.ArgumentMatchers.anyString(),
+                (Message<?>) org.mockito.ArgumentMatchers.any(), anyLong())).thenReturn(flushTimeout);
+
+        boolean sent = producer.publish(envelope(), 1L, TelemetryEventProducer.TAG_PERF);
+        assertFalse(sent, "非 SEND_OK 应返回 false");
+    }
+}

game-cloud/game-module-telemetry/game-module-telemetry-server/src/test/java/com/wanxiang/huijing/game/module/telemetry/service/event/EventIngestServiceImplMqPathTest.java

@@ -0,0 +1,188 @@
+package com.wanxiang.huijing.game.module.telemetry.service.event;
+
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EnvelopeReqVO;
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EventBatchReqVO;
+import com.wanxiang.huijing.game.module.telemetry.controller.app.event.vo.EventBatchResultVO;
+import com.wanxiang.huijing.game.module.telemetry.dal.dataobject.stat.GameStatDO;
+import com.wanxiang.huijing.game.module.telemetry.dal.mysql.event.TelemetryEventMapper;
+import com.wanxiang.huijing.game.module.telemetry.dal.mysql.stat.GameStatMapper;
+import com.wanxiang.huijing.game.module.telemetry.mq.producer.TelemetryEventProducer;
+import com.wanxiang.huijing.game.module.telemetry.service.quality.AnomalyFilterService;
+import com.wanxiang.huijing.game.module.feed.api.FeedApi;
+import com.wanxiang.huijing.game.module.feed.dto.FeedRankUpsertReqDTO;
+import com.wanxiang.huijing.framework.test.core.ut.BaseMockitoUnitTest;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.springframework.beans.factory.ObjectProvider;
+
+import java.time.LocalDate;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * {@link EventIngestServiceImpl} MQ 异步路径单测（切片一 阶段〇 R1）——把守「生产者在席=投递不同步写」路由
+ * 与「消费入口 consumeFromMq 逐字复用 ingestOne 聚合/回灌」。与 {@code EventIngestServiceImplTest}（同步回落路径、
+ * 既有断言逐字不变）互补，共同坐实「聚合行为与 quality_score 回灌行为不变、仅执行位置从请求线程搬到消费线程」。
+ *
+ * @author 绘境AI
+ */
+class EventIngestServiceImplMqPathTest extends BaseMockitoUnitTest {
+
+    @InjectMocks
+    private EventIngestServiceImpl eventIngestService;
+
+    @Mock
+    private TelemetryEventMapper telemetryEventMapper;
+    @Mock
+    private GameStatMapper gameStatMapper;
+    @Mock
+    private FeedApi feedApi;
+    @Mock
+    private AnomalyFilterService anomalyFilterService;
+    @Mock
+    private ObjectProvider<TelemetryEventProducer> telemetryEventProducerProvider; // MQ 生产者软注入
+    @Mock
+    private TelemetryEventProducer telemetryEventProducer;
+
+    /** MQ 开启：ingestBatch 逐条投递、不在请求线程做同步写（DB/feed 零交互）。 */
+    @Test
+    void testIngestBatch_mqEnabled_producesNotSync() {
+        when(telemetryEventProducerProvider.getIfAvailable()).thenReturn(telemetryEventProducer);
+        when(telemetryEventProducer.publish(any(), any(), eq(TelemetryEventProducer.TAG_EVENT))).thenReturn(true);
+
+        EventBatchReqVO reqVO = new EventBatchReqVO();
+        List<EnvelopeReqVO> batch = new ArrayList<>();
+        batch.add(validEnvelope("game_play_start", "t1"));    // 合法 → 投递
+        batch.add(validEnvelope("like", "t2"));               // 合法 → 投递
+        batch.add(validEnvelope("not_registered_evt", "t3")); // 未登记 → rejected（不投递）
+        reqVO.setBatch(batch);
+
+        EventBatchResultVO result = eventIngestService.ingestBatch(reqVO, 99L);
+
+        assertEquals(2, result.getAccepted());
+        assertEquals(1, result.getRejected());
+        assertEquals("t1", result.getTraceId());
+        // 逐条投递 event tag 两次
+        verify(telemetryEventProducer, times(2)).publish(any(), eq(99L), eq(TelemetryEventProducer.TAG_EVENT));
+        // 关键：请求线程内不做同步写（落库/聚合/回灌全在消费侧）
+        verifyNoInteractions(telemetryEventMapper);
+        verifyNoInteractions(gameStatMapper);
+        verifyNoInteractions(feedApi);
+    }
+
+    /** MQ 开启但投递失败：该条计 rejected（前端重传兜底），不误报 accepted。 */
+    @Test
+    void testIngestBatch_mqEnabled_publishFailCountedRejected() {
+        when(telemetryEventProducerProvider.getIfAvailable()).thenReturn(telemetryEventProducer);
+        when(telemetryEventProducer.publish(any(), any(), anyString())).thenReturn(false); // 投递恒失败
+
+        EventBatchReqVO reqVO = new EventBatchReqVO();
+        List<EnvelopeReqVO> batch = new ArrayList<>();
+        batch.add(validEnvelope("game_play_start", "t1"));
+        reqVO.setBatch(batch);
+
+        EventBatchResultVO result = eventIngestService.ingestBatch(reqVO, 99L);
+
+        assertEquals(0, result.getAccepted());
+        assertEquals(1, result.getRejected(), "投递失败应计 rejected 交前端重传");
+    }
+
+    /** MQ 开启：beacon 投递 perf tag、恒返 true、不同步写。 */
+    @Test
+    void testIngestBeacon_mqEnabled_producesPerfTag() {
+        when(telemetryEventProducerProvider.getIfAvailable()).thenReturn(telemetryEventProducer);
+        when(telemetryEventProducer.publish(any(), any(), eq(TelemetryEventProducer.TAG_PERF))).thenReturn(true);
+
+        Boolean ok = eventIngestService.ingestBeacon(validEnvelope("perf_first_screen", "t-beacon"), 99L);
+
+        assertTrue(ok);
+        verify(telemetryEventProducer).publish(any(), eq(99L), eq(TelemetryEventProducer.TAG_PERF));
+        verifyNoInteractions(telemetryEventMapper);
+    }
+
+    /**
+     * 消费入口 consumeFromMq 逐字复用 ingestOne：game_play_end(completed) → 原子累加 → 重读算分(60.00) → 回灌 feed。
+     * 坐实「聚合/回灌行为不变」——与 EventIngestServiceImplTest#testIngestOne_playEndTriggersQualityRefresh 同口径，
+     * 只是入口从 ingestBatch 换成 consumeFromMq（消费线程入口）。
+     */
+    @Test
+    void testConsumeFromMq_reusesIngestOne_aggregatesAndBackfills() {
+        Map<String, Object> props = new HashMap<>();
+        props.put("completed", true);
+        props.put("duration_ms", 5000);
+        EnvelopeReqVO env = gameEnvelope("game_play_end", "t1", "1024", "2048", props);
+        when(telemetryEventMapper.selectByEventId(env.getEventId())).thenReturn(null);
+        GameStatDO accumulated = new GameStatDO();
+        accumulated.setId(7L);
+        accumulated.setPlayCount(0L);
+        accumulated.setPlayEndCount(1L);
+        accumulated.setCompletedCount(1L);
+        accumulated.setTotalDurationMs(5000L);
+        accumulated.setLikeCount(0L);
+        accumulated.setShareCount(0L);
+        when(gameStatMapper.selectByGameAndDate(eq(1024L), any(LocalDate.class))).thenReturn(accumulated);
+
+        eventIngestService.consumeFromMq(env, 99L);
+
+        // 原子累加（play_end=1, completed=1, duration=5000）
+        verify(gameStatMapper).insertOrAccumulate(eq(1024L), any(LocalDate.class),
+                eq(0L), eq(1L), eq(1L), eq(5000L), eq(0L), eq(0L));
+        // 算分写回 + 回灌 feed（QUALITY_REFRESH，60.0）
+        verify(gameStatMapper).updateById(any(GameStatDO.class));
+        verify(feedApi).upsertRank(any(FeedRankUpsertReqDTO.class));
+        // 幂等预查发生（重放同 eventId 不双计的第一道网）
+        verify(telemetryEventMapper).selectByEventId(env.getEventId());
+    }
+
+    /** consumeFromMq 幂等：同 eventId 已落库 → 跳过累加/回灌（重复投递不双计）。 */
+    @Test
+    void testConsumeFromMq_idempotentSkipOnDuplicate() {
+        EnvelopeReqVO env = gameEnvelope("game_play_start", "t1", "1024", "2048", null);
+        when(telemetryEventMapper.selectByEventId(env.getEventId()))
+                .thenReturn(new com.wanxiang.huijing.game.module.telemetry.dal.dataobject.event.TelemetryEventDO());
+
+        eventIngestService.consumeFromMq(env, 99L);
+
+        verify(telemetryEventMapper, never()).insert(any(com.wanxiang.huijing.game.module.telemetry.dal.dataobject.event.TelemetryEventDO.class));
+        verifyNoInteractions(gameStatMapper);
+        verifyNoInteractions(feedApi);
+    }
+
+    // ============================== 测试夹具 ==============================
+
+    private static EnvelopeReqVO validEnvelope(String event, String traceId) {
+        EnvelopeReqVO envelope = new EnvelopeReqVO();
+        envelope.setEventId(UUID.randomUUID().toString());
+        envelope.setEvent(event);
+        envelope.setSchemaVersion("v1");
+        envelope.setTs(1733600000000L);
+        envelope.setTraceId(traceId);
+        return envelope;
+    }
+
+    private static EnvelopeReqVO gameEnvelope(String event, String traceId, String gameId, String versionId, Map<String, Object> props) {
+        EnvelopeReqVO envelope = validEnvelope(event, traceId);
+        EnvelopeReqVO.ContextVO ctx = new EnvelopeReqVO.ContextVO();
+        ctx.setGameId(gameId);
+        ctx.setVersionId(versionId);
+        envelope.setContext(ctx);
+        envelope.setProps(props);
+        return envelope;
+    }
+}

game-cloud/huijing-server/src/main/resources/application.yaml

@@ -190,6 +190,13 @@ rocketmq:
   producer:
     group: ${spring.application.name}_PRODUCER # 生产者分组
 
+# telemetry 遥测异步（切片一 阶段〇 R1）：批量事件入口改 MQ 异步消费的灰度开关
+telemetry:
+  mq:
+    # true=上报入口快速 ACK + 投递 telemetry-event 主题、消费侧异步聚合（幂等）；
+    # false（默认）=生产者/消费者 Bean 缺席、回落原请求线程内同步写（现行行为逐字不变）。经 .env 注入按环境开启。
+    enabled: ${TELEMETRY_MQ_ENABLED:false}
+
 spring:
   # Kafka 配置项，对应 KafkaProperties 配置类
   kafka: