欢迎关注方志朋的博客,回复”666“获面试宝典
来源:/weixin_45089791/article/
details/118028312
在显示项目中遇到了一个问题,需要使用到websocket与小程序建立长链接。由于项目是负载均衡的,存在项目部署在多台机器上。这样就会存在一个问题,当一次请求负载到第一台服务器时,socketsession在第一台服务器线程上,第二次请求,负载到第二台服务器上,需要通过id查找当前用户的session时,是查找不到的。
可以看到,由于websocket的session并没有实现序列化接口。所以无法将session序列化到redis中。
web的中的httpsession 主要是通过下面的两个管理器实现序列化的。
org.apache.catalina.session.StandardManagerorg.apache.catalina.session.PersistentManager
StandardManager是Tomcat默认使用的,在web应用程序关闭时,对内存中的所有HttpSession对象进行持久化,把他们保存到文件系统中。默认的存储文件为
<tomcat安装目录>/work/Catalina/<主机名>/<应用程序名>/sessions.ser
PersistentManager比StandardManager更为灵活,只要某个设备提供了实现org.apache.catalina.Store
接口的驱动类,PersistentManager就可以将HttpSession对象保存到该设备。
所以spring-session-redis
解决分布场景下的session共享就是将session序列化到redis中间件中,使用filter 加装饰器模式解决分布式场景httpsession 共享问题。
解决方案
使用消息中间件解决websocket session共享问题。
使用redis的发布订阅模式解决
本文使用方式二
使用StringRedisTemplate的convertAndSend方法向指定频道发送指定消息:
this.execute((connection)->{connection.publish(rawChannel,rawMessage);returnnull;},true);
redis的命令publish channel message
添加一个监听的容器以及一个监听器适配器
@BeanRedisMessageListenerContainercontainer(RedisConnectionFactoryconnectionFactory,MessageListenerAdapterlistenerAdapter){RedisMessageListenerContainercontainer=newRedisMessageListenerContainer();container.setConnectionFactory(connectionFactory);//可以添加多个messageListener,配置不同的交换机container.addMessageListener(listenerAdapter,newPatternTopic(Constants.REDIS_CHANNEL));//订阅最新消息频道returncontainer;}@BeanMessageListenerAdapterlistenerAdapter(RedisReceiverreceiver){//消息监听适配器returnnewMessageListenerAdapter(receiver,"onMessage");}
添加消息接收器
/***消息监听对象,接收订阅消息*/@ComponentpublicclassRedisReceiverimplementsMessageListener{Loggerlog=LoggerFactory.getLogger(this.getClass());@AutowiredprivateWebSocketServerwebSocketServer;/***处理接收到的订阅消息*/@OverridepublicvoidonMessage(Messagemessage,byte[]pattern){Stringchannel=newString(message.getChannel());//订阅的频道名称Stringmsg="";try{msg=newString(message.getBody(),Constants.UTF8);//注意与发布消息编码一致,否则会乱码if(!StringUtils.isEmpty(msg)){if(Constants.REDIS_CHANNEL.endsWith(channel))//最新消息{JSONObjectjsonObject=JSON.parseObject(msg);webSocketServer.sendMessageByWayBillId(Long.parseLong(jsonObject.get(Constants.REDIS_MESSAGE_KEY).toString()),jsonObject.get(Constants.REDIS_MESSAGE_VALUE).toString());}else{//TODO其他订阅的消息处理}}else{log.info("消息内容为空,不处理。");}}catch(Exceptione){log.error("处理消息异常:"+e.toString());e.printStackTrace();}}}
websocket的配置类
/***@description:websocket的配置类*/@Configuration@EnableWebSocketpublicclassWebSocketConfiguration{@BeanpublicServerEndpointExporterserverEndpointExporter(){returnnewServerEndpointExporter();}}
添加websocket的服务组件
@ServerEndpoint("/websocket/{id}")@ComponentpublicclassWebSocketServer{privatestaticfinallongsessionTimeout=600000;privatestaticfinalLoggerlog=LoggerFactory.getLogger(WebSocketServer.class);/***当前在线连接数*/privatestaticAtomicIntegeronlineCount=newAtomicInteger(0);/***用来存放每个客户端对应的WebSocketServer对象*/privatestaticConcurrentHashMap<Long,WebSocketServer>webSocketMap=newConcurrentHashMap<>();/***与某个客户端的连接会话,需要通过它来给客户端发送数据*/privateSessionsession;/***接收id*/privateLongid;@AutowiredprivateStringRedisTemplatetemplate;/***连接建立成功调用的方法*/@OnOpenpublicvoidonOpen(Sessionsession,@PathParam("id")Longid){session.setMaxIdleTimeout(sessionTimeout);this.session=session;this.id=id;if(webSocketMap.containsKey(id)){webSocketMap.remove(id);webSocketMap.put(id,this);}else{webSocketMap.put(id,this);addOnlineCount();}log.info("编号id:"+id+"连接,当前在线数为:"+getOnlineCount());try{sendMessage("连接成功!");}catch(IOExceptione){log.error("编号id:"+id+",网络异常!!!!!!");}}/***连接关闭调用的方法*/@OnClosepublicvoidonClose(){if(webSocketMap.containsKey(id)){webSocketMap.remove(id);subOnlineCount();}log.info("编号id:"+id+"退出,当前在线数为:"+getOnlineCount());}/***收到客户端消息后调用的方法**@parammessage客户端发送过来的消息*/@OnMessagepublicvoidonMessage(Stringmessage,Sessionsession){log.info("编号id消息:"+id+",报文:"+message);}/***发生错误时调用**@paramsession*@paramerror*/@OnErrorpublicvoidonError(Sessionsession,Throwableerror){log.error("编号id错误:"+this.id+",原因:"+error.getMessage());error.printStackTrace();}/***@description:分布式使用redis去发布消息*@dateTime:/6/1710:31*/publicvoidsendMessage(@NotNullStringkey,Stringmessage){StringnewMessge=null;try{newMessge=newString(message.getBytes(Constants.UTF8),Constants.UTF8);}catch(UnsupportedEncodingExceptione){e.printStackTrace();}Map<String,String>map=newHashMap<String,String>();map.put(Constants.REDIS_MESSAGE_KEY,key);map.put(Constants.REDIS_MESSAGE_VALUE,newMessge);template.convertAndSend(Constants.REDIS_CHANNEL,JSON.toJSONString(map));}/***@description:单机使用外部接口通过指定的客户id向该客户推送消息。*@dateTime:/6/1617:49*/publicvoidsendMessageByWayBillId(@NotNullLongkey,Stringmessage){WebSocketServerwebSocketServer=webSocketMap.get(key);if(!StringUtils.isEmpty(webSocketServer)){try{webSocketServer.sendMessage(message);log.info("编号id为:"+key+"发送消息:"+message);}catch(IOExceptione){e.printStackTrace();log.error("编号id为:"+key+"发送消息失败");}}log.error("编号id号为:"+key+"未连接");}/***实现服务器主动推送*/publicvoidsendMessage(Stringmessage)throwsIOException{this.session.getBasicRemote().sendText(message);}publicstaticsynchronizedAtomicIntegergetOnlineCount(){returnonlineCount;}publicstaticsynchronizedvoidaddOnlineCount(){WebSocketServer.onlineCount.getAndIncrement();}publicstaticsynchronizedvoidsubOnlineCount(){WebSocketServer.onlineCount.getAndDecrement();}}
项目结构
将该项目使用三个端口号启动三个服务
使用下面的这个网站进行演示。
/wstool.html
启动两个页面网址分别是:
ws://127.0.0.1:8081/websocket/456
ws://127.0.0.1:8082/websocket/456
使用postman给http://localhost:8080/socket/456
发送请求
可以看到,我们给8080服务发送的消息,我们订阅的8081和8082 服务可以也可以使用该编号进行消息的推送。
使用8082服务发送这个消息格式{"KEY":456,"VALUE":"aaaa"} 的消息。其他的服务也会收到这个信息。
以上就是使用redis的发布订阅解决websocket 的分布式session 问题。
码云地址是:/jack_whh/dcs-websocket-session
热门内容:
一款基于 Spring Boot 的现代化社区
Spring Boot 3.0 M1 发布,正式弃用 Java 8,最低要求 Java 17。。。
「冰墩墩」代码,开源了!
比 Xshell 还好用的 SSH 客户端神器,MobaXterm 太爱了!
干掉visio,这个画图神器真的绝了!!!
最近面试BAT,整理一份面试资料《Java面试BAT通关手册》,覆盖了Java核心技术、JVM、Java并发、SSM、微服务、数据库、数据结构等等。获取方式:点“在看”,关注公众号并回复666领取,更多内容陆续奉上。
明天见(。・ω・。)ノ♡