JDK AIO編程

時間 2019-11-12

標籤 jdk aio 編程欄目 Java 简体版

原文原文鏈接

NIO2.0引入了新的異步通道的概念，並提供了異步文件通道和異步套接字通道的實現。異步通道提供兩種方式獲取獲取操做結果。java

經過java.util.concurrent.Future類來表示異步操做的結果；
在執行異步操做的時候傳入一個java.nio.channels。

CompletionHandler接口的實現類做爲操做完成的回調。編程

NIO2.0的異步套接字通道是真正的異步非阻塞I/O，它對應UNIX網絡編程中的事件驅動I/O（AIO），它不須要經過多路複用器（Selector）對註冊的通道進行輪詢操做便可實現異步讀寫，從而簡化了NIO的編程模型。數組

服務端代碼示例：

import java.io.IOException;

public class TimeServer {

    public static void main(String[] args) throws IOException {
        int port = 8080;
        if (args != null && args.length > 0) {
            try {
                port = Integer.valueOf(args[0]);
            } catch (NumberFormatException e) {
                // 採用默認值
            }
        }
        //首先建立異步的時間服務器處理類，而後啓動線程將AsyncTimeServerHandler啓動
        AsyncTimeServerHandler timeServer = new AsyncTimeServerHandler(port);
        new Thread(timeServer, "AIO-AsyncTimeServerHandler-001").start();
    }
}


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;


public class AsyncTimeServerHandler implements Runnable {


    CountDownLatch latch;
    AsynchronousServerSocketChannel asynchronousServerSocketChannel;

    public AsyncTimeServerHandler(int port) {
        //在構造方法中，咱們首先建立一個異步的服務端通道AsynchronousServerSocketChannel，
        //而後調用它的bind方法綁定監聽端口，若是端口合法且沒被佔用，綁定成功，打印啓動成功提示到控制檯。
        try {
            asynchronousServerSocketChannel = AsynchronousServerSocketChannel.open();
            asynchronousServerSocketChannel.bind(new InetSocketAddress(port));
            System.out.println("The time server is start in port : " + port);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        //在線程的run方法中，初始化CountDownLatch對象，
        //它的做用是在完成一組正在執行的操做以前，容許當前的線程一直阻塞。
        //在本例程中，咱們讓線程在此阻塞，防止服務端執行完成退出。
        //在實際項目應用中，不須要啓動獨立的線程來處理AsynchronousServerSocketChannel，這裏僅僅是個demo演示。
        latch = new CountDownLatch(1);
        doAccept();
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //用於接收客戶端的鏈接，因爲是異步操做，
    //咱們能夠傳遞一個CompletionHandler＜AsynchronousSocketChannel,? super A＞類型的handler實例接收accept操做成功的通知消息，
    //在本例程中咱們經過AcceptCompletionHandler實例做爲handler來接收通知消息，
    public void doAccept() {
        asynchronousServerSocketChannel.accept(this, new CompletionHandler<AsynchronousSocketChannel, AsyncTimeServerHandler>() {
            @Override
            public void completed(AsynchronousSocketChannel result,
                                  AsyncTimeServerHandler attachment) {
                //咱們從attachment獲取成員變量AsynchronousServerSocketChannel，而後繼續調用它的accept方法。
                //在此可能會心存疑惑：既然已經接收客戶端成功了，爲何還要再次調用accept方法呢？
                //緣由是這樣的：當咱們調用AsynchronousServerSocketChannel的accept方法後，
                //若是有新的客戶端鏈接接入，系統將回調咱們傳入的CompletionHandler實例的completed方法，
                //表示新的客戶端已經接入成功，由於一個AsynchronousServerSocket Channel能夠接收成千上萬個客戶端，
                //因此咱們須要繼續調用它的accept方法，接收其餘的客戶端鏈接，最終造成一個循環。
                //每當接收一個客戶讀鏈接成功以後，再異步接收新的客戶端鏈接。
                attachment.asynchronousServerSocketChannel.accept(attachment, this);
                //鏈路創建成功以後，服務端須要接收客戶端的請求消息，
                //建立新的ByteBuffer，預分配1M的緩衝區。
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                //經過調用AsynchronousSocketChannel的read方法進行異步讀操做。
                //下面咱們看看異步read方法的參數。
                //ByteBuffer dst：接收緩衝區，用於從異步Channel中讀取數據包；
                //A attachment：異步Channel攜帶的附件，通知回調的時候做爲入參使用；
                //CompletionHandler＜Integer,? super A＞：接收通知回調的業務handler，本例程中爲ReadCompletionHandler。
                result.read(buffer, buffer, new ReadCompletionHandler(result));
            }

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

}

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;

public class ReadCompletionHandler implements CompletionHandler<Integer, ByteBuffer> {

    private AsynchronousSocketChannel channel;

    public ReadCompletionHandler(AsynchronousSocketChannel channel) {
        //將AsynchronousSocketChannel經過參數傳遞到ReadCompletion Handler中看成成員變量來使用
        //主要用於讀取半包消息和發送應答。本例程不對半包讀寫進行具體說明
        if (this.channel == null)
            this.channel = channel;
    }

    @Override
    public void completed(Integer result, ByteBuffer attachment) {
        //讀取到消息後的處理，首先對attachment進行flip操做，爲後續從緩衝區讀取數據作準備。
        attachment.flip();
        //根據緩衝區的可讀字節數建立byte數組
        byte[] body = new byte[attachment.remaining()];
        attachment.get(body);
        try {
            //經過new String方法建立請求消息，對請求消息進行判斷，
            //若是是"QUERY TIME ORDER"則獲取當前系統服務器的時間，
            String req = new String(body, "UTF-8");
            System.out.println("The time server receive order : " + req);
            String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(req) ? new java.util.Date(
                    System.currentTimeMillis()).toString() : "BAD ORDER";
            //調用doWrite方法發送給客戶端。
            doWrite(currentTime);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
    }

    private void doWrite(String currentTime) {
        if (currentTime != null && currentTime.trim().length() > 0) {
            //首先對當前時間進行合法性校驗，若是合法，調用字符串的解碼方法將應答消息編碼成字節數組，
            //而後將它複製到發送緩衝區writeBuffer中，
            byte[] bytes = (currentTime).getBytes();
            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            //最後調用AsynchronousSocketChannel的異步write方法。
            //正如前面介紹的異步read方法同樣，它也有三個與read方法相同的參數，
            //在本例程中咱們直接實現write方法的異步回調接口CompletionHandler。
            channel.write(writeBuffer, writeBuffer,
                    new CompletionHandler<Integer, ByteBuffer>() {
                        @Override
                        public void completed(Integer result, ByteBuffer buffer) {
                            //對發送的writeBuffer進行判斷，若是還有剩餘的字節可寫，說明沒有發送完成，須要繼續發送，直到發送成功。
                            if (buffer.hasRemaining())
                                channel.write(buffer, buffer, this);
                        }

                        @Override
                        public void failed(Throwable exc, ByteBuffer attachment) {
                            //關注下failed方法，它的實現很簡單，就是當發生異常的時候，對異常Throwable進行判斷，
                            //若是是I/O異常，就關閉鏈路，釋放資源，
                            //若是是其餘異常，按照業務本身的邏輯進行處理,若是沒有發送完成，繼續發送.
                            //本例程做爲簡單demo，沒有對異常進行分類判斷，只要發生了讀寫異常，就關閉鏈路，釋放資源。
                            try {
                                channel.close();
                            } catch (IOException e) {
                                // ingnore on close
                            }
                        }
                    });
        }
    }

    @Override
    public void failed(Throwable exc, ByteBuffer attachment) {
        try {
            this.channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客戶端代碼示例：

public class TimeClient {

    public static void main(String[] args) {
        int port = 8080;
        //經過一個獨立的I/O線程建立異步時間服務器客戶端handler，
        //在實際項目中，咱們不須要獨立的線程建立異步鏈接對象，由於底層都是經過JDK的系統回調實現的.
        new Thread(new AsyncTimeClientHandler("127.0.0.1", port), "AIO-AsyncTimeClientHandler-001").start();
    }
}

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.CountDownLatch;

public class AsyncTimeClientHandler implements CompletionHandler<Void, AsyncTimeClientHandler>, Runnable {

    private AsynchronousSocketChannel client;
    private String host;
    private int port;
    private CountDownLatch latch;

    //首先經過AsynchronousSocketChannel的open方法建立一個新的AsynchronousSocketChannel對象。
    public AsyncTimeClientHandler(String host, int port) {
        this.host = host;
        this.port = port;
        try {
            client = AsynchronousSocketChannel.open();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        //建立CountDownLatch進行等待，防止異步操做沒有執行完成線程就退出。
        latch = new CountDownLatch(1);
        //經過connect方法發起異步操做，它有兩個參數，
        //A attachment：AsynchronousSocketChannel的附件，用於回調通知時做爲入參被傳遞，調用者能夠自定義；
        //CompletionHandler＜Void,? super A＞ handler：異步操做回調通知接口，由調用者實現。
        client.connect(new InetSocketAddress(host, port), this, this);
        try {
            latch.await();
        } catch (InterruptedException e1) {
            e1.printStackTrace();
        }
        try {
            client.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //異步鏈接成功以後的方法回調——completed方法
    @Override
    public void completed(Void result, AsyncTimeClientHandler attachment) {
        //建立請求消息體，對其進行編碼，而後複製到發送緩衝區writeBuffer中，
        //調用Asynchronous SocketChannel的write方法進行異步寫。
        //與服務端相似，咱們能夠實現CompletionHandler ＜Integer, ByteBuffer＞接口用於寫操做完成後的回調。
        byte[] req = "QUERY TIME ORDER".getBytes();
        ByteBuffer writeBuffer = ByteBuffer.allocate(req.length);
        writeBuffer.put(req);
        writeBuffer.flip();
        client.write(writeBuffer, writeBuffer,
                new CompletionHandler<Integer, ByteBuffer>() {
                    @Override
                    public void completed(Integer result, ByteBuffer buffer) {
                        //若是發送緩衝區中仍有還沒有發送的字節，將繼續異步發送，若是已經發送完成，則執行異步讀取操做。
                        if (buffer.hasRemaining()) {
                            client.write(buffer, buffer, this);
                        } else {
                            //客戶端異步讀取時間服務器服務端應答消息的處理邏輯
                            ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                            //調用AsynchronousSocketChannel的read方法異步讀取服務端的響應消息。
                            //因爲read操做是異步的，因此咱們經過內部匿名類實現CompletionHandler＜Integer，ByteBuffer＞接口,
                            //當讀取完成被JDK回調時，構造應答消息。
　　　　　　　　　　　　　　　　　client.read(readBuffer,readBuffer, new CompletionHandler<Integer, ByteBuffer>() {
                                        @Override
                                        public void completed(Integer result,ByteBuffer buffer) {
                                            //從CompletionHandler的ByteBuffer中讀取應答消息，而後打印結果。
                                            buffer.flip();
                                            byte[] bytes = new byte[buffer.remaining()];
                                            buffer.get(bytes);
                                            String body;
                                            try {
                                                body = new String(bytes,"UTF-8");
                                                System.out.println("Now is : " + body);
                                                latch.countDown();
                                            } catch (UnsupportedEncodingException e) {
                                                e.printStackTrace();
                                            }
                                        }

                                        @Override
                                        public void failed(Throwable exc, ByteBuffer attachment) {
                                            //當讀取發生異常時，關閉鏈路，
                                            //同時調用CountDownLatch的countDown方法讓AsyncTimeClientHandler線程執行完畢，客戶端退出執行。
                                            try {
                                                client.close();
                                                latch.countDown();
                                            } catch (IOException e) {
                                                // ingnore on close
                                            }
                                        }
                                    });
                        }
                    }

                    @Override
                    public void failed(Throwable exc, ByteBuffer attachment) {
                        try {
                            client.close();
                            latch.countDown();
                        } catch (IOException e) {
                            // ingnore on close
                        }
                    }
                });
    }

    @Override
    public void failed(Throwable exc, AsyncTimeClientHandler attachment) {
        exc.printStackTrace();
        try {
            client.close();
            latch.countDown();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

須要指出的是，正如以前的NIO例程，咱們並無完整的處理網絡的半包讀寫，在對例程進行功能測試的時候沒有問題，可是，若是對代碼稍加改造，進行壓力或者性能測試，就會發現輸出結果存在問題。服務器

經過打印線程堆棧的方式看下JDK回調異步Channel CompletionHandler的調用狀況:網絡

從「Thread-2」線程堆棧中能夠發現，JDK底層經過線程池ThreadPoolExecutor來執行回調通知，異步回調通知類由sun.nio.ch.AsynchronousChannelGroupImpl實現，它通過層層調用，最終回調com.phei.netty.aio.AsyncTimeClientHandler$1.completed方法，完成回調通知。異步

由此咱們也能夠得出結論：異步SocketChannel是被動執行對象，咱們不須要像NIO編程那樣建立一個獨立的I/O線程來處理讀寫操做。對於AsynchronousServerSocketChannel和AsynchronousSocketChannel，它們都由JDK底層的線程池負責回調並驅動讀寫操做。async

正由於如此，基於NIO2.0新的異步非阻塞Channel進行編程比NIO編程更爲簡單。ide