服务端常用的WebSocket框架

码客说

发布于 2021-01-20 10:08:12

1.2K0

发布于 2021-01-20 10:08:12

文章被收录于专栏：码客码客

Maven依赖无法下载的解决方案

比如我要下载的包为tio-websocket-server

到中央仓库地址：http://mvnrepository.com/

搜索到的依赖为

<dependency>
  <groupId>org.t-io</groupId>
  <artifactId>tio-websocket-server</artifactId>
  <version>3.7.1.v20210106-RELEASE</version>
</dependency>

对应的地址为

https://mvnrepository.com/artifact/org.t-io/tio-websocket-server/3.7.1.v20210106-RELEASE

输入命令需要修改的 url、groupId、artifactId、version

mvn dependency:get -DremoteRepositories=url -DgroupId=groupId -DartifactId=artifactId -Dversion=version

即

mvn dependency:get -DremoteRepositories=https://mvnrepository.com/artifact/org.t-io/tio-websocket-server -DgroupId=org.t-io -DartifactId=tio-websocket-server -Dversion=3.7.1.v20210106-RELEASE

t-io Starter

t-io官方示例：https://gitee.com/psvmc/tio-websocket-showcase

https://www.tiocloud.com/doc/tio/318?pageNumber=1

https://github.com/fanpan26/tio-websocket-spring-boot-starter

<dependency>
  <groupId>com.github.fanpan26</groupId>
  <artifactId>tio-websocket-spring-boot-starter</artifactId>
  <version>1.0.0.RELEASE</version>
</dependency>

主类

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.tio.websocket.starter.EnableTioWebSocketServer;

@SpringBootApplication
@EnableTioWebSocketServer
public class WsServerZApplication {
    public static void main(String[] args) {
        SpringApplication.run(WsServerZApplication.class, args);
    }
}

主要就是添加了

@EnableTioWebSocketServer

消息监听

import org.springframework.stereotype.Component;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.http.common.HttpRequest;
import org.tio.http.common.HttpResponse;
import org.tio.websocket.common.WsRequest;
import org.tio.websocket.common.WsResponse;
import org.tio.websocket.server.handler.IWsMsgHandler;

@Component
public class MyWebSocketMsgHandler implements IWsMsgHandler {

    @Override
    public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
        return httpResponse;
    }

    @Override
    public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
        System.out.println("握手成功");
    }

    @Override
    public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
        System.out.println("接收到bytes消息");
        return null;
    }

    @Override
    public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
        System.out.println("接收二进制");
        return null;
    }

    @Override
    public Object onText(WsRequest wsRequest, String s, ChannelContext channelContext) throws Exception {
        String msg = "接收文本：" + s + " 客户端ID：" + channelContext.getId();
        System.out.println(msg);
        Tio.send(channelContext, WsResponse.fromText(msg, "utf-8"));
        return null;
    }
}

给对应的客户端发送消息

Tio.send(channelContext, WsResponse.fromText(msg, "utf-8"));

接下来即可在application.properties中配置

tio.websocket.server.port=8888
tio.websocket.cluster.enabled=false
tio.websocket.cluster.redis.ip=127.0.0.1
tio.websocket.cluster.redis.port=6379
tio.websocket.cluster.all=true
tio.websocket.cluster.group=true
tio.websocket.cluster.ip=true
tio.websocket.cluster.user=true

Netty Starter

Netty：https://gitee.com/Yeauty/netty-websocket-spring-boot-starter

基于Netty的Websocket的框架

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-autoconfigure</artifactId>
</dependency>
<dependency>
  <groupId>org.yeauty</groupId>
  <artifactId>netty-websocket-spring-boot-starter</artifactId>
  <version>0.9.5</version>
</dependency>

代码

首先在ServerEndpoint注解中设置

import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.timeout.IdleStateEvent;
import org.springframework.util.MultiValueMap;
import org.yeauty.annotation.*;
import org.yeauty.pojo.Session;

import java.io.IOException;
import java.util.Map;

@ServerEndpoint(path = "${ws.path}",host = "${ws.host}",port = "${ws.port}")
public class MyWebSocket {

    @BeforeHandshake
    public void handshake(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        session.setSubprotocols("stomp");
        if (!req.equals("ok")){
            System.out.println("Authentication failed!");
            session.close();
        }
    }

    @OnOpen
    public void onOpen(Session session, HttpHeaders headers, @RequestParam String req, @RequestParam MultiValueMap reqMap, @PathVariable String arg, @PathVariable Map pathMap){
        System.out.println("new connection");
        System.out.println(req);
    }

    @OnClose
    public void onClose(Session session) throws IOException {
        System.out.println("one connection closed");
    }

    @OnError
    public void onError(Session session, Throwable throwable) {
        throwable.printStackTrace();
    }

    @OnMessage
    public void onMessage(Session session, String message) {
        System.out.println(message);
        session.sendText("Hello Netty!");
    }

    @OnBinary
    public void onBinary(Session session, byte[] bytes) {
        for (byte b : bytes) {
            System.out.println(b);
        }
        session.sendBinary(bytes);
    }

    @OnEvent
    public void onEvent(Session session, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
            switch (idleStateEvent.state()) {
                case READER_IDLE:
                    System.out.println("read idle");
                    break;
                case WRITER_IDLE:
                    System.out.println("write idle");
                    break;
                case ALL_IDLE:
                    System.out.println("all idle");
                    break;
                default:
                    break;
            }
        }
    }

}

接下来即可在application.properties中配置

ws.path=/ws
ws.host=0.0.0.0
ws.port=8888

SpringBoot Starter

这是SpringBoot官方提供的Webscoket框架

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

具体代码

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 serverEndpoint() {
        return new ServerEndpointExporter();
    }
}

WsServerEndpoint

import java.io.IOException;

import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;

@ServerEndpoint("/ws")
@Component
public class MyWebSocket {

    /**
     * 连接成功
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        System.out.println("连接成功");
    }

    /**
     * 连接关闭
     *
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
        System.out.println("连接关闭");
    }

    @OnError
    public void onError(Session session, Throwable error) {
        System.out.println("发生错误");
        error.printStackTrace();
    }

    /**
     * 接收到消息
     *
     * @param text
     */
    @OnMessage
    public String onMessage(String text, Session session) throws IOException {
        return "servet 发送：" + text;
    }

    @OnMessage
    public void onMessage(byte[] messages, Session session) throws IOException {

    }
}

说明：

这里有几个注解需要注意一下，首先是他们的包都在 javax.websocket 下。并不是 spring 提供的，而 jdk 自带的，下面是他们的具体作用。

@ServerEndpoint 通过这个 spring boot 就可以知道你暴露出去的 ws 应用的路径，有点类似我们经常用的@RequestMapping。比如你的启动端口是8080，而这个注解的值是ws，那我们就可以通过 ws://127.0.0.1:8888/ws 来连接你的应用
@OnOpen 当 websocket 建立连接成功后会触发这个注解修饰的方法，注意它有一个 Session 参数
@OnClose 当 websocket 建立的连接断开后会触发这个注解修饰的方法，注意它有一个 Session 参数
@OnMessage 当客户端发送消息到服务端时，会触发这个注解修改的方法，它有一个 String 入参表明客户端传入的值
@OnError 当 websocket 建立连接时出现异常会触发这个注解修饰的方法，注意它有一个 Session 参数

另外一点就是服务端如何发送消息给客户端，服务端发送消息必须通过上面说的 Session 类，通常是在@OnOpen 方法中，当连接成功后把 session 存入 Map 的 value，key 是与 session 对应的用户标识，当要发送的时候通过 key 获得 session 再发送，这里可以通过 session.getBasicRemote().sendText() 来对客户端发送消息。

接下来即可在application.properties中配置

server.port=8888

客户端测试页面

页面

<!DOCTYPE html>
<html>
  <head>
    <meta name="viewport" content="width=device-width, initial-scale=1" />
    <title>聊天客户端</title>
  </head>

  <body>
    <div>
      <div id="content"></div>
      <input type="text" style="width: 100%" id="msg" />
      <button type="button" onclick="emit()">发送</button>
    </div>

    <script
      type="text/javascript"
      src="http://apps.bdimg.com/libs/jquery/2.1.4/jquery.min.js"
    ></script>

    <script type="text/javascript" src="js/index.js"></script>
  </body>
</html>

var socket = new WebSocket("ws://127.0.0.1:8888/ws");

$(function () {
  listen();
})

function encodeScript(data) {
  if (null == data || "" == data) {
    return "";
  }
  return data.replace("<", "&lt;").replace(">", "&gt;");
}

function emit() {
  var text = encodeScript($("#msg").val());
  text = replace_em(text);
  socket.send(text);

  $("#content").append("<kbd style='color: #" + "CECECE" + "; font-size: " + 12 + ";'>" + text + "</kbd><br/>");
  $("#msg").val("");
}

//替换为HTML上的标签
function replace_em(str) {
  str = str.replace(/\</g, '&lt;');
  str = str.replace(/\>/g, '&gt;');
  str = str.replace(/\n/g, '<br/>');
  str = str.replace(/\[em_([0-9]*)\]/g, '<img src="arclist/$1.gif" border="0" />');
  return str;
};

function listen() {
  socket.onopen = function () {
    $("#content").append("<kbd>连接成功! 时间(s):" + parseInt(new Date().getTime() / 1000) + "</kbd></br>");
    heartCheck();
  };
  socket.onmessage = function (evt) {
    $("#content").append(evt.data + "</br>");
  };
  socket.onclose = function (evt) {
    $("#content").append("<kbd>" + "连接关闭! 时间(s):" + parseInt(new Date().getTime() / 1000) + "</kbd></br>");
  }
  socket.onerror = function (evt) {
    $("#content").append("<kbd>" + "ERROR!" + "</kbd></br>");
  }
}

//心跳包
function heartCheck() {
  setInterval(function () {
    if (socket) {
      let buffer = new ArrayBuffer(2); // 初始化14个Byte的二进制数据缓冲区
      let dataView = new DataView(buffer);
      dataView.setInt16(0, 1);
      socket.send(dataView);
      console.info("发送心跳", " 时间(s):" + parseInt(new Date().getTime() / 1000));
    }
  }, 30000);
}

document.onkeydown = function (event) {
  var e = event || window.event || arguments.callee.caller.arguments[0];
  if (e && e.keyCode == 13) { // enter 键
    emit();
  }
};

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2021-01-15，如有侵权请联系 cloudcommunity@tencent.com 删除

https