基于netty的websocket例子

nettyServer

package com.atguigu.netty.websocket;

import javax.annotation.PostConstruct;

import org.springframework.stereotype.Service;

import io.netty.bootstrap.ServerBootstrap;

import io.netty.channel.Channel;

import io.netty.channel.ChannelInitializer;

import io.netty.channel.EventLoopGroup;

import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;

import io.netty.channel.socket.nio.NioServerSocketChannel;

import io.netty.handler.codec.http.HttpObjectAggregator;

import io.netty.handler.codec.http.HttpServerCodec;

import io.netty.handler.stream.ChunkedWriteHandler;

/**

* ClassName:NettyServer 注解式随spring启动

* Function: TODO ADD FUNCTION.

* @author fwz

*/

@Service

public class NettyServer {

public static void main(String[] args) {

new NettyServer().run();

}

@PostConstruct

public void initNetty(){

new Thread(){

public void run() {

new NettyServer().run();

}

}.start();

}

public void run(){

System.out.println("===========================Netty端口启动========");

// Boss线程：由这个线程池提供的线程是boss种类的，用于创建、连接、绑定socket， （有点像门卫）然后把这些socket传给worker线程池。

// 在服务器端每个监听的socket都有一个boss线程来处理。在客户端，只有一个boss线程来处理所有的socket。

EventLoopGroup bossGroup = new NioEventLoopGroup();

// Worker线程：Worker线程执行所有的异步I/O，即处理操作

EventLoopGroup workGroup = new NioEventLoopGroup();

try {

// ServerBootstrap 启动NIO服务的辅助启动类,负责初始话netty服务器，并且开始监听端口的socket请求

ServerBootstrap b = new ServerBootstrap();

b.group(bossGroup, workGroup);

// 设置非阻塞,用它来建立新accept的连接,用于构造serversocketchannel的工厂类

b.channel(NioServerSocketChannel.class);

// ChildChannelHandler 对出入的数据进行的业务操作,其继承ChannelInitializer

b.childHandler(new ChildChannelHandler());

System.out.println("服务端开启等待客户端连接 ... ...");

Channel ch = b.bind().sync().channel();

ch.closeFuture().sync();

} catch (Exception e) {

e.printStackTrace();

}finally{

bossGroup.shutdownGracefully();

workGroup.shutdownGracefully();

}

}

private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

@Override

protected void initChannel(SocketChannel e) throws Exception {

// 设置30秒没有读到数据，则触发一个READER_IDLE事件。

// pipeline.addLast(new IdleStateHandler(30, 0, 0));

// HttpServerCodec：将请求和应答消息解码为HTTP消息

e.pipeline().addLast("http-codec",new HttpServerCodec());

// HttpObjectAggregator：将HTTP消息的多个部分合成一条完整的HTTP消息

e.pipeline().addLast("aggregator",new HttpObjectAggregator());

// ChunkedWriteHandler：向客户端发送HTML5文件

e.pipeline().addLast("http-chunked",new ChunkedWriteHandler());

// 在管道中添加我们自己的接收数据实现方法

e.pipeline().addLast("handler",new MyWebSocketServerHandler());

}

}

}

MyWebSocketServerHandler

package com.atguigu.netty.websocket;

import java.util.Date;

import java.util.List;

import java.util.Map;

import java.util.logging.Level;

import java.util.logging.Logger;

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.HttpHeaderNames;

import io.netty.handler.codec.http.HttpResponseStatus;

import io.netty.handler.codec.http.HttpUtil;

import io.netty.handler.codec.http.HttpVersion;

import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;

import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;

import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;

import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import io.netty.handler.codec.http.websocketx.WebSocketFrame;

import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;

import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;

import io.netty.util.AttributeKey;

import io.netty.util.CharsetUtil;

/**

 * ClassName:MyWebSocketServerHandler Function: TODO ADD FUNCTION.

 *

 * @author fwz

 */

public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger logger = Logger.getLogger(WebSocketServerHandshaker.class.getName());

    private WebSocketServerHandshaker handshaker;

    /**

     * channel 通道 action 活跃的 当客户端主动链接服务端的链接后，这个通道就是活跃的了。也就是客户端与服务端建立了通信通道并且可以传输数据

     */

    @Override

    public void channelActive(ChannelHandlerContext ctx) throws Exception {

// 添加

        Global.group.add(ctx.channel());

        System.out.println("客户端与服务端连接开启：" + ctx.channel().remoteAddress().toString());

    }

    /**

     * channel 通道 Inactive 不活跃的

     * 当客户端主动断开服务端的链接后，这个通道就是不活跃的。也就是说客户端与服务端关闭了通信通道并且不可以传输数据

     */

    @Override

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

// 移除

        Global.group.remove(ctx.channel());

        System.out.println("客户端与服务端连接关闭：" + ctx.channel().remoteAddress().toString());

    }

    /**

     * 接收客户端发送的消息 channel 通道 Read 读

     * 简而言之就是从通道中读取数据，也就是服务端接收客户端发来的数据。但是这个数据在不进行解码时它是ByteBuf类型的

     */

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

// 传统的HTTP接入

        if (msg instanceof FullHttpRequest) {

            handleHttpRequest(ctx, ((FullHttpRequest) msg));

// WebSocket接入

        } else if (msg instanceof WebSocketFrame) {

            System.out.println(handshaker.uri());

            if ("anzhuo".equals(ctx.channel().attr(AttributeKey.valueOf("type")).get())) {

                handlerWebSocketFrame(ctx, (WebSocketFrame) msg);

            } else {

                handlerWebSocketFrame2(ctx, (WebSocketFrame) msg);

            }

        }

    }

    /**

     * channel 通道 Read 读取 Complete 完成 在通道读取完成后会在这个方法里通知，对应可以做刷新操作 ctx.flush()

     */

    @Override

    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

        ctx.flush();

    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {

// 判断是否关闭链路的指令

        if (frame instanceof CloseWebSocketFrame) {

            System.out.println();

            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());

            return;

        }

// 判断是否ping消息

        if (frame instanceof PingWebSocketFrame) {

            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));

            return;

        }

// 本例程仅支持文本消息，不支持二进制消息

        if (!(frame instanceof TextWebSocketFrame)) {

            System.out.println("本例程仅支持文本消息，不支持二进制消息");

            throw new UnsupportedOperationException(

                    String.format("%s frame types not supported", frame.getClass().getName()));

        }

// 返回应答消息

        String request = ((TextWebSocketFrame) frame).text();

        System.out.println("服务端收到：" + request);

        if (logger.isLoggable(Level.FINE)) {

            logger.fine(String.format("%s received %s", ctx.channel(), request));

        }

        TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);

// 群发

        Global.group.writeAndFlush(tws);

// 返回【谁发的发给谁】

// ctx.channel().writeAndFlush(tws);

    }

    private void handlerWebSocketFrame2(ChannelHandlerContext ctx, WebSocketFrame frame) {

// 判断是否关闭链路的指令

        if (frame instanceof CloseWebSocketFrame) {

            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());

            return;

        }

// 判断是否ping消息

        if (frame instanceof PingWebSocketFrame) {

            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));

            return;

        }

// 本例程仅支持文本消息，不支持二进制消息

        if (!(frame instanceof TextWebSocketFrame)) {

            System.out.println("本例程仅支持文本消息，不支持二进制消息");

            throw new UnsupportedOperationException(

                    String.format("%s frame types not supported", frame.getClass().getName()));

        }

// 返回应答消息

        String request = ((TextWebSocketFrame) frame).text();

        System.out.println("服务端2收到：" + request);

        if (logger.isLoggable(Level.FINE)) {

            logger.fine(String.format("%s received %s", ctx.channel(), request));

        }

        TextWebSocketFrame tws = new TextWebSocketFrame(new Date().toString() + ctx.channel().id() + "：" + request);

// 群发

        Global.group.writeAndFlush(tws);

// 返回【谁发的发给谁】

// ctx.channel().writeAndFlush(tws);

    }

    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {

// 如果HTTP解码失败，返回HHTP异常

        if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {

            sendHttpResponse(ctx, req,

                    new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));

            return;

        }

//获取url后置参数

        String uri = req.uri();

// 构造握手响应返回，本机测试

        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(

                "ws://localhost:8081/websocket" , null, false);

        handshaker = wsFactory.newHandshaker(req);

        if (handshaker == null) {

            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());

        } else {

            handshaker.handshake(ctx.channel(), req);

        }

    }

    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {

// 返回应答给客户端

        if (res.status().code() != ) {

            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);

            res.content().writeBytes(buf);

            buf.release();

        }

// 如果是非Keep-Alive，关闭连接

        ChannelFuture f = ctx.channel().writeAndFlush(res);

        if (!HttpUtil.isKeepAlive(req) || res.status().code() != ) {

            f.addListener(ChannelFutureListener.CLOSE);

        }

    }

    /**

     * exception 异常 Caught 抓住 抓住异常，当发生异常的时候，可以做一些相应的处理，比如打印日志、关闭链接

     */

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {

        cause.printStackTrace();

        ctx.close();

    }

    @Override

    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {

        // TODO Auto-generated method stub

    }

}

Global

package com.atguigu.netty.websocket;

import io.netty.channel.group.ChannelGroup;

import io.netty.channel.group.DefaultChannelGroup;

import io.netty.util.concurrent.GlobalEventExecutor;

/**

* ClassName:Global

* Function: TODO ADD FUNCTION.

* @author fwz

*/

public class Global {

public static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

}

controller

package com.atguigu.netty.websocket;

import org.springframework.stereotype.Controller;

import org.springframework.web.bind.annotation.RequestMapping;

@Controller

public class DeptController_Consumer

{

    @RequestMapping(value = "/socketHtml")

    public String html()

    {

        return"WebSocketServer";

    }

}

WebSocketServer

<html>

<head>

<meta charset="UTF-8">

Netty WebSocket 时间服务器

</head>

<br>

<body>

<br>

    <script type="text/javascript">

        var socket;

        if (!window.WebSocket) {

            window.WebSocket = window.MozWebSocket;

        }

        if (window.WebSocket) {

            socket = new WebSocket(

                    "ws://localhost:8081/websocket");

            socket.onmessage = function(event) {

                var ta = document.getElementById("responseText");

                ta.value="";

                ta.value=event.data;

            };

            socket.onopen = function(event) {

                var ta = document.getElementById("responseText");

                ta.value="";

                ta.value="打开WebSocket服务正常，浏览器支持WebSocket！";

            };

            socket.onclose = function(event) {

                var ta = document.getElementById("responseText");

                ta.value="";

                ta.value="WebSocket关闭！";

            };

        }else{

            alert("抱歉，您的浏览器不支持WebSocket")

        }

        function send(message) {

            if (!window.WebSocket) {

                return;

            }

            if (socket.readyState == WebSocket.OPEN) {

                socket.send(message);

            } else {

                alert("The socket is not open.");

            }

        }

    </script>

    <form onsubmit="return false;">

        <input type="text" name="message" value="Hello,Netty-WebSocket" />

        <br>

        <br>

         <input

            type="button" value="发送 WebSocket 请求消息"

            onclick="send(this.form.message.value)" />

            <hr color="blue"/>

            <h3>服务端返回的应答消息</h3>

            <textarea style="width: 500px;height: 300px;" id="responseText" ></textarea>

    </form>

</body>

</html>

启动后如下：

巴特西

基于netty的websocket例子

最新文章

热门文章