java編寫基於netty的RPC框架

 

一 簡單概念

RPC:(Remote Procedure Call),遠程調用過程,是通過網絡調用遠程計算機的進程中某個方法,從而獲取到想要的數據,過程如同調用本地的方法一樣.

阻塞IO:當阻塞I/O在調用InputStream.read()方法是阻塞的,一直等到數據到來時才返回,同樣ServerSocket.accept()方法時,也是阻塞,直到有客戶端連接才返回,I/O通信模式如下:

　　　　缺點:當客戶端多時,會創建大量的處理線程,並且為每一個線程分配一定的資源;阻塞可能帶來頻繁切換上下文,這時引入NIO

NIO:  jdk1.4引入的(NEW Input/Output),是基於通過和緩存區的I/O方式,(插入一段題外話,學的多忘得也多,之前有認真研究過NIO,后來用到的時候,忘得一干二凈,所以學習一些東西,經常返回看看),NIO是一種非阻塞的IO模型,通過不斷輪詢IO事件是否就緒,非阻塞是指線程在等待IO的時候,可以做其他的任務,同步的核心是Selector,Selector代替線程本省的輪詢IO事件,避免了阻塞同時減少了不必要的線程消耗;非阻塞的核心是通道和緩存區,當IO事件的就緒時,可以將緩存區的數據寫入通道

　　　　其工作原理:

　　　　　　1  由專門的線程來處理所有的IO事件,並且負責轉發

　　　　　　 2  事件驅動機制:事件到的時候才觸發,而不是同步監視

　　　　　　 3  線程通訊:線程之間通訊通過wait,notify等方式通訊,保證每次上下文切換都是有意義的,減少沒必要的線程切換

通道:  是對原I/O包中流的模擬,所有數據必須通過Channel對象,常見的通道FileChannel,SocketChannel,ServerSocketChannel,DatagramChannel(UDP協議向網絡連接的兩端讀寫數據)

Buffer緩存區:實際上是一個容器,一個連續的數組,任何讀寫的數據都經過Buffer

Netty:是由JBOSS提供的一個java開源框架,是一個高性能,異步事件驅動的NIO框架,基於JAVA NIO提供的API實現,他提供了TCP UDP和文件傳輸的支持,,所有操作都是異步非阻塞的.通過Futrue-Listener機制,本質就是Reactor模式的現實,Selector作為多路復用器,EventLoop作為轉發器,而且,netty對NIO中buffer做優化,大大提高了性能

 二  Netty 客戶端和服務端的

 Netty中Bootstrap和Channel的生命周期

Bootstrap簡介

　　Bootstarp:引導程序,將ChannelPipeline,ChannelHandler,EventLoop進行整體關聯

 

Bootstrap具體分為兩個實現

　　ServerBootstrap:用於服務端,使用一個ServerChannel接收客戶端的連接,並創建對應的子Channel

　　Bootstrap:用於客戶端,只需要一個單獨的Channel,配置整個Netty程序,串聯起各個組件

二者的主要區別:

　　1 ServerBootstrap用於Server端,通過調用bind()綁定一個端口監聽連接,Bootstrap用於Client端,需要調用connect()方法來連接服務器端,我們也可以調用bind()方法接收返回ChannelFuture中Channel

 　　2 客戶端的Bootstrap一般用一個EventLoopGroup,而服務器的ServerBootstrap會用兩個第一個EventLoopGroup專門負責綁定到端口監聽連接事件,而第二個EventLoopGroup專門用來處處理每個接收的連接,這樣大大提高了並發量

 

public class Server {

    public static void main(String[] args) throws Exception {
        // 1 創建線兩個事件循環組
        // 一個是用於處理服務器端接收客戶端連接的
        // 一個是進行網絡通信的（網絡讀寫的）
        EventLoopGroup pGroup = new NioEventLoopGroup();
        EventLoopGroup cGroup = new NioEventLoopGroup();

        // 2 創建輔助工具類ServerBootstrap，用於服務器通道的一系列配置
        ServerBootstrap b = new ServerBootstrap();
        b.group(pGroup, cGroup) // 綁定倆個線程組
                .channel(NioServerSocketChannel.class) // 指定NIO的模式.NioServerSocketChannel對應TCP, NioDatagramChannel對應UDP
                .option(ChannelOption.SO_BACKLOG, 1024) // 設置TCP緩沖區
                .option(ChannelOption.SO_SNDBUF, 32 * 1024) // 設置發送緩沖大小
                .option(ChannelOption.SO_RCVBUF, 32 * 1024) // 這是接收緩沖大小
                .option(ChannelOption.SO_KEEPALIVE, true) // 保持連接
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel sc) throws Exception {  //SocketChannel建立連接后的管道
                        // 3 在這里配置 通信數據的處理邏輯, 可以addLast多個...
                        sc.pipeline().addLast(new ServerHandler());
                    }
                });

        // 4 綁定端口, bind返回future(異步), 加上sync阻塞在獲取連接處
        ChannelFuture cf1 = b.bind(8765).sync();
        //ChannelFuture cf2 = b.bind(8764).sync();   //可以綁定多個端口
        // 5 等待關閉, 加上sync阻塞在關閉請求處
        cf1.channel().closeFuture().sync();
        //cf2.channel().closeFuture().sync();
        pGroup.shutdownGracefully();
        cGroup.shutdownGracefully();
    }
}

 

 

public class ServerHandler extends ChannelHandlerAdapter {

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("server channel active... ");
    }

    @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("Server :" + body );
            String response = "返回給客戶端的響應：" + body ;
            ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes()));
            // future完成后觸發監聽器, 此處是寫完即關閉(短連接). 因此需要關閉連接時, 要通過server端關閉. 直接關閉用方法ctx[.channel()].close()
            //.addListener(ChannelFutureListener.CLOSE);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)
            throws Exception {
        System.out.println("讀完了");
        ctx.flush();
    }

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

public class Client {

    public static void main(String[] args) throws Exception {
        
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        b.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel sc) throws Exception { 
                sc.pipeline().addLast(new ClientHandler());
            }
        });
        
        ChannelFuture cf1 = b.connect("127.0.0.1", 8765).sync();
        //ChannelFuture cf2 = b.connect("127.0.0.1", 8764).sync();  //可以使用多個端口
        //發送消息, Buffer類型. write需要flush才發送, 可用writeFlush代替
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("777".getBytes()));
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("666".getBytes()));
        Thread.sleep(2000);
        cf1.channel().writeAndFlush(Unpooled.copiedBuffer("888".getBytes()));
        //cf2.channel().writeAndFlush(Unpooled.copiedBuffer("999".getBytes()));
        
        cf1.channel().closeFuture().sync();
        //cf2.channel().closeFuture().sync();
        group.shutdownGracefully();
    }
}

 

public class ClientHandler extends ChannelHandlerAdapter{

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "utf-8");
            System.out.println("Client :" + body );
        } finally {
            // 記得釋放xxxHandler里面的方法的msg參數: 寫(write)數據, msg引用將被自動釋放不用手動處理; 但只讀數據時,!必須手動釋放引用數
             ReferenceCountUtil.release(msg);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    }

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

其他組件:

　　Handle:為了支持各種協議和處理數據的方式,可以是連接,數據接收,異常,數據格式轉換等

　　ChannelHandler 

　　ChannelInboundHandler:最常用的Handler,作用是處理接收數據的事件,來處理我們的核心業務邏輯。

 

　　ChannelInitializer:，當一個鏈接建立時，我們需要知道怎么來接收或者發送數據，當然，我們有各種各樣的Handler實現來處理它,那么ChannelInitializer便是用來配置這些Handler，它會提供一個ChannelPipeline，並把Handler加入到ChannelPipeline。

　　ChannelPipeline:一個Netty應用基於ChannelPipeline機制,這種機制依賴EventLoop和EventLoopGroup,這三個都和事件或者事件處理相關

　　EventLoop: 為Channel處理IO操作,一個EventLoop可以為多個Channel服務

　　EventLoopGroup:包含多個EventLoop　

　　Channel:代表一個Socket連接

　　Future:在Netty中所有的IO操作都是異步的,,因此我們不知道,過來的請求是否被處理了,所以我們注冊一個監聽,當操作執行成功或者失敗時監聽自動觸發,所有操作都會返回一個ChannelFutrue

　　ChannelFuture

Netty是一個非阻塞的,事件驅動的,網絡編程框架,我們通過一張圖理解一下,Channel,EventLoop以及EventLoopGroup之間的關系

解釋一下,當一個連接過來,Netty首先會注冊一個channel,然后EventLoopGroup會分配一個EventLoop綁定到這個channel,在這個channel的整個生命周期過程中,這個EventLoop一直為他服務,這個玩意就是一個線程

 

這下聊一下Netty如何處理數據?

 前面有講到,handler數據處理核心,,而ChannelPipeline負責安排Handler的順序和執行,我們可以這樣理解,數據在ChannelPipeline中流動,其中ChannelHandler就是一個個閥門,這些數據都會經過每一個ChannelHandler並且被他處理,其中ChannelHandler的兩個子類ChannelOutboundHandler和ChannelInboundHandler,根據不同的流向,選擇不同的Handler

由圖可以看出,一個數據流進入ChannelPipeline時,一個一個handler挨着執行,各個handler的數據傳遞,這需要調用方法中ChannelHandlerContext來操作,而這個ChannelHandlerContext可以用來讀寫Netty中的數據流

三 Netty中的業務處理

   netty中會有很多Handler.具體哪一種Handler還要看繼承是InboundAdapter還是OutboundAdapter,Netty中提供一系列的Adapter來幫助我們簡化開發,在ChannelPipeline中的每一個handler都負責把Event傳遞個洗下一個handler,有這些adapter,這些工作可以自動完成,,我們只需覆蓋我們真正實現的部分即可,接下來比較常用的三種ChannelHandler　　

　　Encoders和Decodeers

　　我們在網絡傳輸只能傳輸字節流,在發送數據時,把我們的message轉換成bytes這個過程叫Encode(編碼),相反,接收數據,需要把byte轉換成message,這個過程叫Decode(解碼)

 　　Domain Logic

　　我們真正關心的如何處理解碼以后的數據,我們真正的業務邏輯便是接收處理的數據,Netty提供一個常用的基類就是SimpleChannelInboundHandler<T>,其中T就是Handler處理的數據類型,消息到達這個Handler,會自動調用這個Handler中的channelRead0(ChannelHandlerContext,T)方法,T就是傳過來的數據對象

四 基於netty實現的Rpc的例子

　　這是我的github上項目的位置

　　https://github.com/developerxiaofeng/rpcByNetty

項目目錄結構如下

 

 詳細的項目細節看類中的注釋,很詳細哦!!!