文档章节

Netty之Future和ChannelFuture

秋风醉了
 秋风醉了
发布于 2014/06/12 21:42
字数 2117
阅读 797
收藏 1
点赞 0
评论 0

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=============

© 著作权归作者所有

共有 人打赏支持
秋风醉了
粉丝 222
博文 581
码字总数 411013
作品 0
东城
程序员
Netty 简单样例分析

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

xiaofer2008 ⋅ 2010/12/16 ⋅ 2

java.nio.channels.ClosedChannelException

服务器处理完业务并把处理结果返回给客户端,在关闭ChannelFuture对象时总汇报异常。 代码如下: DLPResponse response = protocol.getResponse(); if (request.getReturnValue()==1){ Chann...

qinbo ⋅ 2013/01/28 ⋅ 6

Netty5源码分析--0.核心领域概念

Netty是什么 由于通用的协议或者实现有时不能满足各种各样的需求,比如我们通常不会用一个HTTP Server来同时进行传输大文件,email以及近实时的消息如金融信息和多玩家游戏数据。我们需要一个...

geecoodeer ⋅ 2014/01/15 ⋅ 3

Netty源码分析 服务器端1

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

hyssop ⋅ 2015/11/25 ⋅ 0

netty 长连接的问题

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

天王盖地虎626 ⋅ 2016/04/20 ⋅ 4

Netty 简单样例分析

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

linugb118 ⋅ 2010/11/11 ⋅ 0

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

Netty tcp连接 客户端如何向服务端发送心跳数据

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

zilei ⋅ 2014/07/08 ⋅ 1

一个http请求在play框架中的前世今生(下)

上一篇提到了play底层的网络通信基于netty实现,于是粗略地研究了一下netty,总结如下。(netty版本是3.2.5,不同版本的实现可能差异较大) 一、netty的组件 channelBuffer: 传输Buffer和抽象...

2k10 ⋅ 2015/03/20 ⋅ 0

Netty服务端无法收到自己构造的复杂对象,但可以收到字符串,求解答啊

我使用Netty,和MessagePack进行编解码,由客户端成功发送的user对象,服务端没有感知到。 但如果客户端发送字符串,服务端正常接收。 问题:服务端无法感知客户端发送的复杂对象 使用的jar信...

eryueyu ⋅ 2016/06/12 ⋅ 5

没有更多内容

加载失败,请刷新页面

加载更多

下一页

JVM堆的理解

在JVM中,我们经常提到的就是堆了,堆确实很重要,其实,除了堆之外,还有几个重要的模块,看下图: 大 多数情况下,我们并不需要关心JVM的底层,但是如果了解它的话,对于我们系统调优是非常...

不羁之后 ⋅ 昨天 ⋅ 0

推荐:并发情况下:Java HashMap 形成死循环的原因

在淘宝内网里看到同事发了贴说了一个CPU被100%的线上故障,并且这个事发生了很多次,原因是在Java语言在并发情况下使用HashMap造成Race Condition,从而导致死循环。这个事情我4、5年前也经历...

码代码的小司机 ⋅ 昨天 ⋅ 1

聊聊spring cloud gateway的RetryGatewayFilter

序 本文主要研究一下spring cloud gateway的RetryGatewayFilter GatewayAutoConfiguration spring-cloud-gateway-core-2.0.0.RC2-sources.jar!/org/springframework/cloud/gateway/config/G......

go4it ⋅ 昨天 ⋅ 0

创建新用户和授予MySQL中的权限教程

导读 MySQL是一个开源数据库管理软件,可帮助用户存储,组织和以后检索数据。 它有多种选项来授予特定用户在表和数据库中的细微的权限 - 本教程将简要介绍一些选项。 如何创建新用户 在MySQL...

问题终结者 ⋅ 昨天 ⋅ 0

android -------- 颜色的半透明效果配置

最近有朋友问我 Android 背景颜色的半透明效果配置,我网上看资料,总结了一下, 开发中也是常常遇到的,所以来写篇博客 常用的颜色值格式有: RGB ARGB RRGGBB AARRGGBB 这4种 透明度 透明度...

切切歆语 ⋅ 昨天 ⋅ 0

CentOS开机启动subversion

建立自启动脚本: vim /etc/init.d/subversion 输入如下内容: #!/bin/bash## subversion startup script for the server## chkconfig: 2345 90 10# description: start the subve......

随风而飘 ⋅ 昨天 ⋅ 0

Nginx + uwsgi @ubuntu

uwsgi 安装 sudo apt-get install python3-pip # 注意 ubuntu python3默认没有安装pippython3 -m pip install uwsgi 代码(test.py) def application(env, start_response): start_res......

袁祾 ⋅ 昨天 ⋅ 0

版本控制工具

CSV , SVN , GIT ,VSS

颖伙虫 ⋅ 昨天 ⋅ 0

【2018.06.19学习笔记】【linux高级知识 13.1-13.3】

13.1 设置更改root密码 13.2 连接mysql 13.3 mysql常用命令

lgsxp ⋅ 昨天 ⋅ 0

LVM

LVM: 硬盘划分分区成物理卷->物理卷组成卷组->卷组划分逻辑分区。 1.磁盘分区: fdisk /dev/sdb 划分几个主分区 输入t更改每个分区类型为8e(LVM) 使用partprobe生成分区的文件:如/dev/sd...

ZHENG-JY ⋅ 昨天 ⋅ 0

没有更多内容

加载失败,请刷新页面

加载更多

下一页

返回顶部
顶部