spring - spring websocket＋rabbitmq stomp：如何获取所有在线用户并断开部分用户的连接？

Question

如果spring store用户的websocket session，怎么获取，如果我用someHandler把session放到map中，那意味着map的大小可能大于10w＋，我觉得不太好。</p>

会话处理程序.java

package hello;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
public class SessionHandler {
    private static final Logger LOGGER = LoggerFactory.getLogger(SessionHandler.class);
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    private final Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();

    public SessionHandler() {
        scheduler.scheduleAtFixedRate(new Runnable() {
            @Override
            public void run() {
                sessionMap.keySet().forEach(k -> {
                    try {
                        sessionMap.get(k).close();
                        sessionMap.remove(k);
                    } catch (IOException e) {
                        LOGGER.error("Error while closing websocket session: {}", e);
                    }
                });
            }
        }, 10, 10, TimeUnit.SECONDS);
    }

    public void register(WebSocketSession session) {
        sessionMap.put(session.getId(), session);
    }

}

像这样： https ://github.com/isaranchuk/spring-websocket-disconnect/blob/master/src/main/java/hello/SessionHandler.java

Answer 1

如下实现 HandshakeInter。

public class HttpSessionIdHandshakeInterceptor implements HandshakeInterceptor {

    public boolean beforeHandshake(ServerHttpRequest request, 
            ServerHttpResponse response, 
            WebSocketHandler wsHandler, 
            Map<String, Object> attributes) 
              throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request;
            HttpSession session = servletRequest.getServletRequest().getSession(false);
            if (session != null) {
                attributes.put(SESSION_ATTR, session.getId());
            }
        }
        return true;
    }

    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception ex) {
    }
}

然后添加到端点。

public class WebSocketConfig extends 
      AbstractWebSocketMessageBrokerConfigurer {

public void registerStompEndpoints(StompEndpointRegistry registry) {
    registry.addEndpoint("/portfolio")
            .withSockJS()
            .setInterceptors(new HttpSessionIdHandshakeInterceptor());

}

... }

接下来我们可以创建一个 ChannelInterceptorAdapter，它使用 session id 来更新使用 Spring Session 的最后访问时间。例如：

@Bean public ChannelInterceptorAdapter sessionContextChannelInterceptorAdapter() {
return new ChannelInterceptorAdapter() {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        Map<String, Object> sessionHeaders = SimpMessageHeaderAccessor.getSessionAttributes(message.getHeaders());
        String sessionId = (String) sessionHeaders.get(SESSION_ATTR);
        if (sessionId != null) {
            Session session = sessionRepository.getSession(sessionId);
            if (session != null) {
                sessionRepository.save(session);
            }
        }
        return super.preSend(message, channel);
    }
};

}