如何通过Spring WebSocket Handler访问已建立的会话？

我正在测试基于Spring WebSocket的OCPP服务器通信，握手过程正常，当充电桩发送BootNotification、StatusNotification等消息时，我可以与客户端正常交互。但有时需要服务器主动向充电桩发送内容（比如请求远程交易、获取设备信息等），无需等待充电桩先发送消息。请问如何通过其他服务访问已建立的WebSocket会话（示例会话地址：ws:localhost:8080/central/station01）？

@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {
    @Bean
    public WebSocketHandler myHandler() {
        return new MyHandler();
    }
    
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(myHandler(), "/central/**")
                .addInterceptors(new HttpSessionHandshakeInterceptor());
    }   
}

public class MyHandler extends TextWebSocketHandler implements SubProtocolCapable {
    
    private final String[] subProtocols = {"ocpp1.6", "ocpp2.0"};
    
    @Autowired
    private ClientRepository clientRepo;
    
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        Global.id="";
        Global.session="";
        Global.client="Close";
        System.out.print("\n Connection Close \n"+"Session: "+session.getId()+"\n");
        session.getHandshakeHeaders();
        System.out.print("session enabled"+session);
        
    }
    @Override
    public void afterConnectionEstablished(WebSocketSession session)
            throws Exception{
        Global.id=session.getId();
        Global.session=session.getUri().toString();
        Global.client="ok";
        
        Client aa= new Client(null,session.getId(),session.getUri().toString(),"ok","");
        System.out.print("\n Connected \n"+"Session: "+session.getId()+"\n");
        
        System.out.print(session.getUri());
        
    }

    @Override
    public void handleMessage(WebSocketSession session,WebSocketMessage<?> message)  throws Exception {
        System.out.print("\n "+message.getPayload());
        Integer id=0;
        Global.ocpp=Global.ocpp+" \n "+message.getPayload().toString();
        
        ZonedDateTime data = ZonedDateTime.now();
        BootNotificationResponse stat=new BootNotificationResponse("Accepted",data.toString().substring(0,data.toString().length()-"[America/Sao_Paulo]".length()),300);
        JSONArray mm=new JSONArray((message.getPayload()).toString());
        id=(int )mm.get(0)+1;
        
        JSONObject ss=new JSONObject(stat);
        System.out.print(session.getHandshakeHeaders());
        JSONArray ja = new JSONArray();
        
        ja.put(3);
        ja.put(mm.get(1));
        ja.put(ss);
        
        Client dados=new Client(null,Global.id,Global.session,Global.client,message.getPayload().toString());
        clientRepo.save(dados);
        if(mm.get(2).equals("Authorize")) {
            
            JSONArray nob = new JSONArray();
            JSONObject iii=new JSONObject(new Auth(new AuthorizeResponse("1233434","ddfd","Accepted")));
            nob.put(3);
            nob.put(mm.get(1));
            nob.put(iii);
            System.out.print(nob);
            
            session.sendMessage(new TextMessage(nob.toString()));
        }
        if(mm.get(2).equals("BootNotification")) {
            System.out.print("Boot \n");
            
            session.sendMessage(new TextMessage(ja.toString()));
        }
        
    }
    
    @Override
    public List<String> getSubProtocols() {
        System.out.print(Arrays.asList(subProtocols));
        return Arrays.asList(subProtocols);
    }
}

要实现服务器主动向指定充电桩发送消息，核心是维护所有活跃的WebSocket会话，并提供给其他服务访问的入口。当前代码使用Global类存储会话，但仅能保存单个会话，无法支持多充电桩连接，需要做以下改造：

1. 创建会话管理组件 #

实现一个线程安全的会话管理器，用于存储和管理所有活跃的WebSocket会话，键使用充电桩唯一标识（比如从会话URI中提取的stationId）：

@Component
public class WebSocketSessionManager {
    // 用ConcurrentHashMap保证线程安全，key为充电桩标识，value为对应的WebSocket会话
    private final ConcurrentHashMap<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();

    // 添加会话
    public void addSession(String stationId, WebSocketSession session) {
        sessionMap.put(stationId, session);
    }

    // 移除会话
    public void removeSession(String stationId) {
        sessionMap.remove(stationId);
    }

    // 根据充电桩标识获取会话
    public WebSocketSession getSession(String stationId) {
        return sessionMap.get(stationId);
    }
}

2. 修改WebSocket Handler，集成会话管理器 #

在连接建立和关闭时，更新会话管理器中的会话记录，同时从会话URI中提取充电桩标识（比如从/central/station01中提取station01）：

public class MyHandler extends TextWebSocketHandler implements SubProtocolCapable {
    
    private final String[] subProtocols = {"ocpp1.6", "ocpp2.0"};
    
    @Autowired
    private ClientRepository clientRepo;
    // 注入会话管理器
    @Autowired
    private WebSocketSessionManager sessionManager;
    
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        // 提取充电桩标识
        String stationId = extractStationIdFromUri(session.getUri());
        sessionManager.removeSession(stationId);
        
        // 原有逻辑保留
        Global.id="";
        Global.session="";
        Global.client="Close";
        System.out.print("\n Connection Close \n"+"Session: "+session.getId()+"\n");
        session.getHandshakeHeaders();
        System.out.print("session enabled"+session);
    }
    
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception{
        // 提取充电桩标识
        String stationId = extractStationIdFromUri(session.getUri());
        sessionManager.addSession(stationId, session);
        
        // 原有逻辑保留
        Global.id=session.getId();
        Global.session=session.getUri().toString();
        Global.client="ok";
        
        Client aa= new Client(null,session.getId(),session.getUri().toString(),"ok","");
        System.out.print("\n Connected \n"+"Session: "+session.getId()+"\n");
        System.out.print(session.getUri());
    }

    // 从URI中提取充电桩标识，比如ws://localhost:8080/central/station01 -> station01
    private String extractStationIdFromUri(URI uri) {
        String path = uri.getPath();
        // 分割路径，取最后一段作为stationId
        String[] pathSegments = path.split("/");
        return pathSegments[pathSegments.length - 1];
    }

    // 原有handleMessage等逻辑保留...
}

3. 在其他服务中主动发送消息 #

在需要主动发送消息的服务中，注入WebSocketSessionManager，获取目标会话后发送消息：

@Service
public class OcppCommandService {
    
    @Autowired
    private WebSocketSessionManager sessionManager;

    // 示例：向指定充电桩发送远程交易请求
    public void sendRemoteStartTransaction(String stationId, String requestContent) throws IOException {
        WebSocketSession session = sessionManager.getSession(stationId);
        if (session != null && session.isOpen()) {
            // 构造符合OCPP协议的消息（根据OCPP版本调整格式）
            TextMessage message = new TextMessage(requestContent);
            session.sendMessage(message);
        } else {
            // 处理会话不存在或已关闭的情况
            System.out.println("会话不存在或已关闭：" + stationId);
        }
    }
}

注意事项 #

• 会话管理必须保证线程安全，使用ConcurrentHashMap避免并发问题
• 发送消息前要检查会话是否处于打开状态，避免抛出异常
• 充电桩标识的提取逻辑可根据实际场景调整（比如从握手Header、请求参数中获取）
• 若需要支持会话超时清理，可在会话管理器中定期检查会话状态并移除失效会话

内容的提问来源于stack exchange，提问作者Lucas silva