基于redis订阅消息和websocket技术实现的消息推送功能 |
您所在的位置：网站首页 › websocketserver怎么安装 › 基于redis订阅消息和websocket技术实现的消息推送功能 |
基于redis订阅消息和websocket技术实现的消息推送功能
|
相关依赖文件 创建Redis消息监听者容器 @Configuration //相当于xml中的beans public class RedisConfig { /** * 需要手动注册RedisMessageListenerContainer加入IOC容器 * @author lijt * @return */ @Bean public RedisMessageListenerContainer redisMessageListenerContainer(RedisConnectionFactory connectionFactory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); //订阅了一个叫chat 的通道 container.addMessageListener(new MessageListener(){ @Override public void onMessage(Message message, byte[] pattern) { String msg = new String(message.getBody()); System.out.println(new String(pattern) + "主题发布：" + msg); } }, new PatternTopic("TOPIC")); return container; } }创建Websocket配置类 @Configuration public class WebSocketConfig { @Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } } 这个配置类的作用是要注入ServerEndpointExporter，这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint。如果是使用独立的servlet容器，而不是直接使用springboot的内置容器，就不要注入ServerEndpointExporter，因为它将由容器自己提供和管理。 本文的例子是采用的springboot的内置tomcat容器，所以还是要创建这个配置类，作用就是注入ServerEndpointExporter。创建消息订阅监听者类 public class SubscribeListener implements MessageListener { private Session session; public Session getSession() { return session; } public void setSession(Session session) { this.session = session; } /** * 接收发布者的消息 */ @Override public void onMessage(Message message, byte[] pattern) { String msg = new String(message.getBody()); System.out.println(new String(pattern) + "主题发布：" + msg); if (null != session && session.isOpen()) { try { session.getBasicRemote().sendText(msg); } catch (IOException e) { e.printStackTrace(); } } } } 这个消息订阅监听者类持有websocket的客户端会话对象（session），当接收到订阅的消息时，通过这个会话对象（session）将消息发送到前端，从而实现消息的主动推送。创建Websocket服务端类 @Component @ServerEndpoint("/websocket/server") public class WebSocketServer { /** * 因为@ServerEndpoint不支持注入，所以使用SpringUtils获取IOC实例 */ private RedisMessageListenerContainer redisMessageListenerContainer = SpringUtils.getBean(RedisMessageListenerContainer.class); //静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。 private static AtomicInteger onlineCount=new AtomicInteger(0); //concurrent包的线程安全Set，用来存放每个客户端对应的webSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识 private static CopyOnWriteArraySet webSocketSet = new CopyOnWriteArraySet(); //与某个客户端的连接会话，需要通过它来给客户端发送数据 private Session session; private SubscribeListener subscribeListener; /** * 连接建立成功调用的方法 * @param session 可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据 */ @OnOpen public void onOpen(Session session){ this.session = session; webSocketSet.add(this); //加入set中 addOnlineCount(); //在线数加1 System.out.println("有新连接加入！当前在线人数为" + getOnlineCount()); subscribeListener = new SubscribeListener(); subscribeListener.setSession(session); //设置订阅topic redisMessageListenerContainer.addMessageListener(subscribeListener, new ChannelTopic("TOPIC")); } /** * 连接关闭调用的方法 */ @OnClose public void onClose() throws IOException { webSocketSet.remove(this); //从set中删除 subOnlineCount(); //在线数减1 redisMessageListenerContainer.removeMessageListener(subscribeListener); System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount()); } /** * 收到客户端消息后调用的方法 * @param message 客户端发送过来的消息 * @param session 可选的参数 */ @OnMessage public void onMessage(String message, Session session) { System.out.println("来自客户端的消息:" + message); //群发消息 for(WebSocketServer item: webSocketSet){ try { item.sendMessage(message); } catch (IOException e) { e.printStackTrace(); continue; } } } /** * 发生错误时调用 * @param session * @param error */ @OnError public void onError(Session session, Throwable error){ System.out.println("发生错误"); error.printStackTrace(); } /** * 这个方法与上面几个方法不一样。没有用注解，是根据自己需要添加的方法。 * @param message * @throws IOException */ public void sendMessage(String message) throws IOException { this.session.getBasicRemote().sendText(message); } public int getOnlineCount() { return onlineCount.get(); } public void addOnlineCount() { WebSocketServer.onlineCount.getAndIncrement(); } public void subOnlineCount() { WebSocketServer.onlineCount.getAndDecrement(); } } @ServerEndpoint 注解是一个类层次的注解，它的功能主要是将目前的类定义成一个websocket服务器端,注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端使用springboot的唯一区别是要@Component声明下，而使用独立容器是由容器自己管理websocket的，但在springboot中连容器都是spring管理的。 虽然@Component默认是单例模式的，但springboot还是会为每个websocket连接初始化一个bean，所以可以用一个静态set保存起来。注意的是在客户端链接关闭的方法onClose中，一定要 删除之前的订阅监听对象，就是下面这行代码：redisMessageListenerContainer.removeMessageListener(subscribeListener); 否则在浏览器刷一下之后，后台会报如下错误: java.lang.IllegalStateException: The WebSocket session [0] has been closed and no method (apart from close()) may be called on a closed session 原因就是当链接关闭之后，session对象就没有了，而订阅者对象还是会接收消息，在用session对象发送消息时会报错。虽然代码中加了判断if (null != session && session.isOpen()) { 可以避免报错，但是为了防止内存泄漏，应该把没有用的监听者对象从容器中删除。创建前端页面 在resource\static目录下创建html页面，命名为websocket.html。代码如下： websocket 使用redis订阅消息和websocket实现消息推送 收到的订阅消息： var websocket = null; //当前浏览前是否支持websocket if("WebSocket" in window){ var url = "ws://localhost:8080/demo/websocket/server"; websocket = new WebSocket(url); }else{ alert("浏览器不支持websocket"); } websocket.onopen = function(event){ setMessage("打开连接"); } websocket.onclose = function(event){ setMessage("关闭连接"); } websocket.onmessage = function(event){ setMessage(event.data); } websocket.onerror = function(event){ setMessage("连接异常"); } //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。 window.onbeforeunload = function(){ closeWebsocket(); } //关闭websocket function closeWebsocket(){ //3代表已经关闭 if(3!=websocket.readyState){ websocket.close(); }else{ alert("websocket之前已经关闭"); } } //将消息显示在网页上 function setMessage(message){ document.getElementById('message_id').innerHTML += message + ''; }启动服务进行测试 1. 启动springboot服务，浏览器输入地址：，此时页面显示如下： 2.打开redis客户端，在命令行输入publish TOPIC “this is test message” 浏览器页面显示如下： 说明刚刚发布的消息已经主动推送到浏览器显示了。 完整代码见： |
【本文地址】
今日新闻 |
推荐新闻 |