springboot websocket集群（stomp协议）连接时候传递参数

<script type="text/javascript" src="${request.contextPath}/scripts/sockjs.min.js"></script><script type="text/javascript" src="${request.contextPath}/scripts/stomp.min.js"></script>var WEB_SOCKET = {   topic : "",  url : "",  stompClient : null,    connect : function(url, topic, callback,userid) {   this.url = url;   this.topic = topic;   var socket = new SockJS(url); //连接SockJS的endpoint名称为"endpointOyzc"   WEB_SOCKET.stompClient = Stomp.over(socket);//使用STMOP子协议的WebSocket客户端   WEB_SOCKET.stompClient.connect({userid:userid},function(frame){//连接WebSocket服务端    // console.log('Connected:' + frame);    //通过stompClient.subscribe订阅/topic/getResponse 目标(destination)发送的消息    WEB_SOCKET.stompClient.subscribe(topic, callback);   });  }};

@Configuration//注解开启使用STOMP协议来传输基于代理(message broker)的消息,这时控制器支持使用@MessageMapping,就像使用@RequestMapping一样@EnableWebSocketMessageBrokerpublic class WebSocketConfig extends AbstractWebSocketMessageBrokerConfigurer { @Autowired private GetHeaderParamInterceptor getHeaderParamInterceptor; @Override //注册STOMP协议的节点(endpoint),并映射指定的url public void registerStompEndpoints(StompEndpointRegistry registry) {  //注册一个STOMP的endpoint,并指定使用SockJS协议  registry.addEndpoint("/endpointOyzc")    .setAllowedOrigins("*")    .withSockJS();  /* registry.addEndpoint("/endpointOyzc")    .setAllowedOrigins("*")    .setHandshakeHandler(xlHandshakeHandler)    .withSockJS();*/ } @Override //配置消息代理(Message Broker) public void configureMessageBroker(MessageBrokerRegistry registry) {  //点对点应配置一个/user消息代理，广播式应配置一个/topic消息代理  registry.enableSimpleBroker("/topic", "/user");  // 全局使用的消息前缀（客户端订阅路径上会体现出来）  //registry.setApplicationDestinationPrefixes("/app");  //点对点使用的订阅前缀（客户端订阅路径上会体现出来），不设置的话，默认也是/user/  registry.setUserDestinationPrefix("/user"); } /**  * 采用自定义拦截器，获取connect时候传递的参数  *  * @param registration  */ @Override public void configureClientInboundChannel(ChannelRegistration registration) {  registration.interceptors(getHeaderParamInterceptor); }}

/** * @author : hao * @description : websocket建立链接的时候获取headeri里认证的参数拦截器。 * @time : 2019/7/3 20:42 */@Componentpublic class GetHeaderParamInterceptor extends ChannelInterceptorAdapter { @Override public Message<?> preSend(Message<?> message, MessageChannel channel) {  StompHeaderAccessor accessor = MessageHeaderAccessor.getAccessor(message, StompHeaderAccessor.class);  if (StompCommand.CONNECT.equals(accessor.getCommand())) {   Object raw = message.getHeaders().get(SimpMessageHeaderAccessor.NATIVE_HEADERS);   if (raw instanceof Map) {    Object name = ((Map) raw).get("userid");    if (name instanceof LinkedList) {     // 设置当前访问的认证用户     accessor.setUser(new JqxxPrincipal(((LinkedList) name).get(0).toString()));    }   }  }  return message; }}/** * @author : hao * @description : 自定义的java.security.Principal * @time : 2019/7/3 20:42 */public class JqxxPrincipal implements Principal { private String loginName; public JqxxPrincipal(String loginName) {  this.loginName = loginName; } @Override public String getName() {  return loginName; }}

/** * @author : hao * @description : websocket发送代理，负责发送消息 * @time : 2019/7/4 11:01 */@Component@Slf4jpublic class WebsocketSendProxy<T> { @Autowired private SimpMessagingTemplate template; @Autowired private SimpUserRegistry userRegistry; @Resource(name = "redisServiceImpl") private RedisService redisService; @Value("spring.redis.message.topic-name") private String topicName; public void sendMsg(RedisWebsocketMsg<T> redisWebsocketMsg) {  SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver());  log.info("发送消息前获取接收方为{},根据Registry获取本节点上这个用户{}", redisWebsocketMsg.getReceiver(), simpUser);  if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) {   //2. 获取WebSocket客户端的订阅地址   WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode());   if (channelEnum != null) {    //3. 给WebSocket客户端发送消息    template.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent());   }  } else {   //给其他订阅了主题的节点发消息,因为本节点没有   redisService.convertAndSend(topicName, redisWebsocketMsg);  } }}

/** * @author : hao * @description : redis消息监听实现类，接收处理类 * @time : 2019/7/3 14:00 */@Component@Slf4jpublic class MessageReceiver { @Autowired private SimpMessagingTemplate messagingTemplate; @Autowired private SimpUserRegistry userRegistry; /**  * 处理WebSocket消息  */ public void receiveMessage(RedisWebsocketMsg redisWebsocketMsg) {  log.info(MessageFormat.format("Received Message: {0}", redisWebsocketMsg));  //1. 取出用户名并判断是否连接到当前应用节点的WebSocket  SimpUser simpUser = userRegistry.getUser(redisWebsocketMsg.getReceiver());  if (simpUser != null && StringUtils.isNotBlank(simpUser.getName())) {   //2. 获取WebSocket客户端的订阅地址   WebSocketChannelEnum channelEnum = WebSocketChannelEnum.fromCode(redisWebsocketMsg.getChannelCode());   if (channelEnum != null) {    //3. 给WebSocket客户端发送消息    messagingTemplate.convertAndSendToUser(redisWebsocketMsg.getReceiver(), channelEnum.getSubscribeUrl(), redisWebsocketMsg.getContent());   }  } }}

/**  * 消息监听器  */ @Bean MessageListenerAdapter messageListenerAdapter(MessageReceiver messageReceiver, Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer){  //消息接收者以及对应的默认处理方法  MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(messageReceiver, "receiveMessage");  //消息的反序列化方式  messageListenerAdapter.setSerializer(jackson2JsonRedisSerializer);  return messageListenerAdapter; } /**  * message listener container  */ @Bean RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory   , MessageListenerAdapter messageListenerAdapter){  RedisMessageListenerContainer container = new RedisMessageListenerContainer();  container.setConnectionFactory(connectionFactory);  //添加消息监听器  container.addMessageListener(messageListenerAdapter, new PatternTopic(topicName));  return container; }