如何将 ConcurrentWebSocketSessionDecorator 与自定义 WebSocketHandler 一起使用答案

【问题标题】：How to use ConcurrentWebSocketSessionDecorator with a custom WebSocketHandler如何将 ConcurrentWebSocketSessionDecorator 与自定义 WebSocketHandler 一起使用
【发布时间】：2021-05-25 05:21:32
【问题描述】：

我有一个自定义的 WebSocket 处理程序（用于自定义子协议），以通常的方式注册：

public class WSConfig implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry webSocketHandlerRegistry) {
        webSocketHandlerRegistry
                .addHandler(new PerConnectionWebSocketHandler(CustomProtocolHandler.class), endpointUrl);
    }
}

在CustomProtocolHandler 内部，所有从WebSocketHandler 继承的方法（例如afterConnectionEstablished、handleMessage 等）都接收WebSocketSession，这不是线程安全的。官方教程说ConcurrentWebSocketSessionDecorator 可以用来防止并发写入，但是当我的自定义处理程序被命中时，已经太迟包装会话了。

public class CustomProtocolHandler implements WebSocketHandler, SubProtocolCapable {

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) {
        //This is not safe
        session.sendMessage(...);

        //And this this doesn't make sense
        new ConcurrentWebSocketSessionDecorator(session).sendMessage(...);
    }
}

现在，我看到 Spring 自己的 SubProtocolWebSocketHandler（和我的一样，实现了 WebSocketHandler 和 SubProtocolCapable）确实在内部装饰了会话，我希望我能够使用它作为基础，但是要创建SubProtocolWebSocketHandler 我需要来自spring-messaging 的MessageChannel。我既不使用 Spring Messaging，也不知道如何获得MessageChannel。我应该实现一个自定义的吗？

那么，ConcurrentWebSocketSessionDecorator 打算如何使用？我希望有一种方法可以用webSocketHandlerRegistry 注册会话包装逻辑，但不是。我唯一的想法是维护另一个ConcurrentHashMap，它将原始会话（或其ID）映射到包装器，但这太可怕了，我不需要更多状态来管理和清理。

【问题讨论】：

标签： java spring websocket spring-websocket

【解决方案1】：

我最终制作了自己的PerConnectionWebSocketHandler 版本，与原始版本一样，它维护了一个会话到处理程序的映射，但我的处理程序将处理程序包装到一个持有人中，该持有人还包含一个装饰会话供处理程序使用。

原来的PerConnectionWebSocketHandler是这样的：

public class PerConnectionWebSocketHandler implements WebSocketHandler, BeanFactoryAware {

    private final Map<WebSocketSession, WebSocketHandler> handlers = new ConcurrentHashMap<>();

    ...

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        WebSocketHandler handler = this.provider.getHandler();
        this.handlers.put(session, handler);
        handler.afterConnectionEstablished(session);
    }
}

我的现在看起来像这样：

public class CustomPerConnectionWebSocketHandler implements WebSocketHandler {

    private final Map<WebSocketSession, HandlerWrapper> handlers = new ConcurrentHashMap<>();

    ...

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        WebSocketHandler handler = ...;
        //Here's a chance to decorate the session as needed
        WebSocketSession decoratedSession = new ConcurrentWebSocketSessionDecorator(session, sendTimeLimit, sendBufferSizeLimit);
        HandlerWrapper wrapper = new HandlerWrapper(handler, decoratedSession);
        this.handlers.put(session, wrapper);
        wrapper.afterConnectionEstablished();
    }


    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
         //Delegate all the calls in this fashion
         handlers.get(session).handleMessage(message);
    }

    ...

    private static class HandlerWrapper {

        private final WebSocketHandler handler;
        private final WebSocketSession session;

        HandlerWrapper(WebSocketHandler handler, WebSocketSession session) {
            this.handler = handler;
            this.session = session;
        }

        void afterConnectionEstablished() throws Exception {
            handler.afterConnectionEstablished(session);
        }

        void handleMessage(WebSocketMessage<?> message) throws Exception {
            handler.handleMessage(session, message);
        }

        ... //Other delegating methods
    }
}

【讨论】：