1. 程式人生 > >手動搭建I/O網路通訊框架4:AIO程式設計模型,聊天室終極改造

手動搭建I/O網路通訊框架4:AIO程式設計模型,聊天室終極改造

第一章:手動搭建I/O網路通訊框架1:Socket和ServerSocket入門實戰,實現單聊

第二章:手動搭建I/O網路通訊框架2:BIO程式設計模型實現群聊

第三章:手動搭建I/O網路通訊框架3:NIO程式設計模型,升級改造聊天室

  上一章講到的NIO程式設計模型比較主流,非常著名的Netty就是基於NIO程式設計模型的。這一章說的是AIO程式設計模型,是非同步非阻塞的。雖然同樣實現的是聊天室功能,但是實現邏輯上稍微要比NIO和BIO複雜一點。不過理好整體脈絡,會好理解一些。首先還是講講概念:

  BIO和NIO的區別是阻塞和非阻塞,而AIO代表的是非同步IO。在此之前只提到了阻塞和非阻塞,沒有提到非同步還是同步。可以用我在知乎上看到的一句話表示:【在處理 IO 的時候,阻塞和非阻塞都是同步 IO,只有使用了特殊的 API 才是非同步 IO】。這些“特殊的API”下面會講到。在說AIO之前,先總結一下阻塞非阻塞、非同步同步的概念。

  阻塞和非阻塞,描述的是結果的請求。阻塞:在得到結果之前就一直呆在那,啥也不幹,此時執行緒掛起,就如其名,執行緒被阻塞了。非阻塞:如果沒得到結果就返回,等一會再去請求,直到得到結果為止。非同步和同步,描述的是結果的發出,當呼叫方的請求進來。同步:在沒獲取到結果前就不返回給呼叫方,如果呼叫方是阻塞的,那麼呼叫方就會一直等著。如果呼叫方是非阻塞的,呼叫方就會先回去,等一會再來問問得到結果沒。非同步:呼叫方一來,如果是非阻塞的叫它先回去,待會有結果了再告訴你。如果是阻塞的,雖然非同步會通知他,但他還是要等著,實屬鐵憨憨。

  AIO中的非同步操作

  CompletionHandler

  在AIO程式設計模型中,常用的API,如connect、accept、read、write都是支援非同步操作的。當呼叫這些方法時,可以攜帶一個CompletionHandler引數,它會提供一些回撥函式。這些回撥函式包括:1.當這些操作成功時你需要怎麼做;2.如果這些操作失敗了你要這麼做。關於這個CompletionHandler引數,你只需要寫一個類實現CompletionHandler口,並實現裡面兩個方法就行了。

  那如何在呼叫connect、accept、read、write這四個方法時,傳入CompletionHandler引數從而實現非同步呢?下面分別舉例這四個方法的使用。

  先說說Socket和ServerSocket,在NIO中,它們變成了通道,配合緩衝區,從而實現了非阻塞。而在AIO中它們變成了非同步通道。也就是AsynchronousServerSocketChannel和AsynchronousSocketChannel,下面例子中物件名分別是serverSocket和socket.

  accept:serverSocket.accept(attachment,handler)。handler就是實現了CompletionHandler介面並實現兩個回撥函式的類,它具體怎麼寫可以看下面的實戰程式碼。attachment為handler裡面可能需要用到的輔助資料,如果沒有就填null。

  read:socket.read(buffer,attachment,handler)。buffer是緩衝區,用以存放讀取到的資訊。後面兩個引數和accept一樣。

  write:socket.write(buffer,attachment,handler)。和read引數一樣。

  connect:socket.connect(address,attachment,handler)。address為伺服器的IP和埠,後面兩個引數與前幾個一樣。

  Future

  既然說到了非同步操作,除了使用實現CompletionHandler介面的方式,不得不想到Future。客戶端邏輯較為簡單,如果使用CompletionHandler的話程式碼反而更復雜,所以下面的實戰客戶端程式碼就會使用Future的方式。簡單來說,Future表示的是非同步操作未來的結果,怎麼理解未來。比如,客戶端呼叫read方法獲取伺服器發來得訊息:

Future<Integer> readResult=clientChannel.read(buffer)

  Integer是read()的返回型別,此時變數readResult實際上並不一定有資料,而是表示read()方法未來的結果,這時候readResult有兩個方法,isDone():返回boolean,檢視程式是否完成處理,如果返回true,有結果了,這時候可以通過get()獲取結果。如果你不事先判斷isDone()直接呼叫get()也行,只不過它是阻塞的。如果你不想阻塞,想在這期間做點什麼,就用isDone()。

  還有一個問題:這些handler的方法是在哪個執行緒執行的?serverSocket.accept這個方法肯定是在主執行緒裡面呼叫的,而傳入的這些回撥方法其實是在其他執行緒執行的。在AIO中,會有一個AsynchronousChannelGroup,它和AsynchronousServerSocketChannel是繫結在一起的,它會為這些非同步通道提供系統資源,執行緒就算其中一種系統資源,所以為了方便理解,我們暫時可以把他看作一個執行緒池,它會為這些handler分配執行緒,而不是在主執行緒中去執行。

   AIO程式設計模型

  上面只說了些零碎的概念,為了更好的理解,下面講一講大概的工作流程(主要針對伺服器,客戶端邏輯較為簡單,程式碼註釋也比較少,可以看前面幾章):

  1.首先做準備工作。跟NIO一樣,先要建立好通道,只不過AIO是非同步通道。然後建立好AsyncChannelGroup,可以選擇自定義執行緒池。最後把AsyncServerSocket和AsyncChannelGroup繫結在一起,這樣處於同一個AsyncChannelGroup裡的通道就可以共享系統資源。

  2.最後一步準備工作,建立好handler類,並實現介面和裡面兩個回撥方法。(如圖:客戶端1對應的handler,裡面的回撥方法會實現讀取訊息和轉發訊息的功能;serverSocket的handler裡的回撥方法會實現accept功能。)

  3.準備工作完成,當客戶端1連線請求進來,客戶端會馬上回去,ServerSocket的非同步方法會在連線成功後把客戶端的SocketChannel存進線上使用者列表,並利用客戶端1的handler開始非同步監聽客戶端1傳送的訊息。

  4.當客戶端1傳送訊息時,如果上一步中的handler成功監聽到,就會回撥成功後的回撥方法,這個方法裡會把這個訊息轉發給其他客戶端。轉發完成後,接著利用handler監聽客戶端1傳送的訊息。

 

  程式碼一共有三個類:

  ChatServer:功能基本上和上面講的工作流程差不多,還會有一些工具方法,都比較簡單,就不多說了,如:轉發訊息,客戶端下線後從線上列表移除客戶端等。

  ChatClient:基本和前兩章的BIO、NIO沒什麼區別,一個執行緒監聽使用者輸入資訊併發送,主執行緒非同步的讀取伺服器資訊。

  UserInputHandler:監聽使用者輸入資訊的執行緒。

  ChatServer

public class ChatServer {
    //設定緩衝區位元組大小
    private static final int BUFFER = 1024;

    //宣告AsynchronousServerSocketChannel和AsynchronousChannelGroup
    private AsynchronousServerSocketChannel serverSocketChannel;
    private AsynchronousChannelGroup channelGroup;

    //線上使用者列表。為了併發下的執行緒安全,所以使用CopyOnWriteArrayList
    //CopyOnWriteArrayList在寫時加鎖,讀時不加鎖,而本專案正好在轉發訊息時需要頻繁讀取.
    //ClientHandler包含每個客戶端的通道,型別選擇為ClientHandler是為了在write的時候呼叫每個客戶端的handler
    private CopyOnWriteArrayList<ClientHandler> clientHandlerList;
    //字元和字串互轉需要用到,規定編碼方式,避免中文亂碼
    private Charset charset = Charset.forName("UTF-8");

    //通過建構函式設定監聽埠
    private int port;
    public ChatServer(int port) {
        this.port = port;
        clientHandlerList=new CopyOnWriteArrayList<>();
    }

    public void start() {
        try {
            /**
             *建立一個執行緒池並把執行緒池和AsynchronousChannelGroup繫結,前面提到了AsynchronousChannelGroup包括一些系統資源,而執行緒就是其中一種。
             *為了方便理解我們就暫且把它當作執行緒池,實際上並不止包含執行緒池。如果你需要自己選定執行緒池型別和數量,就可以如下操作
             *如果不需要自定義執行緒池型別和數量,可以不用寫下面兩行程式碼。
             * */
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            channelGroup = AsynchronousChannelGroup.withThreadPool(executorService);
            serverSocketChannel=AsynchronousServerSocketChannel.open(channelGroup);
            serverSocketChannel.bind(new InetSocketAddress("127.0.0.1",port));
            System.out.println("伺服器啟動:埠【"+port+"】");
            /**
             * AIO中accept可以非同步呼叫,就用上面說到的CompletionHandler方式
             * 第一個引數是輔助引數,回撥函式中可能會用上的,如果沒有就填null;第二個引數為CompletionHandler介面的實現
             * 這裡使用while和System.in.read()的原因:
             * while是為了讓伺服器保持執行狀態,前面的NIO,BIO都有用到while無限迴圈來保持伺服器執行,但是它們用的地方可能更好理解
             * System.in.read()是阻塞式的呼叫,只是單純的避免無限迴圈而讓accept頻繁被呼叫,無實際業務功能。
             */
            while (true) {
                serverSocketChannel.accept(null, new AcceptHandler());
                System.in.read();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }finally {
            if(serverSocketChannel!=null){
                try {
                    serverSocketChannel.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    //AsynchronousSocketChannel為accept返回的型別,Object為輔助引數型別,沒有就填Object
    private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel,Object>{
        //如果成功,執行的回撥方法
        @Override
        public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
            //如果伺服器沒關閉,在接收完當前客戶端的請求後,再次呼叫,以接著接收其他客戶端的請求
            if(serverSocketChannel.isOpen()){
                serverSocketChannel.accept(null,this);
            }
            //如果客戶端的channel沒有關閉
            if(clientChannel!=null&&clientChannel.isOpen()){
                //這個就是非同步read和write要用到的handler,並傳入當前客戶端的channel
                ClientHandler handler=new ClientHandler(clientChannel);
                //把新使用者新增到線上使用者列表裡
                clientHandlerList.add(handler);
                System.out.println(getPort(clientChannel)+"上線啦!");
                ByteBuffer buffer=ByteBuffer.allocate(BUFFER);
                //非同步呼叫read,第一個buffer是存放讀到資料的容器,第二個是輔助引數。
                //因為真正的處理是在handler裡的回撥函式進行的,輔助引數會直接傳進回撥函式,所以為了方便使用,buffer就當作輔助引數
                clientChannel.read(buffer,buffer,handler);
            }
        }
        //如果失敗,執行的回撥方法
        @Override
        public void failed(Throwable exc, Object attachment) {
            System.out.println("連線失敗"+exc);
        }
    }

    private class ClientHandler implements CompletionHandler<Integer, ByteBuffer>{
        private AsynchronousSocketChannel clientChannel;
        public ClientHandler(AsynchronousSocketChannel clientChannel) {
            this.clientChannel = clientChannel;
        }
        @Override
        public void completed(Integer result, ByteBuffer buffer) {
            if(buffer!=null){
                //如果read返回的結果小於等於0,而buffer不為空,說明客戶端通道出現異常,做下線操作
                if(result<=0){
                    removeClient(this);
                }else {
                    //轉換buffer讀寫模式並獲取訊息
                    buffer.flip();
                    String msg=String.valueOf(charset.decode(buffer));
                    //在伺服器上列印客戶端發來的訊息
                    System.out.println(getPort(clientChannel)+msg);
                    //把訊息轉發給其他客戶端
                    sendMessage(clientChannel,getPort(clientChannel)+msg);
                    buffer=ByteBuffer.allocate(BUFFER);

                    //如果使用者輸入的是退出,就從線上列表裡移除。否則接著監聽這個使用者傳送訊息
                    if(msg.equals("quit"))
                        removeClient(this);
                    else
                        clientChannel.read(buffer, buffer, this);
                }
            }
        }

        @Override
        public void failed(Throwable exc, ByteBuffer attachment) {
            System.out.println("客戶端讀寫異常:"+exc);
        }
    }

    //轉發訊息的方法
    private void sendMessage(AsynchronousSocketChannel clientChannel,String msg){
        for(ClientHandler handler:clientHandlerList){
            if(!handler.clientChannel.equals(clientChannel)){
                ByteBuffer buffer=charset.encode(msg);
                //write不需要buffer當輔助引數,因為寫到客戶端的通道就完事了,而讀還需要回調函式轉發給其他客戶端。
                handler.clientChannel.write(buffer,null,handler);
            }
        }
    }
    //根據客戶端channel獲取對應埠號的方法
    private String getPort(AsynchronousSocketChannel clientChannel){
        try {
            InetSocketAddress address=(InetSocketAddress)clientChannel.getRemoteAddress();
            return "客戶端["+address.getPort()+"]:";
        } catch (IOException e) {
            e.printStackTrace();
            return "客戶端[Undefined]:";
        }
    }
    //移除客戶端
    private void removeClient(ClientHandler handler){
        clientHandlerList.remove(handler);
        System.out.println(getPort(handler.clientChannel)+"斷開連線...");
        if(handler.clientChannel!=null){
            try {
                handler.clientChannel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        new ChatServer(8888).start();
    }
}

  ChatClient

public class ChatClient {
    private static final int BUFFER = 1024;
    private AsynchronousSocketChannel clientChannel;
    private Charset charset = Charset.forName("UTF-8");

    private String host;
    private int port;
    //設定伺服器IP和埠
    public ChatClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void start() {
        try {
            clientChannel = AsynchronousSocketChannel.open();
            //連線伺服器
            Future<Void> future = clientChannel.connect(new InetSocketAddress(host, port));
            future.get();
            //新建一個執行緒去等待使用者輸入
            new Thread(new UserInputHandler(this)).start();
            ByteBuffer buffer=ByteBuffer.allocate(BUFFER);
            //無限迴圈讓客戶端保持執行狀態
            while (true){
                //獲取伺服器發來的訊息並存入到buffer
                Future<Integer> read=clientChannel.read(buffer);
                if(read.get()>0){
                    buffer.flip();
                    String msg=String.valueOf(charset.decode(buffer));
                    System.out.println(msg);
                    buffer.clear();
                }else {
                    //如果read的結果小於等於0說明和伺服器連接出現異常
                    System.out.println("伺服器斷開連線");
                    if(clientChannel!=null){
                        clientChannel.close();
                    }
                    System.exit(-1);
                }
            }
        } catch (IOException | InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    public void send(String msg) {
        if (msg.isEmpty())
            return;
        ByteBuffer buffer = charset.encode(msg);
        Future<Integer> write=clientChannel.write(buffer);
        try {
            //獲取傳送結果,如果get方法發生異常說明發送失敗
            write.get();
        } catch (ExecutionException|InterruptedException e) {
            System.out.println("訊息傳送失敗");
            e.printStackTrace();
        }
    }

    public static void main(String[] args) {
        new ChatClient("127.0.0.1",8888).start();
    }
}

  UserInputHandler

public class UserInputHandler implements Runnable {
    ChatClient client;
    public UserInputHandler(ChatClient chatClient) {
        this.client=chatClient;
    }
    @Override
    public void run() {
        BufferedReader read=new BufferedReader(
                new InputStreamReader(System.in)
        );
        while (true){
            try {
                String input=read.readLine();
                client.send(input);
                if(input.equals("quit"))
                    break;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

 

  執行測試: