關於Netty的入門使用

Netty介紹:

Netty是一個提供異步事件驅動的網絡應用框架，用以快速開發高性能、高可靠性的網絡服務器和客戶端程序。

換句話說，Netty是一個NIO框架，使用它可以簡單快速地開發網絡應用程序，比如客戶端和服務端的協議。Netty大大簡化了網絡程序的開發過程比如TCP和UDP的 Socket的開發。

“快速和簡單”並不意味着應用程序會有難維護和性能低的問題，Netty是一個精心設計的框架，它從許多協議的實現中吸收了很多的經驗比如FTP、SMTP、HTTP、許多二進制和基於文本的傳統協議，Netty在不降低開發效率、性能、穩定性、靈活性情況下，成功地找到了解決方案。

有一些用戶可能已經發現其他的一些網絡框架也聲稱自己有同樣的優勢，所以你可能會問是Netty和它們的不同之處。答案就是Netty的哲學設計理念。Netty從第一天開始就為用戶提供了用戶體驗最好的API以及實現設計。正是因為Netty的設計理念，才讓我們得以輕松地閱讀本指南並使用Netty。

接下來，我們看下Client端的代碼實現:

 

package ruizhan.hjf.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.Unpooled;
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.NioSocketChannel;
/**
 * Netty客戶端的程序
 * @author huangjianfei
 */
public class Client {
    /*IP地址*/
    static final String HOST = System.getProperty("host", "127.0.0.1");
    /*端口號*/
    static final int PORT1 = Integer.parseInt(System.getProperty("port", "8765"));
    
    static final int PORT2 = Integer.parseInt(System.getProperty("port", "8764"));
    
    public static void main(String[] args) throws Exception {
        EventLoopGroup workgroup = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();//客戶端
        b.group(workgroup)
        .channel(NioSocketChannel.class)//客戶端 -->NioSocketChannel
        .option(ChannelOption.SO_KEEPALIVE, true)
        .handler(new ChannelInitializer<SocketChannel>() {//handler
            @Override
            protected void initChannel(SocketChannel sc) throws Exception {
                sc.pipeline().addLast(new ClientHandler());  
            }
        });  
        //創建異步連接 可添加多個端口
        ChannelFuture cf1 = b.connect(HOST, PORT1).sync();
        ChannelFuture cf2 = b.connect(HOST, PORT2).sync();   
        
        //buf
        //client向server端發送數據  Buffer形式
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("hello netty".getBytes()));
        cf2.channel().writeAndFlush(Unpooled.copiedBuffer("hello world".getBytes()));
        
        
        cf1.channel().closeFuture().sync();
        cf2.channel().closeFuture().sync();
        
        workgroup.shutdownGracefully();
    }
}

Servler端代碼實現:

package ruizhan.hjf.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
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;
/**
 * Netty實現的服務端程序
 * @author huangjianfei
 */
public class Server
{
    /*端口號*/
    static final int PORT1 = Integer.parseInt(System.getProperty("port", "8765"));
    
    static final int PORT2 = Integer.parseInt(System.getProperty("port", "8764"));
    
    public static void main(String[] args)
    {
        EventLoopGroup bossGroup = null;
        EventLoopGroup workerGroup = null;
        ServerBootstrap b = null;
        try{
            //1:第一個線程組是用於接收Client連接的
            bossGroup = new NioEventLoopGroup(); //(1)
            //2:第二個線程組是用於實際的業務處理操作的
            workerGroup = new NioEventLoopGroup();
            //3:創建一個啟動NIO服務的輔助啟動類ServerBootstrap 就是對我們的Server進行一系列的配置
            b = new ServerBootstrap();//(2)
            //4:綁定兩個線程組
            b.group(bossGroup, workerGroup)
            //5:需要指定使用NioServerSocketChannel這種類型的通道
            .channel(NioServerSocketChannel.class)//(3) 服務端 -->NioServerSocketChannel
            //6:一定要使用childHandler 去綁定具體的事件處理器
            .childHandler(new ChannelInitializer<SocketChannel>() //(4)   childHandler
            {
                @Override
                protected void initChannel(SocketChannel sc) throws Exception
                {
                    //7:將自定義的serverHandler加入到管道中去（多個）
                    sc.pipeline().addLast(new ServerHandler());//handler中實現真正的業務邏輯
//                    sc.pipeline().addLast(new ServerHandler());
//                    sc.pipeline().addLast(new ServerHandler());
                }
            })
            /**
             * 服務器端TCP內核模塊維護兩個隊列，我們稱之為A,B吧
             * 客戶端向服務端connect的時候，會發送帶有SYN標志的包(第一次握手)
             * 服務端收到客戶端發來的SYN時，向客戶端發送SYN ACK確認(第二次握手)
             * 此時TCP內核模塊把客戶端連接加入到A隊列中，最后服務端收到客戶端發來的ACK時(第三次握手)
             * TCP內核模塊把客戶端連接從A隊列移到B隊列，連接成功，應用程序的accept會返回
             * 也就是說accept從B隊列中取出完成三次握手的連接
             * A隊列和B隊列的長度之和是backLog,當A,B隊列的長度之和大於backLog時，新連接將會被TCP內核拒絕
             * 所以，如果backLog過小，可能會出現accept速度跟不上,A,B隊列滿了，導致新的客戶端無法連接，
             * 要注意的是，backLog對程序支持的連接數並無影響，backLog影響的只是還沒有被accept取出的連接
             */
            //8:設置TCP連接的緩沖區
            .option(ChannelOption.SO_BACKLOG, 128)//(5)
//            .option(ChannelOption.SO_SNDBUF, 32*1024) //設置發送緩沖大小
//            .option(ChannelOption.SO_RCVBUF, 32*1024) //設置接收緩沖大小
            //9:保持連接
            .childOption(ChannelOption.SO_KEEPALIVE, true);//(6)
            //10:綁定指定的端口 進行監聽
            //此處端口號先寫死  也可以綁定多個端口  
            ChannelFuture cf2= b.bind(PORT1).sync(); // (7)  
            
            ChannelFuture cf3= b.bind(PORT2).sync(); // (7)   綁定多個端口 

            //Thread.sleep(10000);
            cf2.channel().closeFuture().sync(); //異步等待關閉 
            cf3.channel().closeFuture().sync(); //異步等待關閉
            
        }catch(Exception e){
            e.printStackTrace();
        }finally{
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
}

接下來，就是真正去實現數據傳輸的業務邏輯層代碼的實現，在這里也就是ClientHanlder和ServlerHandler:

package ruizhan.hjf.netty;

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

/**
 * 客戶端業務處理類
 * (編寫主要的業務邏輯)
 * @author huangjianfei
 */
public class ClientHandler extends ChannelHandlerAdapter
{
    // ByteBuf是一個引用計數對象，這個對象必須顯示地調用release()方法來釋放。
    // 請記住處理器的職責是釋放所有傳遞到處理器的引用計數對象。
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
    {
        try{
            //do something
            //接收服務端發來的數據 ByteBuf
            ByteBuf  buf = (ByteBuf)msg;
            //創建一個和buf一樣長度的字節空數組
            byte[] data = new byte[buf.readableBytes()];
            //將buf中的數據讀取到data數組中
            buf.readBytes(data);
            //將data數組驚醒包裝 以String格式輸出
            String response = new String(data,"utf-8");
            System.out.println("client :"+response);
            
            //以上代碼是接收服務端發來的反饋數據//
            
            ctx.close();
        }finally{
            // Discard the received data silently.
            ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
    {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

package ruizhan.hjf.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.ReferenceCountUtil;

/**
 * 服務端業務處理類
 * (編寫主要的業務邏輯)
 * @author huangjianfei
 */
public class ServerHandler extends ChannelHandlerAdapter
{

    /**
     * 每當從客戶端收到新的數據時，這個方法會在收到消息時被調用
     * ByteBuf是一個引用計數對象，這個對象必須顯示地調用release()方法來釋放。
     * 請記住處理器的職責是釋放所有傳遞到處理器的引用計數對象。
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception
    {
        try{
            //do something
            //接收客戶端發送的數據 ByteBuf
            ByteBuf buf = (ByteBuf)msg;
            //創建一個和buf長度一樣的空字節數組
            byte[] data = new byte[buf.readableBytes()];
            //將buf中的數據讀取到data數組中
            buf.readBytes(data);
            //將data數據包裝成string輸出
            String request = new String(data,"utf-8");
            System.out.println("server :"+request);
    
            //以上代碼是接收客戶端信息//
            
            //server端向client發送反饋數據
            //如果是綁定了多個端口 那么都會進行發送
            ctx.writeAndFlush(Unpooled.copiedBuffer("888".getBytes()))
            .addListener(ChannelFutureListener.CLOSE);//添加監聽 當服務端向客戶端發送完數據后，關閉connect連接
            /**
             * ChannelFutureListener,當一個寫請求完成時通知並且關閉Channel
             * 加上監聽 意味着服務端回送數據到客戶端時 連接關閉(短連接)
             * 不加監聽 意味着客戶端與服務端一直保持連接狀態(長連接)
             */
            
            
            ctx.close();
        }finally{
            // Discard the received data silently.
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * exceptionCaught()事件處理方法是當出現Throwable對象才會被調用
     * 即當Netty由於IO錯誤或者處理器在處理事件時拋出的異常時
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
    {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
 
}

以上是Netty的基礎入門實現，詳見並發編程網，http://ifeve.com/netty5-user-guide/