Netty（一）——Netty入門程序

時間  2019-11-06
標籤 netty 入門 程序 欄目 Netty 简体版
原文   原文鏈接

轉載請註明出處：http://www.cnblogs.com/Joanna-Yan/p/7447618.htmlhtml

有興趣的可先了解下：4種I/O的對比與選型java

主要內容包括：sql

  • Netty開發環境的搭建
  • 服務端程序TimeServer開發
  • 客戶端程序TimeClient開發
  • 時間服務器的運行和調試

1.Netty開發環境的搭建

　　前提：電腦上已經安裝了JDK1.7並配置了JDK的環境變量path。bootstrap

　　從Netty官網下載Netty最新安裝包，解壓。數組

　　這時會發現裏面包含了各個模塊的.jar包和源碼，因爲咱們直接以二進制類庫的方式使用Netty，因此只須要netty-all-4.1.15.Final.jar便可。服務器

　　新建Java工程，引入netty-all-4.1.15.Final.jar。微信

2.Netty服務端開發

　　在使用Netty開發TimeServer以前，先回顧一下使用NIO進行服務端開發的步驟。網絡

  1. 建立ServerSocketChannel，配置它爲非阻塞模式；
  2. 綁定監聽，配置TCP參數，例如backlog大小；
  3. 建立一個獨立的I/O線程，用於輪詢多路複用器Selector；
  4. 建立Selector，將以前建立的ServerSocketChannel註冊到Selector上，監聽SelectionKey.ACCEPT;
  5. 啓動I/O線程，在循環體中執行Selector.select()方法，輪詢就緒的Channel；
  6. 當輪詢到了就緒狀態的Channel時，須要對其進行判斷，若是是OP_ACCEPT狀態，說明是新的客戶端接入，則調用ServerSocketChannel.accept()方法接受新的客戶端；
  7. 設置新接入的客戶端鏈路SocketChannel爲非阻塞模式，配置其餘的一些TCP參數；
  8. 將SocketChannel註冊到Selector，監聽OP_READ操做位；
  9. 若是輪詢的Channel爲OP_READ，則說明SocketChannel中有新的就緒的數據包須要讀取，則構造ByteBuffer對象，讀取數據包；
  10. 若是輪詢的Channel爲OP_WRITE，說明數據尚未發送完成，須要繼續發送。

　　一個簡單的NIO服務端程序，若是須要咱們直接使用JDK的NIO類庫進行開發，居然須要通過繁瑣的十多步操做才能完成最基本的消息讀取和發送，這也是咱們選擇Netty等NIO框架的緣由了，下面咱們看看使用Netty是如何輕鬆搞定服務端開發的。框架

package joanna.yan.netty;

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 {
        int port=9090;
        if(args!=null&&args.length>0){
            try {
                port=Integer.valueOf(args[0]);
            } catch (Exception e) {
                // 採用默認值
            }
        }
        new TimeServer().bind(port);
    }
    
    public void bind(int port) throws Exception{
        /*
         * 配置服務端的NIO線程組，它包含了一組NIO線程，專門用於網絡事件的處理，實際上它們就是Reactor線程組。
         * 這裏建立兩個的緣由：一個用於服務端接受客戶端的鏈接，
         * 另外一個用於進行SocketChannel的網絡讀寫。
         */
        EventLoopGroup bossGroup=new NioEventLoopGroup();
        EventLoopGroup workerGroup=new NioEventLoopGroup();
        try {
            //ServerBootstrap對象，Netty用於啓動NIO服務端的輔助啓動類，目的是下降服務端的開發複雜度。
            ServerBootstrap b=new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 1024)
             /*
              * 綁定I/O事件的處理類ChildChannelHandler，它的做用相似於Reactor模式中的handler類，
              * 主要用於處理網絡I/O事件，例如：記錄日誌、對消息進行編解碼等。
              */
             .childHandler(new ChildChannelHandler());
            /*
             * 綁定端口，同步等待成功（調用它的bind方法綁定監聽端口，隨後，調用它的同步阻塞方法sync等待綁定操做完成。
             * 完成以後Netty會返回一個ChannelFuture,它的功能相似於JDK的java.util.concurrent.Future，
             * 主要用於異步操做的通知回調。）
             */
            ChannelFuture f=b.bind(port).sync();
            
            //等待服務端監聽端口關閉（使用f.channel().closeFuture().sync()方法進行阻塞，等待服務端鏈路關閉以後main函數才退出。）
            f.channel().closeFuture().sync();
        }finally{
            //優雅退出，釋放線程池資源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

        @Override
        protected void initChannel(SocketChannel arg0) throws Exception {
            arg0.pipeline().addLast(new TimeServerHandler());
        }      
    }
}
package joanna.yan.netty;

import java.sql.Date;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * 用於對網絡事件進行讀寫操做
 * @author Joanna.Yan
 * @date 2017年11月8日下午4:15:13
 */
//public class TimeServerHandler extends ChannelHandlerAdapter{//已摒棄
public class TimeServerHandler extends ChannelInboundHandlerAdapter{

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        //ByteBuf相似於JDK中的java.nio.ByteBuffer對象，不過它提供了更增強大和靈活的功能。
        ByteBuf buf=(ByteBuf) msg;
        byte[] req=new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body=new String(req, "UTF-8");
        System.out.println("The time server receive order : "+body);
        String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body) ? new 
                Date(System.currentTimeMillis()).toString() : "BAD ORDER";
        ByteBuf resp=Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.write(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        /*
         * ctx.flush();將消息發送隊列中的消息寫入到SocketChannel中發送給對方。
         * 從性能角度考慮，爲了防止頻繁地喚醒Selector進行消息發送，Netty的write方法並不直接將消息寫入SocketChannel中，
         * 調用write方法只是把待發送的消息放到發送緩衝數組中，再經過調用flush方法，將發送緩衝區中的消息所有寫到SocketChannel中。
         */
        ctx.flush();
    }

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

3.Netty客戶端開發

package joanna.yan.netty;

import io.netty.bootstrap.Bootstrap;
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;

public class TimeClient {
    public static void main(String[] args) throws Exception {
        int port=9019;
        if(args!=null&&args.length>0){
            try {
                port=Integer.valueOf(args[0]);
            } catch (Exception e) {
                // 採用默認值
            }
        }
        new TimeClient().connect(port, "127.0.0.1");
    }
    
    public void connect(int port,String host) throws Exception{
        //配置客戶端NIO線程組
        EventLoopGroup group=new NioEventLoopGroup();
        
        try {
            Bootstrap b=new Bootstrap();
            b.group(group)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new TimeClientHandler());
                }
            });
            
            //發起異步鏈接操做
            ChannelFuture f=b.connect(host, port).sync();
            
            //等待客戶端鏈路關閉
            f.channel().closeFuture().sync();
        }finally{
            //優雅退出，釋放NIO線程組
            group.shutdownGracefully();
        }
    }
}
package joanna.yan.netty;

import java.util.logging.Logger;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;

//public class TimeClientHandler extends ChannelHandlerAdapter{//已摒棄
public class TimeClientHandler extends ChannelInboundHandlerAdapter{
    private static final Logger logger=Logger.getLogger(TimeClientHandler.class.getName());
    private final ByteBuf firstMessage;
    
    public TimeClientHandler(){
        byte[] req="QUERY TIME ORDER".getBytes();
        firstMessage=Unpooled.buffer(req.length);
        firstMessage.writeBytes(req);
    }

    /**
     * 當客戶端和服務端TCP鏈路創建成功以後，Netty的NIO線程會調用channelActive方法，
     * 發送查詢時間的指令給服務端。
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //將請求信息發送給服務端
        ctx.writeAndFlush(firstMessage);
    }

    /**
     * 當服務端返回應答消息時調用channelRead方法
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
            throws Exception {
        ByteBuf buf=(ByteBuf) msg;
        byte[] req=new byte[buf.readableBytes()];
        buf.readBytes(req);
        String body=new String(req, "UTF-8");
        System.out.println("Now is :"+body);
    }
    
    /**
     * 發生異常是，打印異常日誌，釋放客戶端資源。
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
            throws Exception {
        //釋放資源
        logger.warning("Unexpected exception from downstream : "+cause.getMessage());
        ctx.close();
    }  
}

 ChannelInboundHandlerAdapter和SimpleChannelInboundHandler的使用區分：異步

 　　ChannelInboundHandlerAdapter是普通類，而SimpleChannelInboundHandler<T>是抽象類，繼承SimpleChannelInboundHandler的類必須實現channelRead0方法；SimpleChannelInboundHandler<T>有一個重要特性，就是消息被讀取後，會自動釋放資源，常見的IM聊天軟件的機制就相似這種。並且SimpleChannelInboundHandler類是繼承了ChannelInboundHandlerAdapter類，重寫了channelRead()方法，並新增抽象類。絕大部分場景均可以用ChannelInboundHandlerAdapter來處理。

4.運行與調試

　　服務端運行結果：

　　客戶端運行結果：

 

　　運行結果正確。能夠發現，經過Netty開發的NIO服務端和客戶端很是簡單，短短几十行代碼就能完成以前NIO程序須要幾百行才能完成的功能。基於Netty的應用開發不但API使用簡單、開發模式固定，並且擴展性和定製性很是好，後面，會經過更多應用來介紹Netty的強大功能。

　　須要指出的是，本示例沒有考慮讀半包的處理，對於功能演示或者測試，上述程序沒有問題，可是稍加改造進行性能或者壓力測試，它就不能正確地工做了。後面咱們會給出可以正確處理半包消息的應用實例。

Netty（二）——TCP粘包/拆包

 若是此文對您有幫助，微信打賞我一下吧~

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程