Netty学习笔记(一)初遇篇

<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.56.Final</version>
</dependency>
复制代码

public class NettyDemo01 {
 public static void main(String[] args) throws InterruptedException {
        // 创建事件循环组  接收连接
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        // 接收连接 并分发给worker线程
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        // 启动Netty
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        ChannelFuture channelFuture
                = serverBootstrap.group(bossGroup, workerGroup).
                channel(NioServerSocketChannel.class).
                childHandler(new NettyServerInitHandler()).bind(8080).sync();
        // Wait until the server socket is closed.
        // 等待直到服务端的socket关闭,在这个例子中，服务端的Socket永远不会关闭，但是你可以优雅的关闭
        // 你的服务
        // In this example, this does not happen, but you can do that to gracefully
        // shut down your server.       
        channelFuture.channel().closeFuture().sync();
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
  }
// 综合初始化器
public class NettyServerInitHandler extends ChannelInitializer<SocketChannel> {
    // 通道被初次注册执行此方法等价于连接被建立
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 将处理器加入
        pipeline.addLast("HttpServerCodec",new HttpServerCodec());
        // 将自定义的处理加入
        pipeline.addLast("NettyServerHandler",new NettyServerHandler());
    }
}
public class NettyServerHandler extends SimpleChannelInboundHandler<HttpObject> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        //如果消息的类型是Http请求,才处理
        if (msg instanceof HttpRequest){
            // 统一编码
            ByteBuf content = Unpooled.copiedBuffer("hello world", CharsetUtil.UTF_8);
            // 发送HTTP响应
            DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
            // 设置请求头 
            response.headers().set(HttpHeaderNames.CONTENT_TYPE,"text/plain");
            // 
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH,content.readableBytes());
            // 返回响应
            ctx.writeAndFlush(response);
        }
    }
}    
复制代码

Selector selector = Selector.open();
    //创建服务端的通道
    ServerSocketChannel serverSocketChannel1 = ServerSocketChannel.open();
    // 将该通道设置为非阻塞方式
    serverSocketChannel1.configureBlocking(false);
    // 创建Socket
    ServerSocket serverSocket = serverSocketChannel1.socket();
    // 绑定端口
    serverSocket.bind(new InetSocketAddress(8888));
    // 为ServerSocketChannel注册对对客户端请求连接事件感兴趣
    // 此时该channel处于选择器得管理之下
    serverSocketChannel1.register(selector, SelectionKey.OP_ACCEPT);
复制代码

/**
    * Gets called after the {@link ChannelHandler} was added to the actual context 
      and it's ready to handle events.
        在对应的ChannelHandler(处理器)被加入到实际的上下文，
          该处理器准备去处理一些事件时      被调用。
     *
     */
   void handlerAdded(ChannelHandlerContext ctx) throws Exception;
复制代码

public class NettyServerHandler extends SimpleChannelInboundHandler<HttpObject> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        if (msg instanceof HttpRequest) {
            HttpRequest httpRequest = (HttpRequest) msg;
            URI uri = new URI(httpRequest.uri());
            // 浏览器请求的时候会请求两次,会默认请求一次icon
            // 这里先不处理这次请求。
            if (!"/favicon.ico".equals(uri.getPath())) {
                System.out.println(httpRequest.method().name());
                ByteBuf content = Unpooled.copiedBuffer("hello world", CharsetUtil.UTF_8);
                DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
                response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
                response.headers().set(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
                ctx.writeAndFlush(response);
            }
        }
    }
     // 增加处理时,自动触发
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("1.增加处理器");
        super.handlerAdded(ctx);
    }
    // 当把通道注册到选择器触发
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("2.通道被注册");
        super.channelRegistered(ctx);
    }
    // 通道被激活,触发另一个方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("3.通道被激活");
        super.channelActive(ctx);
    }
    // 激活的通道,失去激活状态
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("4.通道失活");
        super.channelInactive(ctx);
    }
    // 通道取消注册
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("5.通道取消注册");
        super.channelUnregistered(ctx);
    }
    // 失败捕获
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("6.异常捕获");
        super.exceptionCaught(ctx, cause);
    }
}
复制代码

public class MyNettyServerInit extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        // 加入拆分
        pipeline.addLast("decoder",new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,0,8,0,8));
        // 加入合并
        pipeline.addLast("prepender",new LengthFieldPrepender(8));
        // 加入解码 
        pipeline.addLast("StringDecoder",new StringDecoder(CharsetUtil.UTF_8));
        // 加入编码
        pipeline.addLast("StringEncoder",new StringEncoder(CharsetUtil.UTF_8));
        // 加入自定义的处理器
        pipeline.addLast("MyNettyServerHandler",new MyNettyServerHandler());
    }
}
复制代码

public LengthFieldBasedFrameDecoder(int maxFrameLength, 
int lengthFieldOffset, 
int lengthFieldLength,
int lengthAdjustment, 
int initialBytesToStrip)
复制代码

public class MyNettyServerHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        //接收消息
        System.out.println("服务端 接收到了" + ctx.channel().remoteAddress() + ",消息是：" + msg);
        //发送消息
        System.out.println("请输入内容：");
        String send = new Scanner(System.in).nextLine();
        ctx.channel().writeAndFlush(send);
    }
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush("开始聊天吧...");
    }
}
复制代码

public class MyNettyClient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        // 客户端,所以不是ServerBootStrap
        Bootstrap bootstrap = new Bootstrap();
        ChannelFuture channelFuture = bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).
                handler(new MyNettyClientInit()).connect("127.0.0.1", 8080).sync();
        channelFuture.channel().closeFuture().sync();
        eventLoopGroup.shutdownGracefully();
    }
}
复制代码

public class MyNettyClientInit extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 8));
        pipeline.addLast("pretender", new LengthFieldPrepender(8));
        pipeline.addLast("StringDecoder", new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast("StringEncoder", new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast("MyNettyClientHandler", new MyNettyClientHandler());
    }
}
复制代码

public class MyNettyClientHandler extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println("客户端 接收到了消息:  " + ctx.channel().remoteAddress() + ",消息是" + msg);
        System.out.println("请输入内容:");
        String send = new Scanner(System.in).nextLine();
        ctx.channel().writeAndFlush(send);
    }
}
复制代码

public IdleStateHandler( int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds)
复制代码

public class MyNettyServerInit extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("idleStateEvent", new IdleStateHandler(2,3,6));
        pipeline.addLast("myIdleStatteHandler",new MyIdleStateHandler());
    }
}
public class MyIdleStateHandler extends SimpleChannelInboundHandler<Object> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
    }
    // 读超时 或 写超时触发该方法
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            String eventType = null;
            IdleStateEvent event = (IdleStateEvent)evt;
            switch (event.state()){
                case READER_IDLE:
                    // 客户端发送的两次请求间隔超过两秒触发
                    eventType = "读超时";
                    break;
                case WRITER_IDLE:
                   // 客户端发送的两次请求间隔超过三秒触发
                    eventType = "写超时";
                    break;
                case ALL_IDLE:
                    // 六秒内读或写都没触发，认为读写超时
                    eventType = "读写超时";
                    break;
            }
            System.out.println(eventType);
            ctx.channel().close();
        }
    }
}
复制代码

public class NettyServerInitHandler extends ChannelInitializer<SocketChannel> {
    // 通道被注册执行此方法
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast("HttpServerCodec",new HttpServerCodec());
        // 我们用POST方式请求服务器的时候，对应的参数信息是保存在message body中的,如果只是单纯的用HttpServerCodec
        // 是无法完全的解析Http POST请求的，因为HttpServerCodec只能获取uri中参数，所以需要加上
        // HttpObjectAggregator.
        pipeline.addLast("HttpObjectAggregator", new HttpObjectAggregator(4096));
        // 设定地址
        pipeline.addLast("WebSocketServerProtocolHandler",new WebSocketServerProtocolHandler("/myWebSocket"));
        pipeline.addLast("NettyServerHandler",new MyWebSocketServerHandler());
    }
}
public class MyWebSocketServerHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        System.out.println("server 接收到的客户端消息: " + msg.text());
        ctx.channel().writeAndFlush(new TextWebSocketFrame("hello client"));
    }
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("连接建立....");
    }
    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("连接取消建立....");
    }
}
复制代码

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
    <script type="text/javascript">
        // websocket是ws开头
        var webSocket = new WebSocket("ws://localhost:8080/myWebSocket");
        function sendMessage(msg) {
            if (webSocket.readyState == webSocket.OPEN)
                webSocket.send(msg);
        }
        webSocket.onmessage = function (event) {
            document.getElementById("tip").innerText = "接收的消息: " + event.data;
        }
        webSocket.onopen = function (event) {
            document.getElementById("tip").innerText = "连接开启";
        }
        webSocket.onclose = function (event) {
            document.getElementById("tip").innerText = "连接关闭";
        }
    </script>
</head>
<body>
<form>
              <textarea name="message">
              </textarea>
    <input type="button" onclick="sendMessage(this.form.message.value)" value="发送">
</form>
<div id="tip"></div>
</body>
</html>
复制代码