本文详解如何在 Quarkus WebSocket 服务中，于 @OnMessage 异步处理及 Vert.x EventBus 事件消费阶段，可靠地透传 MDC（Mapped Diagnostic Context）日志上下文，确保 user.id、websocket.sessionId 等关键追踪字段全程可用。

本文详解如何在 quarkus websocket 服务中，于 `@onmessage` 异步处理及 vert.x eventbus 事件消费阶段，可靠地透传 mdc（mapped diagnostic context）日志上下文，确保 `user.id`、`websocket.sessionid` 等关键追踪字段全程可用。

在 Quarkus 的 WebSocket 实现中，MDC（如 Logback 的 MDC）本质上基于 ThreadLocal，其生命周期严格绑定于当前线程。而 WebSocket 的 @OnMessage 回调虽由容器线程触发，但一旦消息被投递至 Vert.x EventBus（如 vertx.eventBus().send()），后续的 @ConsumeEvent 处理将运行在 Vert.x 工作线程池中——此时原始线程的 MDC 上下文已丢失，导致 MDC.get("user.id") 返回 null。

要解决这一问题，核心思路是：主动捕获、显式存储、按需恢复。不能依赖自动传播（Quarkus 未为 WebSocket + EventBus 组合提供开箱即用的 MDC 跨线程透传），而应将 MDC 上下文快照与 WebSocket 会话强绑定，并在异步处理入口处手动还原。

✅ 推荐实现方案：基于 Session ID 的 MDC 快照管理

我们采用静态线程安全映射（ConcurrentHashMap）持久化每个活跃 WebSocket 会话的 MDC 快照，并在关键生命周期节点进行存取操作：

Quark Lazy Cli

基于夸克网盘自动转存项目 QAS 的懒人追剧、追番 SKILL，内置 CLI 工具。即使 QAS 订阅失效，仍可重新搜索资源、遍历多级目录，在多个最新候选中按用户偏好（如更高清晰度）转存新集和缺失集。可交由本地 OpenClaw、Hermes Agent 或其他 Claude 客户端使用。

下载

// UserWebSocketController.java
@Slf4j
@ApplicationScoped
@ServerEndpoint(value = "/users/{userId}")
public class UserWebSocketController {

    // ⚠️ 线程安全的 MDC 快照存储（Key: session.id）
    private static final Map<String, Map<String, String>> SESSION_MDC_MAP =
            new ConcurrentHashMap<>();

    private final WebsocketConnectionService websocketConnectionService;
    private final Vertx vertx;

    public UserWebSocketController(WebsocketConnectionService websocketConnectionService, Vertx vertx) {
        this.websocketConnectionService = websocketConnectionService;
        this.vertx = vertx;
    }

    @OnOpen
    public void onOpen(Session session, @PathParam("userId") String userId) {
        // 1. 初始化 MDC
        MDC.put("websocket.sessionId", session.getId());
        MDC.put("user.id", userId);
        log.info("New WebSocket Session opened for user {}", userId);

        // 2. 持久化当前 MDC 快照（供后续异步使用）
        storeSessionMDC(session.getId());

        websocketConnectionService.addConnection(userId, session);
    }

    @OnMessage
    public void onMessage(Session session, String message, @PathParam("userId") String userId) {
        // 3. 关键：恢复当前会话的 MDC（保障 onMessage 内日志可追溯）
        restoreSessionMDC(session.getId());
        log.info("New message received: {}", message);

        // 4. 将 sessionId 显式携带至事件负载，用于下游还原
        var asyncMsg = new AsyncWebSocketMessage(session.getId(), message);
        vertx.eventBus().send("websocket.message.new", asyncMsg);
    }

    @OnClose
    public void onClose(Session session, @PathParam("userId") String userId) {
        restoreSessionMDC(session.getId());
        log.info("WebSocket Session closed for user {}", userId);
        removeSessionMDC(session.getId()); // 清理资源
        websocketConnectionService.removeSession(userId);
    }

    @OnError
    public void onError(Session session, @PathParam("userId") String userId, Throwable throwable) {
        restoreSessionMDC(session.getId());
        log.error("Error in WebSocket session for user {}", userId, throwable);
        removeSessionMDC(session.getId());
        websocketConnectionService.removeSession(userId);
    }

    // —— MDC 快照工具方法 ——
    private static void storeSessionMDC(String sessionId) {
        SESSION_MDC_MAP.put(sessionId, MDC.getCopyOfContextMap());
    }

    public static void restoreSessionMDC(String sessionId) {
        Map<String, String> contextMap = SESSION_MDC_MAP.get(sessionId);
        if (contextMap != null) {
            MDC.setContextMap(contextMap);
        } else {
            MDC.clear(); // 防止残留
        }
    }

    private static void removeSessionMDC(String sessionId) {
        SESSION_MDC_MAP.remove(sessionId);
        MDC.clear();
    }

    // 事件载体（需可序列化，适配 Vert.x EventBus）
    public static class AsyncWebSocketMessage implements Serializable {
        private final String sessionId;
        private final String payload;

        public AsyncWebSocketMessage(String sessionId, String payload) {
            this.sessionId = sessionId;
            this.payload = payload;
        }

        public String getSessionId() { return sessionId; }
        public String getPayload() { return payload; }
    }
}

? 在事件消费者中还原 MDC

UserService 的 @ConsumeEvent 方法需在执行业务逻辑前，主动调用 UserWebSocketController.restoreSessionMDC(...)，确保日志上下文就位：

// UserService.java
@Slf4j
@ApplicationScoped
public class UserService {

    private final WebsocketConnectionService websocketConnectionService;
    private final Vertx vertx;

    public UserService(WebsocketConnectionService websocketConnectionService, Vertx vertx) {
        this.websocketConnectionService = websocketConnectionService;
        this.vertx = vertx;
    }

    @ConsumeEvent("websocket.message.new")
    public Uni<Void> handleWebSocketMessages(UserWebSocketController.AsyncWebSocketMessage msg) {
        // ✅ 关键：在业务逻辑前恢复 MDC
        UserWebSocketController.restoreSessionMDC(msg.getSessionId());

        // 此时 MDC 已就绪
        String userId = MDC.get("user.id");
        log.info("Processing message for user {} with payload: {}", userId, msg.getPayload());

        // 执行实际业务逻辑（如 DB 查询、外部调用等）
        // ... your business logic ...

        return Uni.createFrom().voidItem();
    }
}

⚠️ 注意事项与最佳实践

• 内存泄漏防护：务必在 @OnClose 和 @OnError 中调用 removeSessionMDC()，避免 SESSION_MDC_MAP 持续增长。生产环境建议增加 TTL 或结合 WeakReference 进一步优化。
• 序列化兼容性：AsyncWebSocketMessage 必须 implements Serializable，且所有字段需可序列化（Vert.x EventBus 默认使用 Jackson 序列化，若使用 Protobuf 等需额外配置）。
• 线程安全性：ConcurrentHashMap 保证了多线程存取安全，但 MDC.setContextMap() 本身是线程局部的，无需额外同步。
• 日志一致性：@OnOpen/@OnMessage/@OnClose 中均调用 restoreSessionMDC()，可确保整个 WebSocket 生命周期内日志链路完整。
• 替代方案考量：若项目已引入 Micrometer Tracing（如 OpenTelemetry），可优先使用 TracedExecutor 或 Tracer.withSpanInScope() 实现分布式追踪上下文透传，MDC 则作为补充的日志增强字段。

通过该方案，你能在 Quarkus WebSocket 的异步事件流中，稳定维持用户身份、会话标识等诊断信息，大幅提升日志可追溯性与故障排查效率。