Spring+Websocket实现消息的推送
jopen
10年前
本文主要有三个步骤 1、用户登录后建立websocket连接，默认选择websocket连接，如果浏览器不支持，则使用sockjs进行模拟连接 2、建立连接后，服务端返回该用户的未读消息 3、服务端进行相关操作后，推送给某一个用户或者所有用户新消息 相关环境 Spring4.0.6(要选择4.0+),tomcat7.0.55
Websocet服务端实现
WebSocketConfig.java
@Configuration @EnableWebMvc @EnableWebSocket public class WebSocketConfig extends WebMvcConfigurerAdapter implements WebSocketConfigurer { @Override public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { registry.addHandler(systemWebSocketHandler(),"/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor()); registry.addHandler(systemWebSocketHandler(), "/sockjs/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor()) .withSockJS(); } @Bean public WebSocketHandler systemWebSocketHandler(){ return new SystemWebSocketHandler(); } }不要忘记在springmvc的配置文件中配置对此类的自动扫描 <context:component-scan base-package="com.ldl.origami.websocket" /><p>@Configuration </p> <p>@EnableWebMvc
@EnableWebSocket
这三个大致意思是使这个类支持以@Bean的方式加载bean，并且支持springmvc和websocket，不是很准确大致这样，试了一下<span>@EnableWebMvc不加也没什么影响，<span>@Configuration本来就支持springmvc的自动扫描</span></span> </p>
registry.addHandler(systemWebSocketHandler(),"/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor())用来注册websocket server实现类，第二个参数是访问websocket的地址
registry.addHandler(systemWebSocketHandler(), "/sockjs/webSocketServer").addInterceptors(new WebSocketHandshakeInterceptor()) .withSockJS(); }这个是使用Sockjs的注册方法
首先SystemWebSocketHandler.java
public class SystemWebSocketHandler implements WebSocketHandler { private static final Logger logger; private static final ArrayList<WebSocketSession> users; static { users = new ArrayList<>(); logger = LoggerFactory.getLogger(SystemWebSocketHandler.class); } @Autowired private WebSocketService webSocketService; @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { logger.debug("connect to the websocket success......"); users.add(session); String userName = (String) session.getAttributes().get(Constants.WEBSOCKET_USERNAME); if(userName!= null){ //查询未读消息 int count = webSocketService.getUnReadNews((String) session.getAttributes().get(Constants.WEBSOCKET_USERNAME)); session.sendMessage(new TextMessage(count + "")); } } @Override public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception { //sendMessageToUsers(); } @Override public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { if(session.isOpen()){ session.close(); } logger.debug("websocket connection closed......"); users.remove(session); } @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception { logger.debug("websocket connection closed......"); users.remove(session); } @Override public boolean supportsPartialMessages() { return false; } /** * 给所有在线用户发送消息 * * @param message */ public void sendMessageToUsers(TextMessage message) { for (WebSocketSession user : users) { try { if (user.isOpen()) { user.sendMessage(message); } } catch (IOException e) { e.printStackTrace(); } } } /** * 给某个用户发送消息 * * @param userName * @param message */ public void sendMessageToUser(String userName, TextMessage message) { for (WebSocketSession user : users) { if (user.getAttributes().get(Constants.WEBSOCKET_USERNAME).equals(userName)) { try { if (user.isOpen()) { user.sendMessage(message); } } catch (IOException e) { e.printStackTrace(); } break; } } } }
然后WebSocketHandshakeInterceptor.java
public class WebSocketHandshakeInterceptor implements HandshakeInterceptor { private static Logger logger = LoggerFactory.getLogger(HandshakeInterceptor.class); @Override public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object > attributes) throws Exception { if (request instanceof ServletServerHttpRequest) { ServletServerHttpRequest servletRequest = (ServletServerHttpRequest) request; HttpSession session = servletRequest.getServletRequest().getSession(false); if (session != null) { //使用userName区分WebSocketHandler，以便定向发送消息 String userName = (String) session.getAttribute(Constants.SESSION_USERNAME); attributes.put(Constants.WEBSOCKET_USERNAME,userName); } } return true; } @Override public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception exception) { } } 这个的主要作用是取得当前请求中的用户名，并且保存到当前的WebSocketHandler中，以便确定WebSocketHandler所对应的用户，具体可参考HttpSessionHandshakeInterceptor
用户登录建立websocket连接
index.jsp
<script type="text/javascript" src="http://localhost:8080/Origami/websocket/sockjs-0.3.min.js"></script> <script> var websocket; if ('WebSocket' in window) { websocket = new WebSocket("ws://localhost:8080/Origami/webSocketServer"); } else if ('MozWebSocket' in window) { websocket = new MozWebSocket("ws://localhost:8080/Origami/webSocketServer"); } else { websocket = new SockJS("http://localhost:8080/Origami/sockjs/webSocketServer"); } websocket.onopen = function (evnt) { }; websocket.onmessage = function (evnt) { $("#msgcount").html("(<font color='red'>"+evnt.data+"</font>)") }; websocket.onerror = function (evnt) { }; websocket.onclose = function (evnt) { } </script>
使用sockjs时要注意
1、这两个的写法
<script type="text/javascript" src="http://localhost:8080/Origami/websocket/sockjs-0.3.min.js"></script>
websocket = new SockJS("http://localhost:8080/Origami/sockjs/webSocketServer");2、web.xml中 <web-app version="3.0" xmlns="http://java.sun.com/xml/ns/javaee" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_3_1.xsd">
version
web-app_3_1.xsd这两个的版本都要是3.0+
然后在这个servlet中加入
<async-supported>true</async-supported>
<servlet> <servlet-name>appServlet</servlet-name> <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class> <init-param> <param-name>contextConfigLocation</param-name> <param-value>classpath*:servlet-context.xml</param-value> </init-param> <load-on-startup>1</load-on-startup> <async-supported>true</async-supported> </servlet>
然后所有的filter中也加入
<async-supported>true</async-supported>
3、添加相关依赖
<dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-annotations</artifactId> <version>2.3.0</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-core</artifactId> <version>2.3.1</version> </dependency> <dependency> <groupId>com.fasterxml.jackson.core</groupId> <artifactId>jackson-databind</artifactId> <version>2.3.3</version> </dependency>好了，现在websocket可以正常建立起来了
返回用户未读的消息
当连接建立后，会进入SystemWebSocketHandler的afterConnectionEstablished方法，代码看上边，取出WebSocketHandshakeInterceptor中保存的用户名
查询信息后使用session.sendMessage(new TextMessage(count + ""));返回给用户，从哪来回哪去
服务端推送消息给用户
@Controller public class AdminController { static Logger logger = LoggerFactory.getLogger(AdminController.class); @Autowired(required = false) private AdminService adminService; @Bean public SystemWebSocketHandler systemWebSocketHandler() { return new SystemWebSocketHandler(); } @RequestMapping("/auditing") @ResponseBody public String auditing(HttpServletRequest request){ //无关代码都省略了 int unReadNewsCount = adminService.getUnReadNews(username); systemWebSocketHandler().sendMessageToUser(username, new TextMessage(unReadNewsCount + "")); return result; } } 在这里可以使用sendMessageToUser给某个用户推送信息，也可以使用sendMessageToUsers给所有用户推送信息
来自：http://my.oschina.net/ldl123292/blog/304360