《Netty 权威指南》—— AIO创建的TimeClient源码分析

简介:

声明:本文是《Netty 权威指南》的样章,感谢博文视点授权并发编程网站发布样章,禁止以任何形式转载此文。

异步非阻塞IO版本的时间服务器服务端已经介绍完毕,下面我们继续看客户端的实现。

首先看下客户端主函数的实现,AIO时间服务器客户端  TimeClient:


01 public class TimeClient {
02  
03     /**
04      * @param args
05      */
06     public static void main(String[] args) {
07     int port = 8080;
08     if (args != null && args.length > 0) {
09         try {
10         port = Integer.valueOf(args[0]);
11         } catch (NumberFormatException e) {
12         // 采用默认值
13         }
14     }
15     new Thread(new AsyncTimeClientHandler("127.0.0.1", port),
16         "AIO-AsyncTimeClientHandler-001").start();
17  
18     }
19 }

第15行我们通过一个独立的IO线程创建异步时间服务器客户端handler,在实际项目中,我们不需要独立的线程创建异步连接对象,因为底层都是通过JDK的系统回调实现的,在后面运行时间服务器程序的时候,我们会抓取线程调用堆栈给大家展示。继续看代码, AsyncTimeClientHandler的实现类源码如下:


001 public class AsyncTimeClientHandler implements
002     CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {
003  
004     private AsynchronousSocketChannel client;
005     private String host;
006     private int port;
007     private CountDownLatch latch;
008  
009     public AsyncTimeClientHandler(String host, int port) {
010     this.host = host;
011     this.port = port;
012     try {
013         client = AsynchronousSocketChannel.open();
014     } catch (IOException e) {
015         e.printStackTrace();
016     }
017     }
018  
019     @Override
020     public void run() {
021     latch = new CountDownLatch(1);
022     client.connect(new InetSocketAddress(host, port), this, this);
023     try {
024         latch.await();
025     } catch (InterruptedException e1) {
026         e1.printStackTrace();
027     }
028     try {
029         client.close();
030     } catch (IOException e) {
031         e.printStackTrace();
032     }
033     }
034  
035     @Override
036     public void completed(Void result, AsyncTimeClientHandler attachment) {
037     byte[] req = "QUERY TIME ORDER".getBytes();
038     ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
039     writeBuffer.put(req);
040     writeBuffer.flip();
041     client.write(writeBuffer, writeBuffer,
042         new CompletionHandler<Integer, ByteBuffer>() {
043             @Override
044             public void completed(Integer result, ByteBuffer buffer) {
045             if (buffer.hasRemaining()) {
046                 client.write(buffer, buffer, this);
047             } else {
048                 ByteBuffer readBuffer = ByteBuffer.allocate(1024);
049                 client.read(
050                     readBuffer,
051                     readBuffer,
052                     new CompletionHandler<Integer, ByteBuffer>() {
053                     @Override
054                     public void completed(Integer result,
055                         ByteBuffer buffer) {
056                         buffer.flip();
057                         byte[] bytes = new byte[buffer
058                             .remaining()];
059                         buffer.get(bytes);
060                         String body;
061                         try {
062                         body = new String(bytes,
063                             "UTF-8");
064                         System.out.println("Now is : "
065                             + body);
066                         latch.countDown();
067                         } catch (UnsupportedEncodingException e) {
068                         e.printStackTrace();
069                         }
070                     }
071  
072                     @Override
073                     public void failed(Throwable exc,
074                         ByteBuffer attachment) {
075                         try {
076                         client.close();
077                         latch.countDown();
078                         } catch (IOException e) {
079                         // ingnore on close
080                         }
081                     }
082                     });
083             }
084             }
085  
086             @Override
087             public void failed(Throwable exc, ByteBuffer attachment) {
088             try {
089                 client.close();
090                 latch.countDown();
091             } catch (IOException e) {
092                 // ingnore on close
093             }
094             }
095         });
096     }
097  
098     @Override
099     public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
100     exc.printStackTrace();
101     try {
102         client.close();
103         latch.countDown();
104     } catch (IOException e) {
105         e.printStackTrace();
106     }
107     }
108 }

由于在AsyncTimeClientHandler中大量使用了内部匿名类,所以代码看起来稍微有些复杂,下面我们就对主要代码进行详细解说。

9-17行是构造方法,首先通过AsynchronousSocketChannel的open方法创建一个新的AsynchronousSocketChannel对象。然后跳到第36行,创建CountDownLatch进行等待,防止异步操作没有执行完成线程就退出。第37行通过connect方法发起异步操作,它有两个参数,分别如下:

1)      A attachment : AsynchronousSocketChannel的附件,用于回调通知时作为入参被传递,调用者可以自定义;

2)      CompletionHandler<Void,? super A> handler:异步操作回调通知接口,由调用者实现。

在本例程中,我们的两个参数都使用AsyncTimeClientHandler类本身,因为它实现了CompletionHandler接口。

接下来我们看异步连接成功之后的方法回调completed方法,代码第39行,我们创建请求消息体,对其进行编码,然后拷贝到发送缓冲区writeBuffer中,调用AsynchronousSocketChannel的write方法进行异步写,与服务端类似,我们可以实现CompletionHandler<Integer, ByteBuffer>接口用于写操作完成后的回调,代码第45-47行,如果发送缓冲区中仍有尚未发送的字节,我们继续异步发送,如果已经发送完成,则执行异步读取操作。

代码第64-97行是客户端异步读取时间服务器服务端应答消息的处理逻辑,代码第49行我们调用AsynchronousSocketChannel的read方法异步读取服务端的响应消息,由于read操作是异步的,所以我们通过内部匿名类实现CompletionHandler<Integer, ByteBuffer>接口,当读取完成被JDK回调时,我们构造应答消息。第56-63行我们从CompletionHandler的ByteBuffer中读取应答消息,然后打印结果。

第197-96行,当读取发生异常时,我们关闭链路,同时调用CountDownLatch的countDown方法让AsyncTimeClientHandler线程执行完毕,客户端退出执行。

需要指出的是,正如之前的NIO例程,我们并没有完整的处理网络的半包读写,当对例程进行功能测试的时候没有问题,但是,如果对代码稍加改造,进行压力或者性能测试,就会发现输出结果存在问题。

由于半包的读写会作为专门的小节在Netty的应用和源码分析章节进行详细讲解,在NIO的入门章节我们就不详细展开介绍,以便读者能够将注意力集中在NIO的入门知识上来。

下面的小节我们运行AIO版本的时间服务器程序,并通过打印线程堆栈的方式看下JDK回调异步Channel CompletionHandler的调用情况。 

目录
相关文章
|
1月前
|
传感器 物联网 微服务
Netty的源码分析和业务场景
【8月更文挑战第2天】Netty 是一款高性能的异步事件驱动网络框架,其源码深邃且复杂。通过采用Reactor模式与主从多线程设计,Netty能高效处理网络事件。例如,`NioEventLoop`负责I/O事件及任务执行,内置线程循环机制。内存管理方面,Netty提供高效内存池与`ByteBuf`类来减少开销并优化内存操作。在业务场景上,Netty广泛应用于分布式系统、微服务架构中的高效通信,以及实时通信场景如在线游戏和直播中的大量并发连接处理,同时也在物联网领域发挥重要作用,确保设备与服务器间稳定快速的数据传输。
|
1月前
|
网络协议 大数据 Linux
Netty的源码分析和业务场景
通过深入分析 Netty 的源码和理解其在不同业务场景下的应用,开发者可以更好地利用这一强大的网络编程框架,构建高效、稳定且可扩展的网络应用。
204 1
|
4月前
|
监控 网络协议 调度
Netty Review - 深入探讨Netty的心跳检测机制:原理、实战、IdleStateHandler源码分析
Netty Review - 深入探讨Netty的心跳检测机制:原理、实战、IdleStateHandler源码分析
317 0
|
4月前
|
消息中间件 编解码 Java
京东T8全面详解Java开源框架,透彻剖析尽在《Netty权威指南》
随着大规模分布式系统、大数据和流式计算框架的兴起,基于Java来构建这些系统已经成为主流,NIO编程和NIO框架在此期间得到了大规模的商用。在互联网领域,阿里的分布式服务框架Dubbo、RocketMQ,大数据的基础序列化和通信框架Avro, 以及很多开源的软件都已经开始使用Netty来构建高性能、分布式通信能力,Netty社区的活跃度也名列前茅。
|
编解码 前端开发 Java
源码分析Netty:核心组件及启动过程分析
本篇从实例出发,了解Netty核心组件的概念、作用及串联过程。从概念到设计原理,再到深入了解实现细节,从而能够清晰地掌握Netty的技术细节甚至存在的问题,才能最终更好地支持我们实际的各项业务。
335 0
|
编解码 架构师 网络协议
美团架构师熬夜整理:Netty权威指南2.0版+英雄传说项目
什么?你现在会觉得使用Netty编程的难度和工作量大了吗?Netty是一个令人惊讶的项目,在短短几年内成为众多Java高并发异步通信的首选框架。
116 0
|
开发框架 Java 中间件
到底什么是Java AIO?为什么Netty会移除AOI?一文搞懂AIO的本质!
Java AIO的这些不合常理的现象难免会令人心存疑惑。所以决定写这篇文章时,我不想只是简单的把AIO的概念再复述一遍,而是要透过现象,深入分析、思考和并理解Java AIO的本质。
812 1
|
Java
Netty入门到超神系列-BIO、NIO、AIO的认识
I/O概述 I/O其实是Input,Output的缩写,意思是输入和输出,比如:把磁盘中的一个文件读入Java内存中,站在Java内存的角度来看就是输入即input, 可以使用Java提供的输入流如:InputStream#read 来实现。如果是把一个段文本内容从Java内存中保存到磁盘上的某个文件,站在Java内存的角度来说就是输出,可以使用Java提供的输出流:OutputStream#write() 来实现。再比如把一个数据通过网络发送给其他服务器就是输出,或者从其他服务器接收一段数据就是输入。 I/O也是有很多种的,比如Java支持三种:BIO,NIO,AIO
300 0
|
Java API 调度
Netty组件EventLoopGroup和EventLoop源码分析
Netty组件EventLoopGroup和EventLoop源码分析
64 0
|
存储 缓存 NoSQL
跟着源码学IM(十一):一套基于Netty的分布式高可用IM详细设计与实现(有源码)
本文将要分享的是如何从零实现一套基于Netty框架的分布式高可用IM系统,它将支持长连接网关管理、单聊、群聊、聊天记录查询、离线消息存储、消息推送、心跳、分布式唯一ID、红包、消息同步等功能,并且还支持集群部署。
13433 1