添加基于netty的nio实现的websocket

2018-02-03 16:14:58 +08:00
parent 536af49857
commit 83b47a2be7
12 changed files with 444 additions and 137 deletions
@@ -1,132 +1,132 @@
-package com.micro.fast.boot.starter.common.response;
-
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.annotation.JsonInclude;
-
-/**
- * 统一服务响应类,方法返回值
- * 序列化中不包含为空值的字段
- * @author lishouyu 18332763730@163.com 2017/10/02
- * @version 1.0
- * @since 1.0
- */
-@JsonIgnoreProperties(ignoreUnknown = true)
-@JsonInclude(JsonInclude.Include.NON_EMPTY)
-public class ServerResponse<T> {
-  /**
-   * 获取数据返回的提示数据字典码
-   */
-  private Integer code;
-  /**
-   * 获取数据成功或者失败的时候提示的消息
-   */
-  private String msg;
-  /**
-   * 返回的数据
-   */
-  T data;
-
-  /**
-   * 所有的构造方法均为私有，不允许外部直接调用构造方法
-   */
-  public ServerResponse() {
-
-  }
-
-  private ServerResponse(Integer code) {
-    this(code, null, null);
-  }
-
-  private ServerResponse(Integer code, String msg) {
-    this(code, msg, null);
-  }
-
-  /**
-   * 为了避免T 为string的时候调用的是 上面的构造方法，需要对构造方法进行包装。构造方法设置为私有，不允许外部调用
-   *
-   * @param code 　响应代码
-   * @param data 　相应数据
-   */
-  private ServerResponse(Integer code, T data) {
-    this(code, null, data);
-  }
-
-  private ServerResponse(Integer code, String msg, T data) {
-    this.code = code;
-    this.msg = msg;
-    this.data = data;
-  }
-
-  public Integer getCode() {
-    return code;
-  }
-
-  public void setCode(Integer code) {
-    this.code = code;
-  }
-
-  public String getMsg() {
-    return msg;
-  }
-
-  public void setMsg(String msg) {
-    this.msg = msg;
-  }
-
-  public T getData() {
-    return data;
-  }
-
-  public void setData(T data) {
-    this.data = data;
-  }
-
-  /**
-   * 返回代码是小于等于0的时候表示请求数据失败，或者进行验证时失败
-   *
-   * @return　是否成功
-   */
-  @JsonIgnore //使其不再json序列化结果之中
-  public boolean isSuccess() {
-    return BaseConst.ServerResponseCode.SUCCESS.getCode().intValue() == this.code;
-  }
-
-  //请求成功
-  public static <T> ServerResponse<T> success() {
-
-    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode());
-  }
-
-  public static <T> ServerResponse<T> successMsg(String msg) {
-
-    return new ServerResponse<T>(
-        BaseConst.ServerResponseCode.SUCCESS.getCode(),
-        msg);
-  }
-
-  public static <T> ServerResponse<T> successData(T data) {
-
-    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode(),
-        data);
-  }
-
-  public static <T> ServerResponse<T> successMsgData(String msg, T data) {
-
-    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode(),
-        msg, data);
-  }
-
-  //请求失败
-  public static <T> ServerResponse<T> error() {
-    return new ServerResponse<T>(BaseConst.ServerResponseCode.ERROR.getCode());
-  }
-
-  public static <T> ServerResponse<T> errorMsg(String msg) {
-    return new ServerResponse<T>(BaseConst.ServerResponseCode.ERROR.getCode(), msg);
-  }
-
-  public static <T> ServerResponse<T> errorCodeMsg(Integer code, String msg) {
-    return new ServerResponse<T>(code, msg);
-  }
-}
+package com.micro.fast.boot.starter.common.response;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
+/**
+ * 统一服务响应类,方法返回值
+ * 序列化中不包含为空值的字段
+ * @author lishouyu 18332763730@163.com 2017/10/02
+ * @version 1.0
+ * @since 1.0
+ */
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+public class ServerResponse<T> {
+  /**
+   * 获取数据返回的提示数据字典码
+   */
+  private Integer code;
+  /**
+   * 获取数据成功或者失败的时候提示的消息
+   */
+  private String msg;
+  /**
+   * 返回的数据
+   */
+  T data;
+
+  /**
+   * 所有的构造方法均为私有，不允许外部直接调用构造方法
+   */
+  public ServerResponse() {
+
+  }
+
+  private ServerResponse(Integer code) {
+    this(code, null, null);
+  }
+
+  private ServerResponse(Integer code, String msg) {
+    this(code, msg, null);
+  }
+
+  /**
+   * 为了避免T 为string的时候调用的是 上面的构造方法，需要对构造方法进行包装。构造方法设置为私有，不允许外部调用
+   *
+   * @param code 　响应代码
+   * @param data 　相应数据
+   */
+  private ServerResponse(Integer code, T data) {
+    this(code, null, data);
+  }
+
+  private ServerResponse(Integer code, String msg, T data) {
+    this.code = code;
+    this.msg = msg;
+    this.data = data;
+  }
+
+  public Integer getCode() {
+    return code;
+  }
+
+  public void setCode(Integer code) {
+    this.code = code;
+  }
+
+  public String getMsg() {
+    return msg;
+  }
+
+  public void setMsg(String msg) {
+    this.msg = msg;
+  }
+
+  public T getData() {
+    return data;
+  }
+
+  public void setData(T data) {
+    this.data = data;
+  }
+
+  /**
+   * 返回代码是小于等于0的时候表示请求数据失败，或者进行验证时失败
+   *
+   * @return　是否成功
+   */
+  @JsonIgnore //使其不再json序列化结果之中
+  public boolean isSuccess() {
+    return BaseConst.ServerResponseCode.SUCCESS.getCode().intValue() == this.code;
+  }
+
+  //请求成功
+  public static <T> ServerResponse<T> success() {
+
+    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode());
+  }
+
+  public static <T> ServerResponse<T> successMsg(String msg) {
+
+    return new ServerResponse<T>(
+        BaseConst.ServerResponseCode.SUCCESS.getCode(),
+        msg);
+  }
+
+  public static <T> ServerResponse<T> successData(T data) {
+
+    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode(),
+        data);
+  }
+
+  public static <T> ServerResponse<T> successMsgData(String msg, T data) {
+
+    return new ServerResponse<T>(BaseConst.ServerResponseCode.SUCCESS.getCode(),
+        msg, data);
+  }
+
+  //请求失败
+  public static <T> ServerResponse<T> error() {
+    return new ServerResponse<T>(BaseConst.ServerResponseCode.ERROR.getCode());
+  }
+
+  public static <T> ServerResponse<T> errorMsg(String msg) {
+    return new ServerResponse<T>(BaseConst.ServerResponseCode.ERROR.getCode(), msg);
+  }
+
+  public static <T> ServerResponse<T> errorCodeMsg(Integer code, String msg) {
+    return new ServerResponse<T>(code, msg);
+  }
+}
@@ -0,0 +1,23 @@
+# boot-starter-netty项目说明
+## 几种io的对比
+io类型 | 通讯模型 | 描述 
+---|------------|----
+bio | 阻塞同步通讯 | 使用一个线程处理一个请求
+伪nio |阻塞同步通讯 | 使用线程池，使用n个线程处理m个请求
+nio | 非阻塞同步通讯 | 使用一个线程处理多个请求，难度大，不方便调试
+aio | 非阻塞异步通讯 | 难度大，不方便调试
+## jdk自带的nio和netty的对比
+- jdk自带的nio的api繁杂、工作量和工作难度大、入门门槛高需要精通多线程、jdk的nio存在bug会出现selector空轮训导致cpu跑满
+- netty商用成功。可扩展性强、入门门槛低、成熟稳定、性能强
+## websocket简单介绍
+### websocket建立连接
+客户端发起握手请求，服务端响应请求，建立连接。
+### websocket的生命周期
+#### 打开事件
+#### 消息事件
+#### 错误事件  
+- 能感知websocket所有方法的错误
+#### 关闭事件 
+- 服务端端关闭底层tcp连接
+- 客户端发起TCP close
+## 基于netty实现websocket协议
@@ -0,0 +1,39 @@
+<?xml version="1.0" encoding="UTF-8" standalone="no"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>boot-starter-netty</artifactId>
+	<packaging>jar</packaging>
+
+	<name>boot-starter-netty</name>
+	<description>Spring boot starter</description>
+
+	<parent>
+		<groupId>com.micro.fast</groupId>
+		<artifactId>boot-starter</artifactId>
+		<version>0.0.1-SNAPSHOT</version>
+	</parent>
+
+	<dependencies>
+		<!-- netty nio -->
+		<dependency>
+			<groupId>io.netty</groupId>
+			<artifactId>netty-all</artifactId>
+		</dependency>
+	</dependencies>
+
+	<build>
+		<plugins>
+			<plugin>
+				<groupId>org.apache.maven.plugins</groupId>
+				<artifactId>maven-compiler-plugin</artifactId>
+			</plugin>
+			<!--<plugin>-->
+			<!--<groupId>org.apache.maven.plugins</groupId>-->
+			<!--<artifactId>maven-javadoc-plugin</artifactId>-->
+			<!--</plugin>-->
+		</plugins>
+	<finalName>boot-starter-netty</finalName>
+</build>
+</project>
@@ -0,0 +1,22 @@
+package com.micro.fast.boot.starter.netty.config;
+
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * netty的全局配置类
+ * @author lsy
+ */
+@Getter
+@Setter
+public class NettyConfig {
+    /**
+     * 存储每一个客户端接入进来的channel对象
+     */
+    public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+
+
+}
@@ -0,0 +1 @@
+package com.micro.fast.boot.starter.netty;
@@ -0,0 +1,30 @@
+package com.micro.fast.boot.starter.netty.websocket;
+
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.handler.codec.http.HttpClientCodec;
+import io.netty.handler.codec.http.HttpObjectAggregator;
+import io.netty.handler.codec.http.HttpObjectEncoder;
+import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.stream.ChunkedWriteHandler;
+
+
+/**
+ * 初始化连接时的各个组件
+ * @author lsy
+ */
+public class WebSocketChannelHandler  extends ChannelInitializer<SocketChannel>{
+
+    /**
+     *
+     * @param socketChannel
+     * @throws Exception
+     */
+    @Override
+    protected void initChannel(SocketChannel socketChannel) throws Exception {
+            socketChannel.pipeline().addLast("http-codec",new HttpServerCodec());
+            socketChannel.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
+            socketChannel.pipeline().addLast("http-chunked",new ChunkedWriteHandler());
+            socketChannel.pipeline().addLast("handler",new WebSocketChannelHandler());
+    }
+}
@@ -0,0 +1,159 @@
+package com.micro.fast.boot.starter.netty.websocket;
+
+import com.micro.fast.boot.starter.netty.config.NettyConfig;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http.websocketx.*;
+
+import java.nio.charset.Charset;
+import java.util.Date;
+
+/**
+ * 接收/处理/响应客户端websocket请求的核心业务处理类
+ * @author lsy
+ */
+public class WebSocketHandler extends SimpleChannelInboundHandler<Object>{
+
+    private WebSocketServerHandshaker handshaker;
+
+    private static final String webSocketUrl = "ws://localhost:8888/websocket";
+    /**
+     * 客户端和服务端创建连接的时候调用
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        NettyConfig.group.add(ctx.channel());
+    }
+
+    /**
+     * 客户端和服务端断开连接的时候调用
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        NettyConfig.group.remove(ctx.channel());
+    }
+
+    /**
+     * 服务端接收客户端数据完成时候调用
+     * @param ctx
+     * @throws Exception
+     */
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        ctx.flush();
+    }
+
+    /**
+     * 工程出现异常的时候调用
+     * @param ctx
+     * @param cause
+     * @throws Exception
+     */
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        ctx.close();
+    }
+
+    /**
+     * 服务端处理客户端webSocket请求的核心方法
+     * @param channelHandlerContext
+     * @param msg
+     * @throws Exception
+     */
+    @Override
+    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
+        //处理客户端向服务端发起http握手请求的业务
+        if (msg instanceof FullHttpRequest){
+            handHttpRequestChannelHandler(channelHandlerContext,(FullHttpRequest)msg);
+        }else if(msg instanceof WebSocketFrame){
+            //处理websocket连接请求的业务
+            handWebSocketFrameChannelHandler(channelHandlerContext,(WebSocketFrame)msg);
+        }
+    }
+
+    /**
+     * 处理http请求
+     * @param ctx
+     * @param request
+     */
+    private void handHttpRequestChannelHandler(ChannelHandlerContext ctx,FullHttpRequest request){
+        if(request.getDecoderResult().isFailure() || "websocket".equals(request.headers().get("Upgrade"))){
+            sendHttpResponse(ctx,request,new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
+            return;
+        }else{
+            WebSocketServerHandshakerFactory webSocketServerHandshakerFactory = new WebSocketServerHandshakerFactory(
+                    webSocketUrl,null,false);
+            //创建webSocket连接
+            handshaker = webSocketServerHandshakerFactory.newHandshaker(request);
+            if (handshaker == null) {
+                WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
+            }else {
+                handshaker.handshake(ctx.channel(),request);
+            }
+        }
+    }
+
+    /**
+     * 服务端向客户端发送响应
+     * @param ctx
+     * @param request
+     * @param response
+     */
+    private void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest request
+            , DefaultFullHttpResponse response){
+        //判断请求是否成功
+        if (response.getStatus().code()!= 200){
+            // 返回请求
+            ByteBuf byteBuf = Unpooled.copiedBuffer(response.getStatus().toString(), Charset.forName("UTF-8"));
+            response.content().writeBytes(byteBuf);
+            byteBuf.release();
+        }
+        ChannelFuture channelFuture = ctx.channel().writeAndFlush(response);
+        if (response.getStatus().code()!=200){
+            // 关闭连接
+            channelFuture.addListener(ChannelFutureListener.CLOSE);
+        }
+    }
+
+    /**
+     * 处理客户端和服务端之间的webSocket业务
+     * @param ctx
+     * @param webSocketFrame
+     */
+    private void handWebSocketFrameChannelHandler(ChannelHandlerContext ctx,WebSocketFrame webSocketFrame){
+        // 判断是否是关闭websocket连接请求
+        if (webSocketFrame instanceof CloseWebSocketFrame){
+            handshaker.close(ctx.channel(),(CloseWebSocketFrame)webSocketFrame.retain());
+        }
+        // 判断是否是pin消息
+        if (webSocketFrame instanceof PingWebSocketFrame){
+            ctx.channel().write(new PongWebSocketFrame(webSocketFrame.content().retain()));
+            return;
+        }
+        // 判断是否是二进制消息，如果是二进制消息，抛出异常
+        if (!(webSocketFrame instanceof TextWebSocketFrame)){
+            throw  new RuntimeException("不支持二进制消息的传递");
+        }
+        // 返回消息
+        TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) (webSocketFrame);
+        //获取客户端向服务端发送的消息
+        String text = textWebSocketFrame.text();
+        TextWebSocketFrame textWebSocketFrame1 = new TextWebSocketFrame(new Date().toString()+":"+text);
+
+        //服务端向每个连接上来的客户端群发消息
+        NettyConfig.group.writeAndFlush(textWebSocketFrame1);
+
+    }
+}
@@ -0,0 +1,33 @@
+package com.micro.fast.boot.starter.netty.websocket;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+
+/**
+ * webSocket启动的入口
+ * @author lsy
+ */
+public class WebSocketMain {
+    public static void main(String[] args) {
+        EventLoopGroup  bossGroup = new NioEventLoopGroup();
+        NioEventLoopGroup workGroup = new NioEventLoopGroup();
+        try {
+            ServerBootstrap serverBootstrap = new ServerBootstrap();
+            serverBootstrap.group(bossGroup,workGroup);
+            serverBootstrap.channel(NioServerSocketChannel.class);
+            serverBootstrap.childHandler(new WebSocketChannelHandler());
+            System.out.println("服务端开启等待客户端连接");
+            Channel channel = serverBootstrap.bind(8888).sync().channel();
+            channel.closeFuture().sync();
+        }catch (Exception e){
+
+        }finally {
+            // 优雅的退出程序
+            bossGroup.shutdownGracefully();
+            workGroup.shutdownGracefully();
+        }
+    }
+}
@@ -0,0 +1 @@
+package com.micro.fast.boot.starter.netty.websocket;
@@ -18,6 +18,7 @@
 		<module>boot-starter-common</module>
 		<module>boot-starter-util</module>
 		<module>boot-starter-maven-plugin</module>
+		<module>boot-starter-netty</module>
 		<module>boot-starter-fastdfs-client</module>
 		<module>boot-starter-elasticsearch</module>
 		<module>boot-starter-security-oauth</module>
@@ -9,7 +9,6 @@ import org.springframework.http.HttpStatus;
 import org.springframework.stereotype.Component;

 import javax.servlet.http.HttpServletRequest;
-import javax.validation.Valid;

 /**
 * 对accessToken进行校验
@@ -42,10 +41,11 @@ public class PreRequestFilter extends ZuulFilter {
        HttpServletRequest request = requestContext.getRequest();
        ServerResponse validate = ValidateJwtToken.validate(request, signingKey);
        if (validate.isSuccess()) {
-            //对请求进行后续处理
+            //todo  对访问权限进行判断
+            // 对请求进行后续处理
            requestContext.setSendZuulResponse(true);
        }else {
-            //不进行后续操作,返回401状态码
+            // 不进行后续操作,返回401状态码
            requestContext.setSendZuulResponse(false);
            requestContext.setResponseStatusCode(HttpStatus.UNAUTHORIZED.value());
        }
@@ -18,8 +18,6 @@ import com.netflix.eureka.util.StatusInfo;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import org.apache.commons.lang.StringUtils;
-import org.apache.http.HttpStatus;
-import org.mortbay.jetty.Server;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.*;
				`@@ -0,0 +1 @@`
				`package com.micro.fast.boot.starter.netty.websocket;`