Netty 用法介绍

观察同一个线程可以服务多个连接；
server端 handler和MVC的controller类比

服务端 Server

package com.masking.study.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class TimeServer {
    public static void main(String[] args) throws Exception {
        new TimeServer().start(8080);
    }
    
    public void start(int port) throws Exception {
        /**
         * 配置服务端NIO线程组
         * NioEventLoopGroup 是个线程组，包含一组NIO线程用于网络事件处理
         * 创建两个是因为：
         *      bossGroup：用于服务端接受客户端的连接
         *      workerGroup：用于进行SocketChannel的网络读写
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            /**
             * 创建ServerBootstrap对象
             * ServerBootstrap是Netty用于启动NIO服务端的辅助启动类，降低开发复杂度
             * 第一步 将两个线程组传入
             * 第二步 设置创建的Channel类型为 NioServerSocketChannel  对应Java NIO里的 ServerSocketChannel
             * 第三步 设置TCP参数 backlog  默认50
             *      服务器端TCP内核模块维护有2个队列，队列A和队列B
                    客户端向服务端connect的时候，发送带有SYN标志的包（第一次握手）
                    服务端收到客户端发来的SYN，向客户端发送SYN ACK 确认(第二次握手) 并把客户端连接加入到A队列中，
                    服务器收到客户端发来的ACK时（第三次握手） 把客户端连接从A队列移到B队列，连接完成；
                    服务端accept从B队列中取出上面已经完成三次握手的连接
             *      backlog对程序支持的连接数并无影响，backlog影响的只是还没有被accept取出的连接
             * 第四步 绑定事件处理类
             */
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new TimeServerHandler());
                    }
                });
            
            /**
             * 调用 bind方法绑定监听端口
             * 调用 sync方法等待绑定操作完成 返回ChannelFuture
             * 绑定端口，同步等待
             */
            ChannelFuture cf = bootstrap.bind("localhost",port).sync();
            System.out.println("Server已启动："+cf.channel().localAddress());
             
            // 等待服务端监听端口关闭
            cf.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端 Handler

package com.masking.study.netty.server;

import java.util.Date;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;

/**
 * TimeServerHandler继承自ChannelHandlerAdapter 用于对网络事件进行读写操作
 * 主要方法 channelRead、channelReadComplete、exceptionCaught
 *
 */
public class TimeServerHandler extends ChannelHandlerAdapter {
    
    /**
     * 收到客户端消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        /**
         * 将msg转为ByteBuf对象
         */
        ByteBuf buf = (ByteBuf) msg;
        byte[] req = new byte[buf.readableBytes()];
        buf.readBytes(req);
        String reqStr = new String(req, "UTF-8").trim();
        System.out.println(Thread.currentThread().getName()+" Server received: "+reqStr);
        if("get".equalsIgnoreCase(reqStr)) {
            ctx.write(Unpooled.copiedBuffer(("服务器时间:"+new Date().toString()+'\n').getBytes()));
        }else if("exit".equalsIgnoreCase(reqStr)) {
            ctx.close();
        }else {
            ctx.write(Unpooled.copiedBuffer("Invalid Request!\n".getBytes()));
        }
        
        /**
         * 为了防止频繁地唤醒 Selector, Netty的write方法不直接将信息写入SocketChannel，只是把信息放到缓冲数组中。
         * 需要调用flush方法将缓冲区信息写入SocketChannel
         */
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
    
}