一、异步模型

1.1、基本介绍

• 异步的概念和同步相对，当一个异步过程调用发出后，调用者不能立刻得到。实际处理这个调用的组件在完成后，通过状态、通知和回调来通知调用者。
• Netty中的IO操作是异步的，包括Bind，Write、Connect等操作会简单的返回一个ChannelFuture。
• 调用者并不能立刻获得结果，而是通过Future-Listener机制，用户可以方便的主动获得或者通过通知机制获得IO操作结果
• Netty的异步模型是建立在future和callback的之上的。callback就是回调。重点说Future，他的核心思想是：假设一个方法fun,计算过程可能非常耗时，等待fun返回显然不合适，那么可以在调用fun的时候，立马返回一个future，后续可以通过future去监控方法fun的处理过程（即：future-listener机制）

1.2 Future说明

• 表示异步的执行结果，可以通过它提供的方法来检测执行是否完成，比如检索计算等等。
• ChannelFuture是一个接口：我们可以添加监听器，当监听的时间发生时，就会通知监听器。

1.3 异步模型

在使用Netty进行编程时，看接操作和转换出入站数据只需要您提供callback或利用future即可。这使得链式操作简单、高效、并有利于编写高可重用的、通用的代码。
Netty框架的目标就是让你的业务逻辑从网络基础应用中分离出来、解脱出来

在每一个handler中的方法里，都可以添加异步操作

1.4 future-Listener机制

当Future对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture来获取操作执行的状态，注册监听函数来执行完成后的操作

常见有如下操作💥

通过isDone方法来判断当前操作是否完成
通过isSuccess方法来判断已完成的当前操作是否成功
通过getCause方法来获取已完成的当前操作失败的原因
通过isCancelled方法来判断已完成的当前操作是否被取消
通过addListener方法来注册监听器，当操作已完成（isDone 方法返回完成），将会通知指定的监听器，如果Future对象已完成，则通知指定的监听器

二、 代码实例演示

实验目的：

• 研究Netty的异步模型的工作机制
• 学习如果监听任务的执行状态
• 掌握ChannelFuture、future-Listener

• 实验步骤

• 服务端启动服务
• 客户端连接客户端，并向客户端发送一条消息
• 服务端接收客户端消息后向客户端回复一条消息
• 向客户端的发消息任务添加一个 监听事件 和 判断发送结果的方法
• 
  

  2.1 NettyServer代码

  package com.netty.taskQueue;import io.netty.bootstrap.ServerBootstrap;
  import io.netty.buffer.Unpooled;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelInitializer;
  import io.netty.channel.ChannelOption;
  import io.netty.channel.nio.NioEventLoopGroup;
  import io.netty.channel.socket.SocketChannel;
  import io.netty.channel.socket.nio.NioServerSocketChannel;
  import io.netty.util.CharsetUtil;
  import io.netty.util.concurrent.Future;
  import io.netty.util.concurrent.GenericFutureListener;import java.net.ServerSocket;public class NettyServer {public static void main(String[] args) throws InterruptedException {//创建线程池对象NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workGroup = new NioEventLoopGroup();//创建配置启动类ServerBootstrap serverBootstrap = new ServerBootstrap();try {//添加相关配置serverBootstrap.group(bossGroup, workGroup)//添加线程池组.channel(NioServerSocketChannel.class)//.option(ChannelOption.SO_BACKLOG, 128)//.childOption(ChannelOption.SO_KEEPALIVE, true)//.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//可以使用一个集合管理SocketChannel,再推送消息时，可以将业务加入到各个channel//对应的NIOEventLoop的ch.eventLoop().execute(new Runnable() {@Overridepublic void run() {ch.writeAndFlush(Unpooled.copiedBuffer("这是由第三种场景实现的（非当前reactor使用channel的方法）", CharsetUtil.UTF_8));}});//每个客户端的ch.hashcode都不同，可以作为用户标识System.out.println("客户端的ch.hashcode="+ch.hashCode());ch.pipeline().addLast(new NettyServerHandler());}});//绑定端口ChannelFuture cf = serverBootstrap.bind(6668).sync();//监听通道关闭cf.channel().closeFuture().sync();}finally {bossGroup.shutdownGracefully();workGroup.shutdownGracefully();}}
  }

  2.2 NettyServerHandler代码实现

  package com.netty.taskQueue;import io.netty.buffer.ByteBuf;
  import io.netty.buffer.Unpooled;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.channel.ChannelInboundHandlerAdapter;
  import io.netty.util.CharsetUtil;import java.util.concurrent.TimeUnit;public class NettyServerHandler extends ChannelInboundHandlerAdapter {//读取客户端的数据@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {System.out.println("来自客户端：" + ctx.channel().remoteAddress() + "  ：" + ((ByteBuf) msg).toString(CharsetUtil.UTF_8));//加入自定义定时任务//schedule()方法 有三个参数： 1、Runnable对象，2、定时时间（多长时间后执行）， 3、时间单位ctx.channel().eventLoop().schedule(new Runnable() {@Overridepublic void run() {ctx.channel().writeAndFlush(Unpooled.copiedBuffer("10秒后...第二条消息回复",CharsetUtil.UTF_8));}},10, TimeUnit.SECONDS);}//回复消息给客户端@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("第一条消息回复",CharsetUtil.UTF_8));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.channel().close();cause.printStackTrace();}
  }

  2.3 NettyClient 代码实现

  package com.netty.taskQueue;import io.netty.bootstrap.Bootstrap;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelInitializer;
  import io.netty.channel.nio.NioEventLoopGroup;
  import io.netty.channel.socket.SocketChannel;
  import io.netty.channel.socket.nio.NioSocketChannel;public class NettyClient {public static void main(String[] args) throws InterruptedException {NioEventLoopGroup group = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();try {bootstrap.group(group).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new NettyClientHandler());}});ChannelFuture cf = bootstrap.connect("localhost", 6668).sync();cf.channel().closeFuture().sync();}finally {group.shutdownGracefully();}}
  }

  2.4 NettyClientHandler代码实现

  package com.netty.taskQueue;import com.sun.org.apache.bcel.internal.classfile.Unknown;
  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.Unpooled;
  import io.netty.channel.*;
  import io.netty.util.CharsetUtil;
  import io.netty.util.concurrent.Future;
  import io.netty.util.concurrent.GenericFutureListener;public class NettyClientHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {String str="hello 小沙弥";ChannelFuture channelFuture = ctx.channel().writeAndFlush(Unpooled.copiedBuffer("hello ,小沙弥", CharsetUtil.UTF_8));//判断任务的执行结果是否成功boolean success = channelFuture.isSuccess();if (success){System.out.println("发送消息成功....");}else{System.out.println("发送消息失败");}//为任务添加监听事件，当任务执行完后会执行监听方法channelFuture.addListener(new ChannelFutureListener() {@Overridepublic void operationComplete(ChannelFuture future) throws Exception {System.out.println("消息已发送");}});}@Overridepublic void channelRead (ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf buf = (ByteBuf) msg;System.out.println("来自服务端："+ctx.channel().remoteAddress()+" ："+buf.toString(CharsetUtil.UTF_8));}}

  2.5 结果 及总结

  

  
  💥推荐阅读💥

  上一篇：六、Netty核心技术之TaskQueue自定义任务

  下一篇：六、Netty核心技术之TaskQueue自定义任务