[网络协议]spring-boot&cloud Websocket 使用

spring-boot&cloud Websocket 使用

1 WebSocket介绍：
WebSocket是HTML5新增的协议，它的目的是在浏览器和服务器之间建立一个不受限的双向通信的通道，比如说，服务器可以在任意时刻发送消息给浏览器。
为什么传统的HTTP协议不能做到WebSocket实现的功能？这是因为HTTP协议是一个请求－响应协议，请求必须先由浏览器发给服务器，服务器才能响应这个请求，再把数据发送给浏览器。换句话说，浏览器不主动请求，服务器是没法主动发数据给浏览器的。

2 websocket springboot 项目实现：
2.1 引入websocket jar包和增加配置文件：
2.1.1 jar 依赖

  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>

2.1.1 WebSocketConfig.java 配置文件

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

@Configuration
@EnableWebSocket
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}

2.2 定义websocket 连接点：

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
//import xxxx.common.ResponseDTO;
//import xxxx.constant.websocket.FlushRedisConstant;
//import xxxx.session.dto.websocket.WebSocketReqDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;



@Slf4j
@Component
@ServerEndpoint("/websocket/wxkf/activity")
public class FlushWebSocket {

    /**
     * 记录当前在线连接数
     */
    private static AtomicInteger onlineCount = new AtomicInteger(0);

    /**
     * 用于存所有的连接服务的客户端，这个对象存储是安全的
     */
    private volatile static ConcurrentHashMap<String, Session> webSocketSet = new ConcurrentHashMap<>();
    /**
     * 用户客户端
     */
    public volatile static ConcurrentHashMap<String, Session> userWebSocketSet = new ConcurrentHashMap<>();


    /**
     * 会话key id
     */
    // private static String sessionKeyIdFormat = "%s_%s";
    /**
     * 会话key id
     */
    // private String sessionKeyId;

    /**
     * 与某个客户端的连接对话，需要通过它来给客户端发送消息
     */
    // private Session session;

    @OnOpen
    public void OnOpen(Session session) {
        onlineCount.incrementAndGet(); // 在线数加1
        webSocketSet.put(session.getId(),session);
        log.info("[WebSocket] 连接成功，当前连接数为：={}", webSocketSet.size());
    }


    @OnClose
    public void OnClose(Session session) {
        onlineCount.decrementAndGet(); // 在线数减1
        webSocketSet.remove(session.getId());
        log.info("[WebSocket] 退出成功，当前连接数为：={}", webSocketSet.size());
    }

    /**
     * 收到消息时执行
     *
     * @param message
     * @param session
     */
    @OnMessage
    public void onMessage(String message, Session session) {
        log.debug("来自客户端的消息:{}" + message);
        WebSocketReqDto activityParam = null;
        try {
            activityParam = JSONObject.parseObject(message, WebSocketReqDto.class);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("消息格式化出错，sessionId:" + session.getId() + ",message:" + message);
            // sendMessage(session, ResponseDTO.responseData500("消息格式化出错，sessionId:" + session.getId() + ",message:" + message));
        }
        if (activityParam != null && !StringUtils.isEmpty(activityParam.getOpenUserId()) && !StringUtils.isEmpty(activityParam.getKfOpenId())
                && !StringUtils.isEmpty(activityParam.getCorpId())) {
            String sessionKey = activityParam.getCorpId()+ FlushRedisConstant.linkStr+activityParam.getKfOpenId()+FlushRedisConstant.linkStr+ activityParam.getOpenUserId();
            if (!userWebSocketSet.containsKey(sessionKey)) {
                userWebSocketSet.put(sessionKey, session);
            }
            sendMessage(session, ResponseDTO.defaultResponse("客户端接到消息sessionId:" + session.getId() + ",message:" + message));
        } else {
            // sendMessage(session, ResponseDTO.responseParam404("参数缺失，sessionId:" + session.getId() + ",message:" + message));
            sendMessage(session, ResponseDTO.defaultResponse("客户端接到消息，sessionId:" + session.getId() + ",message:" + message));
        }
    }


    /**
     * 连接错误时执行
     *
     * @param session
     * @param error
     */
    @OnError
    public void onError(Session session, Throwable error) {
        log.debug("sessionId：{}的连接发送错误", session.getId());
    }

    /**
     * 发送消息
     *
     * @param session
     * @param responseDTO
     * @return
     */
    public static boolean sendMessage(Session session, ResponseDTO responseDTO) {
        String res = JSONObject.toJSONString(responseDTO, SerializerFeature.WriteDateUseDateFormat);
        session.getAsyncRemote().sendText(res);
        return true;
    }

    /**
     * 发送消息
     *
     * @param session
     * @param data
     * @return
     */
    public static boolean sendMessage(Session session, Object data) {
        ResponseDTO responseDTO = ResponseDTO.defaultResponse(data);
        String res = JSONObject.toJSONString(responseDTO, SerializerFeature.WriteDateUseDateFormat);
        session.getAsyncRemote().sendText(res);
        return true;
    }
    /**
     * 发送消息
     *
     * @param userId
     * @param responseDTO
     * @return
     */
    public static boolean sendMessage(String userId, ResponseDTO responseDTO) {
        String res = JSONObject.toJSONString(responseDTO, SerializerFeature.WriteDateUseDateFormat);
        if (StringUtils.isEmpty(userId)){
            log.error("消息发送出错，userId为空，responseDTO:", res);
            return false;
        }
        Session session = userWebSocketSet.get(userId);
        if (null == session){
            log.error("消息发送出错，session为空，responseDTO:", res);
            return false;
        }
        session.getAsyncRemote().sendText(res);
        return true;
    }
    /**
     * 发送消息
     *
     * @param userId
     * @param data
     * @return
     */
    public static boolean sendMessage(String userId,  Object data) {
        ResponseDTO responseDTO = ResponseDTO.defaultResponse(data);
        String res = JSONObject.toJSONString(responseDTO, SerializerFeature.WriteDateUseDateFormat);
        if (StringUtils.isEmpty(userId)){
            log.error("消息发送出错，userId为空，responseDTO:", res);
            return false;
        }
        Session session = userWebSocketSet.get(userId);
        if (null == session){
            log.error("消息发送出错，session为空，responseDTO:", res);
            return false;
        }
        session.getAsyncRemote().sendText(res);
        return true;
    }

}

2.3 公共类：

2.3.1 ResponseDTO：

import lombok.Data;
import lombok.experimental.Accessors;

import java.util.ArrayList;
import java.util.List;

@Data
@Accessors(chain = true)
public class ResponseDTO<T> {

  private static final long serialVersionUID = 3918877423924837166L;

  private int code = 200;
  private T body;
  private String msg;
  private boolean redirect = true;
  private List<ResponseMessage> messages;
  private List<String> message = new ArrayList<>();
  private int errNum;
  private String errorMsg;

  private boolean success = true;

  public static ResponseDTO response400() {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_400.getStatus());
  }

  public static ResponseDTO response401() {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_401.getStatus());
  }

  public static ResponseDTO response403() {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_403.getStatus());
  }

  public static ResponseDTO response404() {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_404.getStatus());
  }

  public static ResponseDTO response403(List<ResponseMessage> messages) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_403.getStatus()).setMessages(messages);
  }

  public static ResponseDTO response405(List<ResponseMessage> messages) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_405.getStatus()).setMessages(messages);
  }

  public static ResponseDTO response406(List<ResponseMessage> messages) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_406.getStatus()).setMessages(messages);
  }

  public static ResponseDTO response500() {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_500.getStatus());
  }

  public static ResponseDTO responseParam404(String msg) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_404.getStatus()).setSuccess(false).setMsg(msg);
  }

  public static ResponseDTO responseData500(String msg) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_500.getStatus()).setSuccess(false).setMsg(msg);
  }
  public static ResponseDTO responseData500(String msg, String errorMsg) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_500.getStatus()).setSuccess(false).setMsg(msg).setErrorMsg(errorMsg);
  }

  public static <T> ResponseDTO defaultResponse(T t) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_200.getStatus()).setBody(t);
  }

  public static ResponseDTO defaultErrorResponse(int httpStatus, List<ResponseMessage> singleMessage) {
      return new ResponseDTO().setCode(httpStatus).setMessages(singleMessage);
  }

  public static ResponseDTO defaultErrorResponse(List<ResponseMessage> singleMessage) {
      return new ResponseDTO().setCode(BizHttpStatus.HTTP_STATUS_200.getStatus()).setMessages(singleMessage);
  }

  public static ResponseDTO defaultErrorResponse(BizHttpStatus bizHttpStatus, List<ResponseMessage> singleMessage) {
      return new ResponseDTO().setSuccess(false).setCode(bizHttpStatus.getStatus()).setMessages(singleMessage);
  }


  public static ResponseDTO errorResponse(BizHttpStatus bizHttpStatus, List<ResponseMessage> singleMessage) {
      return new ResponseDTO().setCode(bizHttpStatus.getStatus()).setMessages(singleMessage);
  }

  public static ResponseDTO errorResponse(BizErrorEnum bizErrorEnum) {
      return new ResponseDTO().setCode(bizErrorEnum.getErrorCode()).setMsg(bizErrorEnum.getErrorMessage()).setSuccess(false);
  }

  public static ResponseDTO errorResponse(BizErrorEnum bizErrorEnum, List<String> message) {
      return new ResponseDTO().setCode(bizErrorEnum.getErrorCode()).setMsg(bizErrorEnum.getErrorMessage()).setMessage(message).setSuccess(false);
  }

}

2.3.2 FlushRedisConstant：

public class FlushRedisConstant {
public static  final String linkStr = "#thisislinkstr#";
}

2.3.3 WebSocketReqDto：

import lombok.Data;

import java.io.Serializable;
@Data
public class WebSocketReqDto implements Serializable {
    /**
     * 用户id
     */
    private String openUserId;
    private String kfOpenId;
    private String corpId;
    /**
     * 本次请求唯一标识 uuid
     */
    private String uuid;
}

2.3.4 在resource 下新建static 文件夹 放入测试的index.html 页面

<!DOCTYPE HTML>
<html>
<head>
    <title>My WebSocket</title>
</head>

<body>
<input id="text" type="text" />
<button onclick="send()">Send</button>
<button onclick="closeWebSocket()">Close</button>
<div id="message"></div>
</body>

<script type="text/javascript">
    var websocket = null;

    //判断当前浏览器是否支持WebSocket, 主要此处要更换为自己的地址
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://localhost:port/websocket/wxkf/activity");
    } else {
        alert('Not support websocket')
    }

    //连接发生错误的回调方法
    websocket.onerror = function() {
        setMessageInnerHTML("error");
    };

    //连接成功建立的回调方法
    websocket.onopen = function(event) {
        console.log("连接成功建立");
        //setMessageInnerHTML("open");
    }

    //接收到消息的回调方法
    websocket.onmessage = function(event) {
        setMessageInnerHTML(event.data);
    }

    //连接关闭的回调方法
    websocket.onclose = function() {
        setMessageInnerHTML("close");
    }

    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function() {
        websocket.close();
    }

    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }

    //关闭连接
    function closeWebSocket() {
        websocket.close();
    }

    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>

2.3.5 通过http://localhost:端口/index.html 进行测试

3 websocket spring-cloud+gateway 实现和boot 项目相同：
3.1 出现返回值非200 情况：

检查是否被全局的filter 做了拦截
（1）在spring-cloud gateway 中添加监控 Pom.xml 增加actuator的依赖：

<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-actuator</artifactId>
</dependency>

在spring-cloud gateway resources文件夹下的配置文件bootstrap.yml中放开对gateway 的监控（如果resources文件夹下没有bootstrap.yml 则手动添加bootstrap.yml）

management:
  endpoints:
    web:
      exposure:
        include:
          - gateway

http://ip:端口/actuator/gateway/globalfilters 检查全局filter，检查自定义的一些filter是否对改请求进行了拦截；

如果发现某个filter 对websocket 请求进行了拦截，则可以加入一下代码：

 String upgrade = exchange.getRequest().getHeaders().getUpgrade();
        if ("websocket".equals(upgrade)) {
            return chain.filter(exchange);
        }