`

在 Java 7 中体会 NIO.2 异步执行的快乐

 
阅读更多

简单介绍 Asynchronous I/O

JDK7 已经大致确定发布时间。JSR 203 提出很久了。2009.11.13,JDK7 M5(b76)已经发布。JSR 203 习惯上称为 NIO.2,主要包括新的:

  • 异步 I/O(简称 AIO);
  • Multicase 多播;
  • Stream Control Transport Protocol(SCTP);
  • 文件系统 API;
  • 以及一些 I/O API 的更新,例如:java.io.File.toPath,NetworkChannel 的完整抽象,等等。

本文将主要关注 AIO。AIO 包括 Sockets 和 Files 两部分的异步通道接口及其实现,并尽量使用操作系统提供的原生本地 I/O 功能进行实现。例如 Windows 版本的实现就使用了所谓的完成端口模型(IOCP)。其实 JDK 7 中 AIO 实现本质上说应该是 Proactor 模式的实现。Alexander Libman 提供 NIO 版本的 JProactor 的实现。NIO.2 版本 JProactor 正在进行。Grizzly 也已经提供新的基于 AIO 的实现。如果您只想检查这些最新的 API,NIO.2 项目的 Javadoc 站点只列出了 NIO.2 部分的 API。

  • AIO 的核心概念:发起非阻塞方式的 I/O 操作。当 I/O 操作完成时通知。
  • 应用程序的责任就是:什么时候发起操作? I/O 操作完成时通知谁?

AIO 的 I/O 操作,有两种方式的 API 可以进行:

  • Future 方式;
  • Callback 方式。

下面我们分别对这两种方式的 API 进行举例说明。

Future 方式

Future 方式:即提交一个 I/O 操作请求,返回一个 Future。然后您可以对 Future 进行检查,确定它是否完成,或者阻塞 IO 操作直到操作正常完成或者超时异常。使用 Future 方式很简单,比较典型的代码通常像清单 1 所示。

清单 1. 使用 Future 方式的代码示例
 AsynchronousSocketChannel ch = AsynchronousSocketChannel.open(); 

 // 连接远程服务器,等待连接完成或者失败
 Future<Void> result = ch.connect(remote); 
 // 进行其他工作,例如,连接后的准备环境,f.e. 
 //prepareForConnection(); 
 //Future 返回 null 表示连接成功
 if(result.get()!=null){ 
    // 连接失败,清理刚才准备好的环境,f.e. 
    //clearPreparation(); 
    return; 
 } 

 // 网络连接正常建立
 ... 

 ByteBuffer buffer = ByteBuffer.allocateDirect(8192); 

 // 进行读操作
 Future<Integer> result = ch.read(buffer); 

 // 此时可以进行其他工作,f.e. 
 //prepareLocalFile(); 
 // 然后等待读操作完成
 try { 
    int bytesRead = result.get(); 
    if(bytesRead==-1){ 
    // 返回 -1 表示没有数据了而且通道已经结束,即远程服务器正常关闭连接。
        //clear(); 
        return; 
    } 
    
    // 处理读到的内容,例如,写入本地文件,f.e. 
    //writeToLocolFile(buffer); 
 } catch (ExecutionExecption x) { 
    //failed 
 }

需要注意的是,因为 Future.get()是同步的,所以如果不仔细考虑使用场合,使用 Future 方式可能很容易进入完全同步的编程模式,从而使得异步操作成为一个摆设。如果这样,那么原来旧版本的 Socket API 便可以完全胜任,大可不必使用异步 I/O。

Callback 方式

Callback 方式:即提交一个 I/O 操作请求,并且指定一个 CompletionHandler。当异步 I/O 操作完成时,便发送一个通知,此时这个 CompletionHandler 对象的 completed 或者 failed 方法将会被调用,样例代码如清单 2 所示。

清单 2. 完成处理接口
 public interface CompletionHandler<V,A> { 
    // 当操作完成后被调用,result 参数表示操作结果,
    //attachment 参数表示提交操作请求时的参数。
    void completed(V result, A attachment); 

    // 当操作失败是调用,exc 参数表示失败原因。attachment 参数同上。
    void failed(Throwable exc, A attachment); 
 }

关于 Attachment 参数

Attachment 参数是不是看着十分眼熟呢?是的,NIO 中也使用类似的方法。当然 I/O 操作是不会对这个参数进行任何操作的,可以用于在不同的 CompletionHandler 对象之间进行通信。

  • V表示结果值的类型。对于异步网络通道的读写操作而言,这个结果值 V 都是整数类型,表示已经操作的卦数,如果是 -1,NIO.2 内核实现保证传递的 ByteBuffer参数不会有变化。
  • A表示关联到 I/O 操作的对象的类型。用于传递操作环境。通常会封装一个连接环境。
  • 如果成功则 completed 方法被调用。如果失败则 failed 方法被调用。

准备好 CompletionHandler 之后,如何使用 CompletionHandler 呢? AIO 提供四种类型的异步通道以及不同的 I/O 操作接受一个 CompletionHandler 对象,它们分别是:

  • AsynchronousSocketChannel:connect,read,write
  • AsynchronousFileChannel:lock,read,write
  • AsynchronousServerSocketChannel:accept
  • AsynchronousDatagramChannel:read,write,send,receive

本文重点关注 AsynchronousSocketChannel 的使用,首先简单浏览一下该类型的 API。

AsynchronousSocketChannel
 public abstract class AsynchronousSocketChannel 
    implements AsynchronousByteChannel, NetworkChannel
创建一个异步网络通道,并且绑定到一个默认组。
 public static AsynchronousSocketChannel open() throws IOException
将异步网络通道连接到远程服务器,使用指定的 CompletionHandler 听候完成通知。
 public abstract <A> void connect(SocketAddress remote, 
    A attachment, 
    CompletionHandler<Void,? super A> handler)
从异步网络通道读取数据到指定的缓冲区,使用指定的 CompletionHandler 听候完成通知。
 public final <A> void read(ByteBuffer dst, 
    A attachment, 
    CompletionHandler<Integer,? super A> handler)
向异步网络通道写缓冲区中的数据,使用指定的 CompletionHandler 听候完成通知。
 public final <A> void write(ByteBuffer src, 
    A attachment, 
    CompletionHandler<Integer,? super A> handler)

开始简单的异步 I/O 网络客户端程序

本文重点关注 AIO 的 socket 部分。接下来,我们以 AIO 方式的 FTP 客户端程序为例,开始体会异步执行的快乐。需要提醒的是:快乐和痛苦如影随行。好,那“痛并快乐着”吧。

为什么选择客户端编程的主题呢?

为什么选择客户端编程的主题呢?相对来说,其他文章和资料通常使用网络服务器作为主题。客户端的相对较少。使用 AIO 进行客户端编程有什么好处呢?想象一个 UI 的客户端程序,再看看时下流行的下载工具,线程一大堆,搞得你手上的工作做得不爽。好有一比,工厂希望有订单的时候多些工人,订单少的时候就少些工人。使用 AIO,程序通常可以使用更少的线程。

使用 AIO,可以想象一个在线视频播放的应用场景。使用异步 I/O 回调方式,可以这样描述一边下载视频一边播放的功能:

  1. 准备好网络连接
  2. 准备一个缓冲区,提交读操作希望下载部分视频内容,(这个读请求马上完成)
  3. 等待读请求完成操作,此时可以进行其他工作,比如播放广告
  4. 读操作真正完成,得到通知,CompletionHandler#completed 方法被调用,
  5. 启动另外的播放线程,从下载的缓冲区读取内容播放视频。
  6. 再准备一个另外的缓冲区,回到第二步

这样,第二步到第六步自动构成一个执行循环,但不是 while 之类的代码循环。

本文以 FTP 客户端程序为例,来阐述如何使用异步 I/O 进行网络程序的编写。

FTP 分为两个通道进行处理:控制通道和数据通道。

首先,开始 FTP 的控制通道的编程。FTP 的控制通道使用 telnet 行命令方式进行请求和响应处理。第一个例子不会复杂,我们只是连接到一个远程服务器,并且检查某个文件的大小,然后退出。基本步骤如下:

  1. 连接到 FTP 服务器。为了便于测试,本文将“攻击”ftp.gnu.org 服务器。
  2. 读取服务器的欢迎信息,检查远程服务器是否已经准备就绪。
  3. 如果服务器没有准备好,关闭连接,退出
  4. 如果服务器没有问题,发送登录命令。
  5. 检查登录命令结果。如果登录失败,转到第 8 步。
  6. 如果服务器没有问题,发送检查文件大小的命令。
  7. 检查命令结果。并且显示结果。
  8. 发送退出命令
  9. 关闭连接。

使用进行一个简单的设计:

第一个简单有问题的例子
 import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.nio.ByteBuffer; 
import java.nio.channels.AsynchronousSocketChannel; 
import java.nio.channels.CompletionHandler; 

public class FTPClient1 { 
    public static void main(String[] args) throws IOException { 
        // 第一个,创建异步网络通道
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(); 
        // 连接到服务器,以 ftp.gnu.org 为目标
        channel.connect(new InetSocketAddress("ftp.gnu.org", 21), channel, 
            // 使用连接完成的回调
            new CompletionHandler<Void, AsynchronousSocketChannel>() { 
            @Override 
            public void completed(Void result, AsynchronousSocketChannel attachment) {
                // 完成连接后,启动 FTP 的控制逻辑
                FTPClient1 client = new FTPClient1(); 
                client.start(attachment); 
            } 

            @Override 
            public void failed(Throwable exc, AsynchronousSocketChannel attachment) {
                exc.printStackTrace(); 
            } 
        }); 
        //connect 的调用异步执行,马上完成,阻止 JVM 退出
        System.in.read(); 
    } 

    AsynchronousSocketChannel channel; 

    public void start(AsynchronousSocketChannel channel) { 
        this.channel = channel; 
        // 准备读缓冲区
        ByteBuffer buffer = ByteBuffer.allocateDirect(128); 
        // 发出读操作请求,
        channel.read(buffer, buffer, 
        // 读操作完成后通知
        new CompletionHandler<Integer, ByteBuffer>() { 
            @Override 
            public void completed(Integer result, ByteBuffer attachment) { 
                if (result > 0) { 
                    // 简单处理读到的响应结果
                    int position = attachment.position() - 1; 
                    if (attachment.get(position - 1) == 13 && 
                    attachment.get(position) == 10) { 
                        if (isValidReply(attachment)) { 
                            attachment.flip(); 
                            showReply(attachment); 
                            if (getReplyCode(attachment) == 220) 
                                login(); 
                        } 
                    } else { 
                       // 继续读
                        FTPClient1.this.channel.read(attachment, attachment, this);
                    } 
                 } else { 
                     System.out.println("remote server closed"); 
                 } 
             } 

             @Override 
             public void failed(Throwable exc, ByteBuffer attachment) { 
                 exc.printStackTrace(); 
             } 
         }); 
     } 

     protected void login() { 
        // 准备写缓冲区
         String user = "user anonymous\r\n"; 
         ByteBuffer buffer = ByteBuffer.wrap(user.getBytes()); 
         // 发出写操作请求
         channel.write(buffer, buffer, 
         // 写操作完成通知
         new CompletionHandler<Integer, ByteBuffer>() { 
             @Override 
             public void completed(Integer result, ByteBuffer attachment) { 
                 if (attachment.hasRemaining()) 
                     channel.write(attachment, attachment, this); 
                 else { 
                     // channel.read(dst, attachment, handler); 
                     // readReply(); 
                     // 此处有问题
                 } 
             } 
             @Override 
             public void failed(Throwable exc, ByteBuffer attachment) { 
                 exc.printStackTrace(); 
             } 
         }); 
     } 

     protected void showReply(ByteBuffer attachment) { 
         while (attachment.hasRemaining()) 
             System.out.print((char) attachment.get()); 
     } 

     public static int getReplyCode(ByteBuffer buffer) { 
         return Character.digit(buffer.get(0), 10) * 100 + 
         Character.digit(buffer.get(1), 10) * 10 
                 + Character.digit(buffer.get(2), 10); 
     } 

     public static boolean isValidReply(ByteBuffer buffer) { 
         return buffer.get(3) == 32 && Character.isDigit(buffer.get(0))
         && Character.isDigit(buffer.get(1)) 
                 && Character.isDigit(buffer.get(2)); 
     } 
 }

问题:上面的代码中,login 方法中,完成 login 命令之后,如何继续?

答案是:不能继续。实际上,上面的例子代码回到了同步处理时代。典型的错误使用方式。痛。同时,CompletionHandler 的创建也成了问题,需要不停地创建这种类型的对象吗?痛!回顾前面提到的核心:应用程序的责任:什么时候发起操作? I/O 操作完成时通知谁?

就本例而言,FTPClient 本身应该承担应用程序的责任,正如 Client 名称所示,应该由 Client 来实现 CompletionHandler。 Client 负责发出 I/O 操作请求,I/O 操作完成通知 Client。正如世界上其他诸多问题一样,名称本身就是个问题。此处的 Client 的意思是真正的顾客。

可以想象另外一个场景:去一个有叫号机的银行大厅办理业务。“我”到银行,“我”决定办理个人业务,所以取个人业务的号码。然后看看前面等待的其他客人还不少,计算一下时间,“我”决定去隔壁馋嘴一个冰淇淋,回来后,在大厅到处晃晃,这时候,大厅广播通知,333 号顾客请到 3 号窗口办理业务,“我”听到了,检查一下号码,“我”持有 333 号,所以“我”去 3 号窗口。

上面这个场景中有几个非常重要的事实 “我”决定取个人业务号码,“我”听到了,“我”是顾客。 因此,上面例子应该让 FTPClient1 实现 CompletionHandler。这是对的。但是 FTPClient1 有两个重要的职责:发出读操作请求和发出写操作请求。需要两个 CompletionHandler 的角色,但是不能重复实现 CompletionHandler 接口,此时内部类是个不错的选择。修改上面的例子,如下:

第二个简单的例子
 import java.io.IOException; 
import java.net.InetSocketAddress; 
import java.nio.ByteBuffer; 
import java.nio.channels.AsynchronousSocketChannel; 
import java.nio.channels.CompletionHandler; 

public class FTPClient2 { 
    public static void main(String[] args) throws IOException { 
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(); 
        channel.connect(new InetSocketAddress("ftp.gnu.org", 21), channel, 
                new CompletionHandler<Void, AsynchronousSocketChannel>() { 
                    @Override 
                    public void completed(Void result, 
                        AsynchronousSocketChannel attachment) { 
                        FTPClient2 client = new FTPClient2(); 
                        client.start(attachment); 
                    } 

                    @Override 
                    public void failed(Throwable exc, 
                        AsynchronousSocketChannel attachment) { 
                        exc.printStackTrace(); 
                    } 
                }); 
        System.in.read(); 
    } 

    AsynchronousSocketChannel channel; 

    void readResponse() { 
        ByteBuffer buffer = ByteBuffer.allocateDirect(128); 
        read(buffer); 
    } 

    void read(ByteBuffer buffer) { 
        channel.read(buffer, buffer, reader); 
    } 
   // 使用内部类接收读操作完成通知    
    CompletionHandler<Integer, ByteBuffer> reader = 
    new CompletionHandler<Integer, ByteBuffer>() { 
        @Override 
        public void completed(Integer result, ByteBuffer attachment) { 
            if (result > 0) { 
                int position = attachment.position() - 1; 
                if (attachment.get(position - 1) == 13 && 
                    attachment.get(position) == 10) { 
                    if (isValidReply(attachment, 0)) { 
                        attachment.flip(); 
                        showReply(attachment); 
                        // 状态逻辑,处理响应
                        onReply(getReplyCode(attachment, 0)); 
                    } else { 
                        removeLine(attachment, position - 2); 
                        if (isValidReply(attachment, 0)) { 
                            attachment.flip(); 
                            showReply(attachment); 
                            onReply(getReplyCode(attachment, 0)); 
                        } else 
                            read(attachment); 
                    } 
                } else { 
                    if (!attachment.hasRemaining()) 
                        removeLine(attachment, position - 1); 
                    read(attachment); 
                } 
            } else { 
                System.out.println("remote server closed"); 
            } 
        } 
         @Override 
        public void failed(Throwable exc, ByteBuffer attachment) { 
            exc.printStackTrace(); 
        } 
    }; 

    public void start(AsynchronousSocketChannel channel) { 
        this.channel = channel; 
        readResponse(); 
    } 

    protected void onReply(int replyCode) { 
       // 按照前面定义好的步骤,处理状态逻辑
        if (replyCode == 220) 
            login(); 
        if (replyCode == 230) 
            writeCommand("size README"); 
        else if (replyCode == 213) 
            writeCommand("QUIT"); 
        else if (replyCode == 221) 
            try { 
                channel.close(); 
            } catch (IOException e) { 
                e.printStackTrace(); 
            } 
    } 

    void writeCommand(String string) { 
        System.out.print("==>"); 
        System.out.println(string); 
        ByteBuffer buffer = ByteBuffer.wrap((string + "\r\n").getBytes()); 
        write(buffer); 
    } 

    void write(ByteBuffer buffer) { 
        channel.write(buffer, buffer, writer); 
    } 
    // 使用内部类接收写操作完成通知
    CompletionHandler<Integer, ByteBuffer> writer = 
    new CompletionHandler<Integer, ByteBuffer>() { 
        @Override 
        public void completed(Integer result, ByteBuffer attachment) { 
            if (attachment.hasRemaining()) 
                channel.write(attachment, attachment, this); 
            else 
                readResponse(); 
        } 
        @Override 
        public void failed(Throwable exc, ByteBuffer attachment) { 
             exc.printStackTrace(); 
        } 
    }; 

    protected void login() { 
        String user = "user anonymous"; 
        writeCommand(user); 
    } 
    
   // 处理多行响应
    protected void removeLine(ByteBuffer buffer, int position) { 
        int limit = buffer.position(); 
        byte c; 
        while (position >= 0) { 
            c = buffer.get(position); 
            if (c == 13 || c == 10) { 
                showReply(buffer, position); 
                buffer.position(position + 1); 
                buffer.limit(limit); 
                buffer.compact(); 
                break; 
            } 
            position--; 
        } 
    } 
   // 针对多数 FTP 服务器的响应的偷懒的方法,不用费劲处理 String。
    protected void showReply(ByteBuffer buffer) { 
        while (buffer.hasRemaining()) 
            System.out.print((char) buffer.get()); 
    } 

    protected void showReply(ByteBuffer buffer, int position) { 
        for (int i = 0; i < position; i++) 
            System.out.print((char) buffer.get(i)); 
    } 

    public static int getReplyCode(ByteBuffer buffer, int start) { 
        return Character.digit(buffer.get(start), 10) * 100 + 
        Character.digit(buffer.get(start + 1), 10) * 10 
                + Character.digit(buffer.get(start + 2), 10); 
    } 

    public static boolean isValidReply(ByteBuffer buffer, int start) { 
        return buffer.get(start + 3) == 32 && 
        Character.isDigit(buffer.get(start)) 
                && Character.isDigit(buffer.get(start + 1)) 
                && Character.isDigit(buffer.get(start + 2)); 
    } 

    public static int findCRLF(ByteBuffer buffer, int start, int end) { 
        while (start < end) { 
            if (buffer.get(start++) == 13) { 
                if (start < end) { 
                    if (buffer.get(start) == 10) { 
                        return start + 1; 
                    } 
                } 
            } 
        } 
        return -1; 
    } 
}

对比两个代码,可以发现:修改后的代码的 onReply 方法,与上文中描述的需求步骤基本上一模一样。与使用阻塞模式编写的代码相比,应该更加简洁。阻塞模式下,你至少需要一个控制循环。似乎有点快乐了。


继续 FTP 的编程,升华完成通知类型

因为读写操作的使用远远多于其他类型的操作,所以重点考虑如何处理读写操作。回顾前面的第二个例子中的 reader 和 writer 成员,其实与对象编程理论和实践中的一个很重要的原理“单一职责原理”比较吻合。但是,如果需要写很多的网络程序,或者提供一个网络编程的框架(虽然现在有不少,例如:grizzly,JProactor),那么内部类的方式显然显得局限。

重用?重用什么?如何重用?

完成读写操作的回调次数?

发送 1K 的数据,到底需要几次回调呢?鬼知道。实际上鬼也不知道。

需要注意的是:AIO 读写操作并不保证操作一次全部完成。单个读写操作请求可能收到多次完成通知

多数网络应用程序发送响应或者请求消息,都需要将准备好的缓冲区全部内容发送出去。可以预见,前面的 writer 内部类成员可以独立,改编为抽象的 Writer 类型。这时候,前文中内部类的隐式引用好处就会失去,而且诞生出新的回调接口。

BUG ?太重复的劳动

很多情况下都应该检查 CompletionHandler.completed的 Integer 类型结果是否为 -1,看看是否通常已经关闭。

if(result==-1) {
// 通道已经关闭,执行 onChannelClosed
} else {
// 正常处理
}

实际上 AIO 的内核实现已经对 result 是否等于 -1 做出了判断,不知道基于何种考虑,completed 方法的 result 参数包含 -1 值。read() 方法的传统吗?这种处理直接导致人类的资源浪费:重复考虑这种判断,重复考虑判断后的处理。作者在 NIO2 的 dev list 中已经提起“诉讼”,但是看样子不果。

FTP 使用 telnet 协议的消息格式。消息以 <CRLF> 结束。 Telnet 协议家族的响应消息基本上都使用“code<SPACE>message<CRLF>"。

从处理 FTP 或者 telnet 协议家族的响应消息来看,前文的 reader 成员应该可以独立,至少可以抽象一个专门用于读取 Telnet 响应的 TelnetReader 类型。同样,也诞生出新的回调接口。对于 Reader 类型,还可以想象几种应用模式:

  • 读取指定长度的数据,SizeReader;
  • 一直读直到对方关闭通道,EOFReader。
  • 多数情况下读操作都会去检查读到的数据长度是否为 -1,以检测对方是否已经关闭通道

这样,对于 Reader 类型,某种程度的策略模式的应用需求已经浮现出来。

但是 Client 类型本身至少也可以实现一种类型的 CompletionHandler。如果这样,将产生一个争论:继承还是委托?很多情况下这实际上是口味的问题,并非优劣的选择。

同时,对于读写操作而言,CompletionHandler 的类型是确定的 Integer 类型,似乎增加一个新的派生接口 Callback<T> 更加满足需要。

新的读写操作回调接口
 public interface Callback<T> extends CompletionHandler<Integer, T> { 
	 @Override 
	 void completed(Integer result, T context); 

	 @Override 
	 void failed(Throwable cause, T context); 
 }

除上述考虑之外,最重要的一点是,有状态还是无状态。CompletionHandler 或者 Callback 接口本身无状态可言,但其实现存在有无状态的选择。AIO 内核并不关心 CompletionHandler 的 attachment 参数,内核不会使用也不会施加任何限制。但是实现类则大不同。有状态和无状态的设计将直接影响到 attachment 参数的使用。如您所看见,Callback 接口已经将 attachment 参数更名为 context。同时,因为 AsynchronousChannel 都需要 ByteBuffer,attachment 的使用也必须考虑 ByteBuffer 的使用方式。对于每一个读写操作而言,有三个因素是必须考虑的:AsynchronousChannel,ByteBuffer,attachment。普通应用程序也好,还是框架,实际上只考虑一个问题,就是如何组合这三个因素。某种程度上说,AIO 编程其实是 attachment编程,实不为过。怎一个痛字了得!

与此同时,因为诞生新的回调接口,预示着 Client 的层次在不断增加,也意味着 Client 的职责在进行分化。某些网络应用框架中的 filter 类型与此类似。

在没有更好的方案的时候,作者选择有状态方式的设计。

简单的有状态写操作类型
 public class BufferWriter implements Callback<WriteCallback> { 
	 private AsynchronousSocketChannel channel; 
	 private ByteBuffer buffer; 
	 private Charset charset; 

	 public BufferWriter(AsynchronousSocketChannel channel, Charset charset) { 
		 this.channel = channel; 
		 this.charset = charset; 
	 } 

	 public void write(String string, WriteCallback write) { 
		 buffer = ByteBuffer.wrap(string.getBytes(charset)); 
		 channel.write(buffer, write, this); 
	 } 

	 @Override 
	 public void completed(Integer result, WriteCallback context) { 
		 if (buffer.hasRemaining()) 
			 channel.write(buffer, context, this); 
		 else { 
			 buffer = null; 
			 context.writeCompleted(); 
		 } 
	 } 

	 @Override 
	 public void failed(Throwable cause, WriteCallback context) { 
		 buffer = null; 
		 context.writeFailed(cause); 
	 } 
 }
抽象读操作模板类型
 public abstract class AbstractReadCallback<T> implements Callback<T> { 
	 protected abstract void readCompleted(Integer result, T context); 

	 protected abstract void onChannelClose(T context); 

	 @Override 
	 public void completed(Integer result, T context) { 
	    // 重新分发回调通知
		 if (result > 0) 
			 readCompleted(result, context); 
		 else 
			 onChannelClose(context); 
	 } 
 }
简单的有状态读操作类型
public class TelnetReplyReader extends AbstractReadCallback
    <ResponseCallback<Reply>> { 
    private AsynchronousSocketChannel channel; 
    private CharsetDecoder decoder; 
    // 简单的 ByteBuffer 工厂,来自 JDK 的 corba 中的实现
    private ByteBufferPool pool; 
    private ByteBuffer buffer; 
    //FTP 响应数据对象
    private Reply reply = new Reply(); 

    public TelnetReplyReader(AsynchronousSocketChannel channel, 
    ByteBufferPool pool, Charset charset) { 
        this.channel = channel; 
        this.pool = pool; 
        decoder = charset.newDecoder(); 
    } 

    public void read(ResponseCallback<Reply> protocol) { 
        reply.reset(); 
        if (buffer == null) 
            buffer = pool.get(1024); 
        buffer.clear(); 
        channel.read(buffer, protocol, this); 
    } 

    @Override 
    protected void onChannelClose(ResponseCallback<Reply> context) { 
        try { 
            channel.close(); 
        } catch (IOException e) { 
            // ignore; 
        } 
        // 转换为特定的异常类型
        failed(new ClosedChannelException(), context); 
    } 

    @Override 
    protected void readCompleted(Integer result,ResponseCallback<Reply> context){
        ByteBuffer buffer = this.buffer; 
        try { 
           // 响应代码的处理逻辑,直到获得有效的响应代码,否则哭到长城
            int position = buffer.position(); 
            if (buffer.get(position - 2) == 13 && 
                buffer.get(position - 1) == 10) { 
                // Yes check reply code; 
                if (findReplyCode(buffer, position - 2)) { 
                    // buffer position at the code first char; 
                    int first = buffer.position(); 
                    reply.code = getReplyCode(buffer, first); 
                    if (first > 0) { 
                        buffer.flip(); 
                        reply.other.append(decoder.decode(buffer)); 
                    } 
                    buffer.limit(position - 2); 
                    buffer.position(first + 4); 
                    reply.message = decoder.decode(buffer).toString(); 
                    returnBuffer(); 
                    context.onResponse(reply); 
                    return; 
                } 
                buffer.flip(); 
                reply.other.append(decoder.decode(buffer)); 
                buffer.clear(); 
                channel.read(buffer, context, this); 
                return; 
            } 

            // No reply code, consider cache other message 
            if (buffer.hasRemaining()) { 
                channel.read(buffer, context, this); 
                return; 
            } 

         // Have to cache some message, but may be have reply code, so just check CRLF;
            int index = findLF(buffer, position - 2); 
            if (index == -1) { 
                buffer.flip(); 
                reply.other.append(decoder.decode(buffer)); 
            } else { 
                buffer.position(0).limit(index + 1); 
                reply.other.append(decoder.decode(buffer)); 
                buffer.position(index); 
            } 
            buffer.limit(position); 
            buffer.compact(); 
            channel.read(buffer, context, this); 
        } catch (CharacterCodingException ex) { 
            failed(ex, context); 
        } 
    } 

    @Override 
    public void failed(Throwable cause, ResponseCallback<Reply> context) { 
        returnBuffer(); 
        context.failed(cause); 
    } 

    private void returnBuffer() { 
        pool.releaseBuffer(buffer); 
        buffer = null; 
    } 
...
使用有状态读写操作类型的控制类
 public class FTPClient implements ResponseCallback<Reply>, 
 WriteCallback, CommandProvider { 
	 private TelnetReplyReader reader; 
	 private BufferWriter writer; 
	 private Semaphore semaphore = new Semaphore(0); 
	 // 传输通道的处理环境
	 private TransferContext transferContext; 

	 protected void start(Context context, AsynchronousSocketChannel channel) { 
		 InetSocketAddress remote; 
		 try { 
			 remote = (InetSocketAddress) channel.getRemoteAddress(); 
		 } catch (IOException e) { 
			 failed(e); 
			 return; 
		 } 
		 InetSocketAddress local; 
		 try { 
			 local = (InetSocketAddress) channel.getLocalAddress(); 
		 } catch (IOException e) { 
			 failed(e); 
			 return; 
		 } 
		 Charset charset = Charset.forName("UTF-8"); 
		 reader = new TelnetReplyReader(channel, context.pool(), charset); 
		 writer = new BufferWriter(channel, charset); 
		 // 发起读操作请求
		 reader.read(this); 
		 // 同时,预备传输通道环境
		 transferContext = new SimpleTransferContext(context, 
		  remote.getAddress(), local.getAddress()); 
	 } 

	 @Override 
	 public void onResponse(Reply reply) { 
	    // 简单的响应处理逻辑
		 try { 
			 transferContext.check(reply); 
		 } catch (Throwable ex) { 
			 ex.printStackTrace(); 
		 } 
		 // If reply not process right, just pending any advance operation. 
		 if (reply.code / 100 == 1) 
			 reader.read(this); 
		 else 
			 semaphore.release(); 
	 } 

	 @Override 
	 public void writeCompleted() { 
	    //FTP 规则,发出请求命令后,开始等待对方的响应
		 reader.read(this); 
	 } 
	 ...

除协议相关的部分代码,其余的看上去还蛮简单,似乎抽象 Reader 和 Writer 的代价值得的。上面代码中的 Context,Reply 等小类型,可以在完整的源代码中检查。


继续 FTP,处理传输通道

本文之所以选择 FTP 作为 AIO 的实践例子,FTP 的控制通道必须协调单独的数据传输通道。不仅如此,使用 Port 方式的话,客户端程序还需要建立一个简单的网络服务器。

关于防火墙

本文的编程环境是 Windows7,由于防火墙的原因,忽略服务器方式。如果网络程序出现故障,防火墙是否为问题的根源,可以优先考虑。作者在实践过程中,曾经遭遇到这样的问题。Windows 7 自带的防火墙识别 FTP 的 PASV 命令,并且阻止该命令的执行。而且,AIO 的内核实现使用转换后系统错误消息作为异常消息,会让你痛的哭。但是,请放心,长城不会倒。

上文中,我们尽量回避建立网络连接的 CompletionHandler 的再处理问题。FTP 的数据传输通道

  • 要么使用服务器方式,使用 AsynchronousServerSocketChannel.accept 方法
  • 要么使用客户端方式,使用 AsynchronousSocketChannel.connect 方法,与前文类似

延续上文的处理思路,继续抽象用于 connect 的 CompletionHandler 类型。与前不同的是,该连接回调类型使用无状态方式设计。该例演示下载文件的处理。

无状态类型的连接回调类型
 public class SocketConnector implements Connector<Object[]> { 
	 public void connect(InetSocketAddress remote, ConnectionCallback client) 
	 throws IOException { 
	     // 创建新的异步网络通道
		 AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(); 
		 // 无状态方式处理,将所有需要的参数打包为单个 attachment 参数
		 Object[] attachment = { client, remote, channel }; 
		 // 启动连接操作
		 channel.connect(remote, attachment, this); 
	 } 

	 public void connect(InetSocketAddress remote, InetSocketAddress local, 
	 ConnectionCallback client)throws IOException { 
		 AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(); 
		 // 绑定本地网络地址,对于客户端而言,通常是 IP,对于服务器而言,一定需要端口号
		 channel.bind(local); 
		 Object[] attachment = { client, remote, channel }; 
		 channel.connect(remote, attachment, this); 
	 } 

	 @Override 
	 public void completed(Void result, Object[] attachment) { 
	     // 连接完成,通知 Client 启动协议控制逻辑
		 ((ConnectionCallback) attachment[0]). 
		 start((AsynchronousSocketChannel) attachment[2]); 
	 } 

	 @Override 
	 public void failed(Throwable cause, Object[] attachment) { 
		 ((ConnectionCallback) attachment[0]). 
		 connectFailed(new Exception(attachment[1].toString(), cause)); 
	 } 
 }

但是,有些 FTP 服务器要求数据传输通道必须使用与控制通道相同的 ip 地址,导致连接必须知道并保持控制通道的 ip 地址。唉!又痛到有状态方式了。

有状态类型的连接回调类型
public class TransferConnector extends SocketConnector { 
     private InetAddress localAddress; 
     private InetAddress remoteAddress; 

     public TransferConnector(InetAddress remoteAddress, InetAddress localAddress) {
         this.remoteAddress = remoteAddress; 
         this.localAddress = localAddress; 
     } 

     protected InetSocketAddress createRemoteAddress(int port) { 
         return new InetSocketAddress(remoteAddress, port); 
     } 

     protected InetSocketAddress createLocalAddress() { 
         return new InetSocketAddress(localAddress, 0); 
     } 

     public void connect(int port, ConnectionCallback client) throws IOException {
         if (port < 1) 
             throw new IOException("Error remote server port number: " + port); 
         super.connect(createRemoteAddress(port), createLocalAddress(), client); 

     } 
 }
使用连接回调类型建立数据传输通道
 public class SimpleTransferContext 
  implements TransferContext, ConnectionCallback, FileLockCallback { 
    ... 
	 // 使用单独的传输连接回调对象再次进行连接完成通知
    connector = new TransferConnector(remoteAddress, localAddress); 
	 ... 
	 // 发起传输通道的连接操作请求
	 case RETR: 
			 connector.connect(port, this); 
			 // clear for next time 
			 port = 0; 
			 ... 

    // 传输通道连接完成 
	 @Override 
	 public void start(AsynchronousSocketChannel channel) { 
		 this.channel = channel; 
	 }

因为涉及到文件的处理,FTP 的数据传输通道起始控制看起来相当简单。快乐其实是很简单的东西。


继续 FTP,使用 AIO 的异步文件操作

AsynchronousFileChannel 没有 connect 方法,但是有一个类似的方法 lock。JDK7 中该方法的声明如下:

异步文件通道的 lock 方法 API
 public abstract <A> void lock(long position, 
                              long size, 
                              boolean shared, 
                              A attachment, 
                              CompletionHandler<FileLock,? super A> handler)
无状态的文件连接回调类型
public class FileLocker implements CompletionHandler<FileLock, FileLockCallback> {
     public void lock(String filename, long position, long size, 
        boolean shared, FileLockCallback client, 
         OpenOption... options) throws IOException { 
         // 使用新的 AIO 中的 Path API 
         Path path = Paths.get(filename); 
         // 创建异步文件通道对象
         AsynchronousFileChannel file = AsynchronousFileChannel.open(path, options);
         // 锁定要写的区域
         file.lock(position, size, shared, client, this); 
     } 

     @Override 
     public void completed(FileLock result, FileLockCallback attachment) {
         // 文件锁(或者文件连接)完成通知传输通道环境可以工作
         attachment.start(result); 
     } 

     @Override 
     public void failed(Throwable cause, FileLockCallback attachment) {
         attachment.lockFailed(cause); 
     } 
}
使用文件连接回调类型建立文件通道
 public class SimpleTransferContext 
 implements TransferContext, ConnectionCallback, FileLockCallback {
 ... 
  
    @Override 
    public void start(FileLock fileLock) { 
        this.fileLock = fileLock; 
        // at here socket channel already prepared 
        // 启动下载过程
        startDownload(); 
    } 

    private Downloader download; 

    private void startDownload() { 
        download = new Downloader(context, channel, fileLock, size);
        channel = null; 
        fileLock = null; 
        download.run(); 
    } 
    
    public void check(Reply reply) { 
        if (currentCommand == null) { 
            System.out.println(reply); 
            return; 
        } 
        int code = reply.code; 
        String message = reply.message; 
        switch (currentCommand) { 
        case SIZE... 
        case RETR: 
            if (code == 150) { 
                // 150 Opening BINARY mode data connection for README (1765 bytes).
                int end = message.lastIndexOf(')'); 
                if (end != -1) { 
                    int start = message.lastIndexOf('(', end - 1); 
                    if (start != -1) { 
                       //RETR 命令响应正确,检查本地文件,预备下载
                        lockFile(checkSize(message.substring(start + 1, end - 6)));
                        break; 
                    } 
                } 
                ... 
    } 
    
    protected void lockFile(long size) { 
        try { 
            locker.lock(filename, 0, size, false, this, 
               StandardOpenOption.CREATE, 
                StandardOpenOption.READ, 
                StandardOpenOption.WRITE); 
        } catch (IOException e) { 
            e.printStackTrace(); 
        } 
    }

当 FTP 的 RETR 命令正确响应后,准备下载文件。首先准备好要写入的本地文件通道,锁住文件。文件锁完成后,创建新的 Downloader 对象,开始真正的下载操作。

使用文件连接回调类型建立文件通道
 public abstract class Transfer { 
    // 用于 Socket 和 File 读写操作使用的 ByteBuffer 的交换队列
	 protected BlockingDeque<ByteBuffer> bufferQueue = 
	 new LinkedBlockingDeque<ByteBuffer>(); 
	 protected Context context; 

	 public Transfer(Context context) { 
		 this.context = context; 
	 } 

	 public ByteBuffer getBuffer(int size) { 
		 return context.pool().get(size); 
	 } 

	 protected void releaseBuffer(ByteBuffer buffer) { 
		 context.pool().releaseBuffer(buffer); 
	 } 
 }
下载实现,读和写
 public class Downloader extends Transfer 
    implements ReadCallback, FileWriteCallback2, Runnable { 
	 // 读入指定长度内容的回调对象,处理网络内容
	 private SizeReader reader; 
	 // 写入指定长度内容的回调对象,处理文件内容
	 private FileWriter2 writer; 
	 private AtomicBoolean writable = new AtomicBoolean(true); 
	 // 用于显示网络数据传输速率的工具
	 private ConsoleProgress progress = new ConsoleProgress(); 

	 public Downloader(Context context, AsynchronousSocketChannel socket, 
	 FileLock fileLock, long size) { 
		 super(context); 
		 reader = new SizeReader(socket, size, this); 
		 writer = new FileWriter2(fileLock, this); 
		 progress.reset(size); 
	 } 

	 @Override 
	 public void run() { 
		 reader.read(); 
	 } 

	 @Override 
	 public void writeCompleted(ByteBuffer buffer) { 
	    // 一个缓冲区写入文件完毕
		 releaseBuffer(buffer); 
		 buffer = bufferQueue.poll(); 
		 if (buffer != null) 
		    // 如果网络已经读好一个缓冲区,继续写入文件
			 writer.write(buffer); 
		 else 
		    // 否则清除写状态
			 writable.set(true); 
	 } 

	 @Override 
	 public void readCompletedBytes(Integer bytes, long start, long end) { 
	    // 显示网络传输进度
		 progress.update(bytes, start, end); 
		 progress.run(); 
	 } 

	 @Override 
	 public void completedReadBuffer(ByteBuffer buffer) { 
		 if (writable.compareAndSet(true, false)) { 
		    // 从网络下载了一个缓冲区的内容,如果写文件空闲,通知写文件
			 writer.write(buffer); 
		 } else { 
		    // 如果文件正在写,将当前缓冲区放入后备队列
			 bufferQueue.offer(buffer); 
		 } 
	 } 

	 @Override 
	 public void writeCompleted() { 
		 System.out.println("file saved OK"); 
	 } 

	 @Override 
	 public void readCompleted() { 
		 System.out.println("file transfer OK"); 
	 } 
	 ...
  • Downloader 使用 SizeReader 读取网络数据。SizeReader 使用自主的缓冲区申请,不需要调用者传递 ByteBuffer 参数。
  • Downloader 使用 FileWriter2 写文件内容。FileWriter2 使用一次性写完外部传递的缓冲区的策略。需要调用者传递 ByteBuffer 参数。

限于篇幅,具体实现可以在源代码中检查。


 

源于:http://www.ibm.com/developerworks/cn/java/j-lo-nio2/index.html

分享到:
评论

相关推荐

    java NIO.zip

    自Java 1.4版本引入NIO后,它为Java开发者提供了更高效的数据传输方式,尤其是在处理大量并发连接时。NIO的核心在于通道(Channels)和缓冲区(Buffers)的概念,与传统的流(Streams)有所不同。 1. **通道...

    Pro Java 7 NIO.2.pdf

    根据提供的文件信息,“Pro Java 7 NIO.2.pdf”由Anghel Leonard于2011年编写,主要介绍了Java 7中的新输入/输出(NIO)API,特别是NIO.2(JSR 203)所带来的增强功能。这本书通过一系列章节详细讲解了如何使用NIO.2...

    Java IO, NIO and NIO.2(Apress,2015)

    Java I/O, NIO, and NIO.2 is a power-packed book that accelerates your mastery of Java's various I/O APIs. In this book, you'll learn about classic I/O APIs (File, RandomAccessFile, the stream classes ...

    Java IO, NIO and NIO.2

    Java IO、NIO以及NIO.2是Java中用于处理输入/输出操作的三种主要机制。本书《Java IO, NIO and NIO.2》旨在深入浅出地介绍这些机制,同时书中内容均为英文。接下来将详细介绍这些知识点。 **Java IO** Java IO是...

    Apress.Pro.Java.7.NIO.2.2011

    《Apress.Pro.Java.7.NIO.2.2011》这本书专注于讲解Java 7中的非阻塞I/O(Non-blocking I/O, NIO)和NIO 2的高级特性,是Java开发者深入理解这一关键领域的重要参考资料。NIO在Java平台中扮演着至关重要的角色,特别...

    nio.rar_FastCopyFile.java_NIO_UseFloatBuffer.java_java nio_文件锁

    在给定的压缩包文件中,我们关注的是"FastCopyFile.java"、"UseFloatBuffer.java"以及NIO中的文件锁功能。 首先,让我们详细了解一下`FastCopyFile.java`。这个文件很可能是一个示例程序,演示了如何使用Java NIO...

    Java IO, NIO and NIO.2 原版pdf by Friesen

    New I/O (NIO), and NIO.2 categories. You learn what each category offers in terms of its capabilities, and you also learn about concepts such as paths and Direct Memory Access. Chapters 2 through 5 ...

    Java.NIO资源下载资源下载

    Java NIO 是 Java 平台的一个重要特性,首次出现在 Java 1.4 版本中。它为 Java 开发者提供了一套全新的 I/O 处理方式,相比于传统的 I/O API,新版本提供了更高效、更灵活的解决方案。本书《Java™ NIO》由 Ron ...

    Java NIO 中英文版 + Pro Java 7 NIO.2

    `Pro Java 7 NIO.2`这本书由Anghel Leonard著,深入探讨了Java NIO.2 API,这是Java 7引入的进一步扩展,包括: 1. **文件系统API增强**:新增了AsynchronousFileChannel,支持异步文件操作,可以在后台线程中执行...

    java nio.pdf

    java nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava nio.pdfjava ...

    ProJava7NIO.2PDFBooks.pdf 英文原版

    Pro Java 7 NIO.2 – PDF Books

    优雅的操作文件:java.nio.file 库介绍.pdf

    `java.nio.file.Paths`模块提供静态方法创建`Path`对象,代表文件系统中的路径。`Path`类则包含一系列方法,用于处理和操作路径。例如: ```java import java.nio.file.Path; import java.nio.file.Paths; public ...

    Java I/O, NIO and NIO.2

    NIO.2,又称为New I/O 2或Java NIO 2,是在Java 7中引入的进一步扩展,主要增加了文件系统操作的支持,包括异步I/O操作、文件通道、文件锁定以及对符号链接的支持。AsynchronousFileChannel是NIO.2中新增的异步I/O...

    java-nio.rar_NIO_java nio

    Java NIO,全称为New Input/Output,是Java在1.4版本引入的一个新特性,是对传统的I/O模型的一种改进。...通过阅读"java-nio.pdf"这份文档,你将能够深入学习Java NIO技术及其在异步连接池中的应用。

    java org.apache.http.nio jar包

    找了好久,终于找到了,java刷新同步获取网络资源

    java.io:clojure.java.io 的 JK7 java.nio.file.Path 兼容性

    java.io clojure.java.io 的 JK7 java.nio.file.Path 兼容性依赖信息该库托管在 Releases 上。 依赖: [me.moocar/java.io " 0.1.0 " ]用法是 JDK7 中引入的文件路径的抽象。 这个库提供了和 Paths 之间的兼容性。 ...

Global site tag (gtag.js) - Google Analytics