文档章节

Netty之Future和ChannelFuture

秋风醉了
 秋风醉了
发布于 2014/06/12 21:42
字数 2117
阅读 1046
收藏 3

Future,在Netty中所有的IO操作都是异步的,因此,你不能立刻得知消息是否被正确处理,但是我们可以过一会等它执行完成或者直接注册一个监听,具体的实现就是通过Future和ChannelFuture,他们可以注册一个监听,当操作执行成功或失败时监听会自动触发。总之,所有的操作都会返回一个ChannelFuture。

Netty中的异步,就不得不提ChannelFuture。Netty中的IO操作是异步的,包括bind、write、connect等操作会简单的返回一个ChannelFuture,调用者并不能立刻获得结果。

在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操作结果。

比如:

/**
 * 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状态图

ChannelFuture对象状态只有uncompleted和completed。当一个I/O操作开始时,一个ChannelFuture实例被创建(我知道的暂时除close方法),刚开始时,future对象的实例即不是succeeded,failed,cancelled。因为真正的I/O操作还没有完成。如果正的I/O操作已经完成,那么future的状态将是completed,无论结果是succeeded,failed,cancelled。

netty ChannelFuture - zhanghua.1199 - 郁,加没

netty中Future接口中的方法

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

/**
 * 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;

/**
 * 1、Client向Server发送消息:Are you ok?
 * 2、Server接收客户端发送的消息,并打印出来。
 * 3、Server端向客户端发送消息:I am ok!
 * 4、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=============

© 著作权归作者所有

共有 人打赏支持
秋风醉了
粉丝 239
博文 572
码字总数 416654
作品 0
朝阳
程序员
私信 提问
Netty源码分析 服务器端1

Netty是由JBOSS提供的一个java开源框架。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。 做过NIO开发的人都遇到很多难以解决...

hyssop
2015/11/25
54
0
Netty 简单样例分析

Netty 是JBoss旗下的io传输的框架,他利用java里面的nio来实现高效,稳定的io传输。 作为io传输,就会有client和server,下面我们看看用netty怎样写client和server Client: 需要做的事情: 1...

linugb118
2010/11/11
0
0
netty 长连接的问题

android 客户端里面实现一个推送功能,客户端采用netty 客户端连接服务端(也是netty), 每当客户端和服务端连接上后,建立一个信道,如果信道断开,再次发起请求建立新的信道 但是,通过日...

天王盖地虎626
2016/04/20
942
4
Netty tcp连接 客户端如何向服务端发送心跳数据

tcp连接 服务端是Python写得,netty写客户端。客户端如何向服务发送心跳信息,来验证服务端没有挂掉。比如服务端网络异常,下面代码不会抛异常 public void channelActive(final ChannelHan...

zilei
2014/07/08
4.5K
1
Netty_EventLoop_EventLoopGroup

All IO operations in Netty are performed asynchronously. Future和ChannelFuture So when you connect to a host for example, this is done asynchronously(异步的) by default. The ......

秋风醉了
2014/07/01
0
0

没有更多内容

加载失败,请刷新页面

加载更多

如何在Chrome浏览器中启动deviceready事件(尝试调试phonegap项目)?

我正在开发PhoneGap应用程序,我希望能够在Chrome中调试它,而不是在电话上调试。但是,我在onGetReady()函数中初始化我的代码,该函数在PhoneGap触发“deviceready”事件时触发。由于Chr...

kisshua
21分钟前
1
0
支付宝客户端架构分析:自动化日志收集及分析

摘要: 《支付宝客户端架构解析》系列将从支付宝客户端的架构设计方案入手,带领大家进一步了解支付宝在客户端架构上的迭代与优化历程。 小蚂蚁说: 《支付宝客户端架构解析》系列将从支付宝...

阿里云官方博客
24分钟前
1
0
nginx中部署vue打包后的静态文件

如何在nginx中部署静态资源就不描述了, 请看我的这篇博客 将vue脚手架项目打包后的静态文件放到nginx上, 发现有个问题, 即url上有#, 怎么去掉这个#呢. 1 项目中router的mode 路由的mode要为h...

克虏伯
41分钟前
7
0
JS容易理解错误的地方

在这端代码执行的末尾,你会不会hi变量回事函数中的hi了?你会不会认为这不是按引用传递了? 对值传递和引用传递产生质疑了? 1 var hi = {};2 function sayHello(hi) { ...

器石_
43分钟前
4
0
Java开发学习--MongoDB

之前只学过sql,第一次使用非关系型数据库。以前对于关系型数据库与非关系型数据库的概念很模糊,通过这次的学习对这两者有了一个清晰的概念。 主键 在MongoDB中,主键名叫"_id",如果在生成...

微笑向暖wx
46分钟前
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部