spring 集成 websocket

spring4.0以后加入了对websocket技术的支持，目前项目用的是SSM（springMVC+spring+MyBatis）框架，所以选择了spring自带的websocket。

开始集成:

在pom.xml文件中增加以下依赖

<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-websocket</artifactId>
	<version>4.3.7.RELEASE</version>
</dependency>
<dependency>
	<groupId>org.springframework</groupId>
	<artifactId>spring-messaging</artifactId>
	<version>4.3.7.RELEASE</version>
</dependency>
<dependency>
	<groupId>javax.websocket</groupId>
	<artifactId>javax.websocket-api</artifactId>
	<version>1.1</version>
	<scope>provided</scope>
</dependency>

握手拦截器

public class WebSocketHandshakeInterceptor extends HttpSessionHandshakeInterceptor {

    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response,
                                   WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        if (request instanceof ServletServerHttpRequest) {
            String openid = ((ServletServerHttpRequest) request).getServletRequest().getParameter("openid");

            attributes.put("WEBSOCKET_USERNAME", openid);
        }
        return super.beforeHandshake(request, response, wsHandler, attributes);
    }

    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response,
                               WebSocketHandler wsHandler, Exception ex) {
        super.afterHandshake(request, response, wsHandler, ex);
    }
}

业务拦截器

public class UnionpaySocketHandler extends TextWebSocketHandler {

    private static final Logger logger = LoggerFactory.getLogger(UnionpaySocketHandler.class);

    private final static ConcurrentMap<String, WebSocketSession> sessions;

    static {
        sessions = new ConcurrentHashMap<>();
    }

    //接收文本消息，并发送出去
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {

        String msg = message.getPayload();

        TextMessage returnMessage = new TextMessage(msg);
        session.sendMessage(returnMessage);

        super.handleTextMessage(session, message);
    }

    //连接建立后处理
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String openid = (String) session.getAttributes().get("WEBSOCKET_USERNAME");
        if (StringUtils.isNotBlank(openid)) {
            logger.info("用户 >> {} 建立连接", openid);

            sessions.put(openid, session);
        }
    }

    //抛出异常时处理
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        if (session.isOpen()) {
            session.close();
        }

        String openid = (String) session.getAttributes().get("WEBSOCKET_USERNAME");

        sessions.remove(openid);
    }

    //连接关闭后处理
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {

        String openid = (String) session.getAttributes().get("WEBSOCKET_USERNAME");

        logger.info("用户 >> {} 退出连接", openid);

        sessions.remove(openid);
    }

    /**
     * 给某个用户发送消息
     *
     * @param userName
     * @param msg
     */
    public void sendMessageToUser(String userName, String msg) {
        WebSocketSession session = sessions.get(userName);
        if (session != null) {
            try {
                TextMessage message = new TextMessage(msg);

                session.sendMessage(message);
            } catch (IOException e) {
                logger.error("发送信息异常, ex = {}", ExceptionUtils.getStackTrace(e));
            }
        }
    }
}

配置类

@Configuration
@EnableWebSocket
public class WebSocketConfiguration implements WebSocketConfigurer {

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry
                .addHandler(unionpaySocketHandler(), "/unionpayServer")
                .setAllowedOrigins("*")//允许所有域访问
                .addInterceptors(myInterceptor());
        registry
                .addHandler(unionpaySocketHandler(), "/unionpayServer/sockjs")
                .setAllowedOrigins("*")
                .addInterceptors(myInterceptor())
                .withSockJS();
    }

    @Bean
    public UnionpaySocketHandler unionpaySocketHandler() {
        return new UnionpaySocketHandler();
    }

    @Bean
    public WebSocketHandshakeInterceptor myInterceptor() {
        return new WebSocketHandshakeInterceptor();
    }

}

web.xml增加配置

<servlet>
  <servlet-name>springDispatcherServlet</servlet-name>
  <servlet-class>org.springframework.web.servlet.DispatcherServlet</servlet-class>
  <init-param>
    <param-name>contextConfigLocation</param-name>
    <param-value>classpath*:spring/spring-mvc.xml</param-value>
  </init-param>
  <load-on-startup>1</load-on-startup>
  **<async-supported>true</async-supported>//需要增加此项  允许异步(未测试是否必须配置)
</servlet>

以下重点

如果使用了nginx做代理，则nginx需要增加配置，否则websocket开启会报错301或者400或者500等

• 以下为例子

location /websocket {
		proxy_pass   http://127.0.0.1:9090/unionpayServer;
		proxy_http_version 1.1;
		proxy_set_header Upgrade $http_upgrade;
		proxy_set_header Connection "upgrade";
		proxy_read_timeout 300s;
}

前端调用方法示例

websocket = new WebSocket("ws://localhost:8080/websocket?openid=123");//如果为https,ws改为wss

至此，整个websocket集成配置已完成。