springboot集成websocket实现消息通信

springboot 集成websocket

引入jar包

  <!-- Spring boot websocket -->
<dependency>
   <groupId>org.springframework.boot</groupId>
   <artifactId>spring-boot-starter-websocket</artifactId>
</dependency>
复制代码

websocket配置文件

@Configuration
public class WebSocketConfig {

   @Bean
   public ServerEndpointExporter serverEndpointExporter() {
      return new ServerEndpointExporter();
   }
}
复制代码

工具

package me.zhengjie.modules.mnt.websocket;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.CopyOnWriteArraySet;
/**
 * @author tony
 * websocket 消息通信
 */
@ServerEndpoint("/webSocket/{sid}")
@Slf4j
@Component
public class WebSocketServer {

   /**
    * concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。
    */
   private static CopyOnWriteArraySet<WebSocketServer> webSocketSet = new CopyOnWriteArraySet<WebSocketServer>();

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

   /**
    * 接收sid
    */
   private String sid="";
   /**
    * 连接建立成功调用的方法
    * */
   @OnOpen
   public void onOpen(Session session,@PathParam("sid") String sid) {
      System.out.println("连接成功-----sid："+sid);
      this.session = session;
      //如果存在就先删除一个，防止重复推送消息
      for (WebSocketServer webSocket:webSocketSet) {
         if (webSocket.sid.equals(sid)) {
            webSocketSet.remove(webSocket);
         }
      }
      webSocketSet.add(this);
      this.sid=sid;
   }

   /**
    * 连接关闭调用的方法
    */
   @OnClose
   public void onClose() {
      webSocketSet.remove(this);
   }

   /**
    * 收到客户端消息后调用的方法
    * @param message 客户端发送过来的消息*/
   @OnMessage
   public void onMessage(String message, Session session) {
      log.info("收到来"+sid+"的信息:"+message);
      //群发消息
      for (WebSocketServer item : webSocketSet) {
         try {
            item.sendMessage(message);
         } catch (IOException e) {
            log.error(e.getMessage(),e);
         }
      }
   }

   @OnError
   public void onError(Session session, Throwable error) {
      log.error("发生错误");
      error.printStackTrace();
   }
   /**
    * 实现服务器主动推送
    */
   private void sendMessage(String message) throws IOException {
      System.out.println("---------发送消息------------");
      this.session.getBasicRemote().sendText(message);
   }

   public static void main(String[] args) throws IOException {
      SocketMsg socketMsg=new SocketMsg("有巨石",MsgType.INFO);
      sendInfo(socketMsg,"deploy");
   }

   /**
    * 群发自定义消息
    * */
   public static void sendInfo(SocketMsg socketMsg,@PathParam("sid") String sid) throws IOException {
      String message = JSONObject.toJSONString(socketMsg);
      log.info("推送消息到"+sid+"，推送内容:"+message);
      for (WebSocketServer item : webSocketSet) {
         try {
            //这里可以设定只推送给这个sid的，为null则全部推送
            if(sid==null) {
               item.sendMessage(message);
            }else if(item.sid.equals(sid)){
               item.sendMessage(message);
            }
         } catch (IOException ignored) { }
      }
   }

   @Override
   public boolean equals(Object o) {
      if (this == o) {
         return true;
      }
      if (o == null || getClass() != o.getClass()) {
         return false;
      }
      WebSocketServer that = (WebSocketServer) o;
      return Objects.equals(session, that.session) &&
            Objects.equals(sid, that.sid);
   }

   @Override
   public int hashCode() {
      return Objects.hash(session, sid);
   }
}
复制代码

前台vue

initWebSocket() {
  const wsUri = process.env.VUE_APP_WS_API + '/webSocket/deploy'
  this.websock = new WebSocket(wsUri)
  this.websock.onerror = this.webSocketOnError
  this.websock.onmessage = this.webSocketOnMessage
},
webSocketOnError(e) {
  this.$notify({
    title: 'WebSocket连接发生错误',
    type: 'error',
    duration: 0
  })
},
webSocketOnMessage(e) {
  const data = JSON.parse(e.data)
  this.$notify({
    title: '',
    message: data,
    type: 'success',
    dangerouslyUseHTMLString: true,
    duration: 5500
  })
  if (data.msgType === 'INFO') {
    this.$notify({
      title: '',
      message: data.msg,
      type: 'error',
      dangerouslyUseHTMLString: true,
      duration: 5500
    })
  } else if (data.msgType === 'ERROR') {
    this.$notify({
      title: '',
      message: data.msg,
      dangerouslyUseHTMLString: true,
      type: 'error',
      duration: 0
    })
  }
},
webSocketSend(agentData) {
  this.websock.send(agentData)
}
复制代码