Netty之Future和ChannelFuture

時間 2019-11-19

標籤 netty future channelfuture 欄目 Netty 简体版

原文原文鏈接

Future，在Netty中全部的IO操做都是異步的，所以，你不能馬上得知消息是否被正確處理，可是咱們能夠過一會等它執行完成或者直接註冊一個監聽，具體的實現就是經過Future和ChannelFuture,他們能夠註冊一個監聽，當操做執行成功或失敗時監聽會自動觸發。總之，全部的操做都會返回一個ChannelFuture。java

‍‍‍Netty中的異步，就不得不提ChannelFuture。Netty中的IO操做是異步的，包括bind、write、connect等操做會簡單的返回一個ChannelFuture，調用者並不能馬上得到結果。‍‍‍bootstrap

在netty中全部的io操做都是異步的，也就是說咱們在發送完消息後，netty內部是採用線程池去處理，方法當即返回了，但有時候咱們須要外部方法等待服務器的響應，整個過程須要同步處理，那麼就須要將異步調用轉爲同步調用，原理很簡單，就是在調用異步方法後，主線程阻塞，直到異步方法返回結果數組

在netty中全部的I/O操做都是異步，這意味着netty提供的I/O方法調用都將當即返回，會返回一個ChannelFuture對象的實像，它將會給你一些信息，關於I/O執行狀態的結果，但此時不能保證真正的I/O操做已經完成。服務器

推薦使用addListener(ChannelFutureListener)異步獲得通知當一個I/O操做完成後,作任何後續任務，而不是經過調用await方法（下降吞吐量）。但若是你想要業務場景是必須先執行A,而後同步執行B（異步通知不合適的場景）,使用await是比較方便的。但await有一個限制，調用await方法的線程不能是I/O 線程（work線程），不然會拋出一個異常，避免死鎖。框架

做爲一個異步NIO框架，Netty的全部IO操做都是異步非阻塞的，經過Future-Listener機制，用戶能夠方便的主動獲取或者經過通知機制得到IO操做結果。異步

好比：socket

/**
 * Connect a {@link Channel} to the remote peer.
 */
public ChannelFuture connect(String inetHost, int inetPort) {
    return connect(new InetSocketAddress(inetHost, inetPort));
}

/**
 * Create a new {@link Channel} and bind it.
 */
public ChannelFuture bind(int inetPort) {
    return bind(new InetSocketAddress(inetPort));
}

/**
 * Request to write a message via this {@link Channel} through the {@link ChannelPipeline}.
 * This method will not request to actual flush, so be sure to call {@link #flush()}
 * once you want to request to flush all pending data to the actual transport.
 */
ChannelFuture write(Object msg);

判斷Future狀態的方法

Future狀態圖
ide

ChannelFuture對象狀態只有uncompleted和completed。當一個I/O操做開始時，一個ChannelFuture實例被建立（我知道的暫時除close方法），剛開始時，future對象的實例即不是succeeded,failed,cancelled。由於真正的I/O操做尚未完成。若是正的I/O操做已經完成，那麼future的狀態將是completed,不管結果是succeeded,failed,cancelled。oop

netty中Future接口中的方法

netty的Future是繼承自java.util.concurrent.Future接口this

/**
 * Returns {@code true} if and only if the I/O operation was completed
 * successfully.
 */
boolean isSuccess();

/**
 * returns {@code true} if and only if the operation can be cancelled via {@link #cancel(boolean)}.
 */
boolean isCancellable();

java.util.concurrent.Future接口中的方法：

/**
 * Waits for this future until it is done, and rethrows the cause of the failure if this future
 * failed.
 */
Future<V> sync() throws InterruptedException;

/**
 * Waits for this future until it is done, and rethrows the cause of the failure if this future
 * failed.
 */
Future<V> syncUninterruptibly();

/**
 * Waits for this future to be completed.
 *
 * @throws InterruptedException
 *         if the current thread was interrupted
 */
Future<V> await() throws InterruptedException;

/**
 * Waits for this future to be completed without
 * interruption.  This method catches an {@link InterruptedException} and
 * discards it silently.
 */
Future<V> awaitUninterruptibly();

/**
 * Waits for this future to be completed within the
 * specified time limit.
 *
 * @return {@code true} if and only if the future was completed within
 *         the specified time limit
 *
 * @throws InterruptedException
 *         if the current thread was interrupted
 */
boolean await(long timeout, TimeUnit unit) throws InterruptedException;

/**
 * Waits for this future to be completed within the
 * specified time limit.
 *
 * @return {@code true} if and only if the future was completed within
 *         the specified time limit
 *
 * @throws InterruptedException
 *         if the current thread was interrupted
 */
boolean await(long timeoutMillis) throws InterruptedException;

/**
 * Waits for this future to be completed within the
 * specified time limit without interruption.  This method catches an
 * {@link InterruptedException} and discards it silently.
 *
 * @return {@code true} if and only if the future was completed within
 *         the specified time limit
 */
boolean awaitUninterruptibly(long timeout, TimeUnit unit);

/**
 * Waits for this future to be completed within the
 * specified time limit without interruption.  This method catches an
 * {@link InterruptedException} and discards it silently.
 *
 * @return {@code true} if and only if the future was completed within
 *         the specified time limit
 */
boolean awaitUninterruptibly(long timeoutMillis);

維持netty中Future的生命週期的方法

sync()

syncUninterruptibly()

await()

await(long timeout, TimeUnit unit)

awaitUninterruptibly(long timeout, TimeUnit unit)：

awaitUninterruptibly(long timeoutMillis);

示例DEMO

服務器端代碼

package hello.netty.lyx.com;

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 HelloServer {
    public void start(int port) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            // 註冊handler
                            ch.pipeline().addLast(new HelloServerInHandler());
                        }
                    }).option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);
            long t1 = System.currentTimeMillis();
            ChannelFuture f = b.bind(port).sync();
            f.channel().closeFuture().sync();
            long t2 = System.currentTimeMillis();
            System.out.print("diff in seconds:" + (t2 - t1) / 1000 + "\n");
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    public static void main(String[] args) throws Exception {
        HelloServer server = new HelloServer();
        server.start(9090);
    }
}

package hello.netty.lyx.com;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

// 該handler是InboundHandler類型
public class HelloServerInHandler extends ChannelInboundHandlerAdapter {
    @Override
    public boolean isSharable() {
        System.out.println("==============handler-sharable==============");
        return super.isSharable();
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel-register==============");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel-unregister==============");
    }

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

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel-inactive==============");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        System.out.println("==============channel-read==============");

        ByteBuf result = (ByteBuf) msg;
        byte[] result1 = new byte[result.readableBytes()];
        // msg中存儲的是ByteBuf類型的數據，把數據讀取到byte[]中
        result.readBytes(result1);
        String resultStr = new String(result1);
        // 接收並打印客戶端的信息
        System.out.println("Client said:" + resultStr);
        // 釋放資源，這行很關鍵
        result.release();

        // 向客戶端發送消息
        String response = "I am ok!";
        // 在當前場景下，發送的數據必須轉換成ByteBuf數組
        ByteBuf encoded = ctx.alloc().buffer(4 * response.length());
        encoded.writeBytes(response.getBytes());
        ctx.writeAndFlush(encoded);
        Thread.sleep(10000);
        System.out.println("thread sleep end");

        ctx.close();

//        Thread.sleep(10000);
//        System.out.println("thread sleep end");
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel-read-complete==============");
        ctx.flush();
    }
}

客戶端代碼

package hello.netty.lyx.com;

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;

/**
 * 一、Client向Server發送消息：Are you ok?
 * 二、Server接收客戶端發送的消息，並打印出來。
 * 三、Server端向客戶端發送消息：I am ok!
 * 四、Client接收Server端發送的消息，並打印出來，通信結束。
 */

public class HelloClient {
    public void connect(String host, int port) throws Exception {
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup);
            b.channel(NioSocketChannel.class);
            b.option(ChannelOption.SO_KEEPALIVE, true);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new HelloClientIntHandler());
                }
            });

            // Start the client.
            /**
             * wait()方法：Waits for this future to be completed.
             * Waits for this future until it is done, and rethrows the cause of the failure if this future
             * failed.
             */
            long t1 = System.currentTimeMillis();
            ChannelFuture f = b.connect(host, port).await();
            // Wait until the connection is closed.
            f.channel().closeFuture().await();    //closeFuture方法返回通道關閉的結果
            long t2 = System.currentTimeMillis();
            System.out.print("diff in seconds:" + (t2 - t1) / 1000 + "\n");
        } finally {
            workerGroup.shutdownGracefully();
        }

    }

    public static void main(String[] args) throws Exception {
        HelloClient client = new HelloClient();
        client.connect("127.0.0.1", 9090);
    }
}

package hello.netty.lyx.com;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;


//InboundHandler類型
public class HelloClientIntHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel--register==============");
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel--unregistered==============");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel--inactive==============");
    }

    // 鏈接成功後，向server發送消息
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("==============channel--active==============");
        String msg = "Are you ok?";
        /**
         * 分配ByteBuf
         * Return the assigned {@link io.netty.buffer.ByteBufAllocator} which will be used to allocate {@link ByteBuf}s.
         */
        ByteBuf encoded = ctx.alloc().buffer(4 * msg.length());
        encoded.writeBytes(msg.getBytes());
        ctx.write(encoded);
        ctx.flush();
    }

    // 接收server端的消息，並打印出來
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("==============channel--read==============");
        //先等待兩秒
        Thread.sleep(2000);
        ByteBuf result = (ByteBuf) msg;
        byte[] result1 = new byte[result.readableBytes()];
        result.readBytes(result1);
        System.out.println("Server said:" + new String(result1));
        result.release();
    }
}

客戶端的異步IO

讓這個demo異步方式運行則客戶端的代碼應該是這樣的：

long t1 = System.currentTimeMillis();
ChannelFuture f = b.connect(host, port).await();
long t2 = System.currentTimeMillis();
System.out.print("diff in seconds:" + (t2 - t1) / 1000 + "\n");

看運行結果：

==============channel--register==============

diff in seconds:0

==============channel--active==============

==============channel--inactive==============

==============channel--unregistered==============

和原來的代碼相比，經過運行結果能夠分析出沒有read服務器的數據。

在看一段異步的代碼：

long t1 = System.currentTimeMillis();
ChannelFuture f = b.connect(host, port).await();
f = f.channel().closeFuture();
f.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        System.out.println("success complete!!ok!!");
    }
});
long t2 = System.currentTimeMillis();
System.out.print("diff in seconds:" + (t2 - t1) / 1000 + "\n");

運行結果：

==============channel--register==============

==============channel--active==============

diff in seconds:0

success complete!!ok!!

==============channel--inactive==============

==============channel--unregistered==============

給通道的關閉Future註冊了監聽事件，監聽事件等這個關閉Future完成後打印了字符串，而客戶端沒有讀取服務器的數據。

在看一段代碼

long t1 = System.currentTimeMillis();
ChannelFuture f = b.connect(host, port).await();
f = f.channel().closeFuture().await();
f.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        System.out.println("success complete!!ok!!");
    }
});
long t2 = System.currentTimeMillis();
System.out.print("diff in seconds:" + (t2 - t1) / 1000 + "\n");

運行結果：

==============channel--register==============

==============channel--active==============

==============channel--read==============

Server said:I am ok!

==============channel--inactive==============

==============channel--unregistered==============

diff in seconds:2

success complete!!ok!!

能夠讀取服務器的數據，而且監聽事件也起了做用，但這不是一個異步調用。

=============END=============