springboot websocket集群(stomp协议)连接时候传递参数

 更新时间:2019年07月06日 08:31:24   作者:郝二驴  
这篇文章主要介绍了springboot websocket集群(stomp协议)连接时候传递参数,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

最近在公司项目中接到个需求。就是后台跟前端浏览器要保持长连接,后台主动往前台推数据。

网上查了下,websocket stomp协议处理这个很简单。尤其是跟springboot 集成。

但是由于开始是单机玩的,很顺利。

但是后面部署到生产搞集群的话,就会出问题了。

假如集群两个节点,浏览器A与节点A建立连接,A节点发的消息浏览器A节点肯定能收到。但是B节点由于没有跟浏览器A建立连接。B节点发的消息浏览器就收不到了。

网上也查了好多,但是没有一个说的很清楚的,也很多都是理论层面的。

还有很多思路都是通过session获取信息的。但是这都不是我需要的。我需要的是从前台传递参数,连接的时候每个节点保存下。然后通过SimpleUserRegistry.getUser获取。

话不多说,直接上代码。

<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);
   });
  }
};

这是响应的前端代码。只需要引入两个js。调用new SockJS(url) 就代表跟服务器建立连接了。

@Configuration

//注解开启使用STOMP协议来传输基于代理(message broker)的消息,这时控制器支持使用@MessageMapping,就像使用@RequestMapping一样
@EnableWebSocketMessageBroker
public 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);
 }
}

注:上面的endpointOyzc就是前端的url。后面注册端点,前台链接。

然后注意下configureClientInboundChannel这个方法,这个方法里面注入拦截器就是为了链接时候接收参数的。

/**
 * @author : hao
 * @description : websocket建立链接的时候获取headeri里认证的参数拦截器。
 * @time : 2019/7/3 20:42
 */
@Component
public 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
@Slf4j
public 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);
  }

 }
}

可以发现上面代码利用了redis监听模型,也就是redis模型的消息队列

/**
 * @author : hao
 * @description : redis消息监听实现类,接收处理类
 * @time : 2019/7/3 14:00
 */
@Component
@Slf4j
public 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());
   }
  }
 }
}

redis消息模型只贴部分代码就好了

/**
  * 消息监听器
  */
 @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;
 }

上面的思路大体如下:客户端简历链接时候,传过来userid保存起来。发消息的时候 通过userRegistry获取,能获取到就证明是跟本节点建立的链接,直接用本节点发消息就好了。

如果不是就利用redis消息队列,把消息推出去。每个节点去判断获取看下是不是本节点的userid。这样就实现了集群的部署。

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。

相关文章

  • Java线程状态及jstack命令详解

    Java线程状态及jstack命令详解

    jstack是Java虚拟机(JVM)提供的一个非常有用的命令行工具,它允许开发人员和系统管理员在运行时获取Java应用程序的线程堆栈跟踪,在某些情况下,可能需要以管理员或root用户的身份运行jstack命令,这篇文章主要介绍了Java线程状态及jstack命令详解,需要的朋友可以参考下
    2024-03-03
  • 解决mybatis update并非所有字段需要更新问题

    解决mybatis update并非所有字段需要更新问题

    这篇文章主要介绍了解决mybatis update并非所有字段需要更新问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-11-11
  • Intellij IDEA如何查看所有断点

    Intellij IDEA如何查看所有断点

    这篇文章主要介绍了Intellij IDEA如何查看所有断点问题,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2023-03-03
  • SpringBoot注册FilterRegistrationBean相关情况讲解

    SpringBoot注册FilterRegistrationBean相关情况讲解

    这篇文章主要介绍了SpringBoot注册FilterRegistrationBean相关情况,借助FilterRegistrationBean来注册filter,可以避免在web.xml种配置filter这种原始的写法
    2023-02-02
  • 详解MyBatis开发Dao层的两种方式(Mapper动态代理方式)

    详解MyBatis开发Dao层的两种方式(Mapper动态代理方式)

    这篇文章主要介绍了详解MyBatis开发Dao层的两种方式(Mapper动态代理方式),小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
    2018-12-12
  • 在SpringBoot项目中整合拦截器的详细步骤

    在SpringBoot项目中整合拦截器的详细步骤

    在系统中经常需要在处理用户请求之前和之后执行一些行为,例如检测用户的权限,或者将请求的信息记录到日志中,即平时所说的"权限检测"及"日志记录",下面这篇文章主要给大家介绍了关于在SpringBoot项目中整合拦截器的相关资料,需要的朋友可以参考下
    2022-09-09
  • 聊聊springboot2.2.3升级到2.4.0单元测试的区别

    聊聊springboot2.2.3升级到2.4.0单元测试的区别

    这篇文章主要介绍了springboot 2.2.3 升级到 2.4.0单元测试的区别,具有很好的参考价值,希望对大家有所帮助。如有错误或未考虑完全的地方,望不吝赐教
    2021-10-10
  • Java数据结构之栈与队列实例详解

    Java数据结构之栈与队列实例详解

    这篇文章主要给大家介绍了关于Java数据结构之栈与队列的相关资料,算是作为用java描述数据结构的一个开始,文中通过实例代码介绍的非常详细,需要的朋友可以参考下
    2021-11-11
  • HttpMessageConverter报文信息转换器的深入讲解

    HttpMessageConverter报文信息转换器的深入讲解

    在Spring中内置了大量的HttpMessageConverter,通过请求头信息中的MIME类型,选择相应的HttpMessageConverter,这篇文章主要给大家介绍了关于HttpMessageConverter报文信息转换器的相关资料,需要的朋友可以参考下
    2022-01-01
  • java变量的声明与赋值分离规范示例

    java变量的声明与赋值分离规范示例

    这篇文章主要为大家介绍了java变量的声明与赋值分离规范示例,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
    2023-09-09

最新评论