[心得] 從0開始 3.8 非阻塞式IO的聊天室

看板mud (網路地下城/文字遊戲)作者gasbomb (虛空雷神獸)時間5年前 (2019/12/19 17:55)推噓3(3推 0噓 3→)

留言6則, 4人參與討論串1/2 (看更多)

之前實作的聊天室由於使用了阻塞式的 IO 在等待使用者輸入指令時整個執行緒都必須暫停所以說線上有幾個使用者就等於我們要同時開啟幾條執行緒這是非常浪費資源的在後來的 java 版本有提供了非阻塞式的 IO 讓我們可以只用一條執行緒就可以應付許多連線這次就使用 AsynchronousServerSocketChannel 來實作聊天室 (簡稱 AIO) 以下就是聊天室的程式碼由於 AIO 有非常多的細節, 但是我們的目的是要開發 MUD 因此這邊我不打算解釋的太詳細 // GeneralAioEchoServer.java // ✂--------------請沿虛線剪下-------------- package test; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.AsynchronousChannelGroup; import java.nio.channels.AsynchronousServerSocketChannel; import java.nio.channels.AsynchronousSocketChannel; import java.nio.channels.CompletionHandler; import java.nio.charset.StandardCharsets; import java.util.HashSet; import java.util.LinkedList; import java.util.Queue; import java.util.Set; import java.util.concurrent.*; public class GeneralAioEchoServer { private AsynchronousServerSocketChannel assc; private Set<AsynchronousSocketChannel> users = new HashSet<>(); public static void main(String[] args) throws Exception { GeneralAioEchoServer server = new GeneralAioEchoServer(); server.start(); // AIO 因為不會阻塞, 所以必須要有無限迴圈來維持 main thread while (true) { Thread.sleep(5000L); } } // 建立連線池, 設定 server port, 啟動 private void start() throws IOException { ExecutorService pool = Executors.newSingleThreadExecutor(); AsynchronousChannelGroup channelGroup = AsynchronousChannelGroup.withThreadPool(pool); assc = AsynchronousServerSocketChannel.open(channelGroup); assc.bind(new InetSocketAddress(4000)); // 設定 callback method assc.accept(null, new AcceptHandler()); } private class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> { @Override public void completed(AsynchronousSocketChannel asc, Object o) { assc.accept(null, this); try { asc.write(StandardCharsets.UTF_8.encode( "歡迎來到 aio telnet chat server\r\n")).get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } ByteBuffer bb = ByteBuffer.allocate(1024); asc.read(bb, null, new ReadHandler(asc, bb)); users.add(asc); } @Override public void failed(Throwable throwable, Object o) { } } private class ReadHandler implements CompletionHandler<Integer, Object> { private AsynchronousSocketChannel asc; private ByteBuffer bb; private MyByteArrayOutputStream byteArrayOutputStream = new MyByteArrayOutputStream(); private boolean firstChar = true; private Queue<String> inputs = new LinkedList<>(); public ReadHandler(AsynchronousSocketChannel asc, ByteBuffer bb) { this.asc = asc; this.bb = bb; } @Override public void completed(Integer result, Object o) { if (result == -1) return; // 逐 byte 讀取玩家輸入的字元 byte[] bytes = new byte[result]; bb.flip().get(bytes).clear(); for (byte b : bytes) { switch (b) { case '\n': if (firstChar) { firstChar = false; continue; } case '\r': inputs.offer( new String(byteArrayOutputStream.toByteArray(), StandardCharsets.UTF_8)); byteArrayOutputStream.reset(); firstChar = true; continue; case 127: byteArrayOutputStream.backspace(); continue; default: byteArrayOutputStream.write(b); firstChar = false; } } try { while (!inputs.isEmpty()) { String message = inputs.poll() + "\r\n"; for (AsynchronousSocketChannel user : users) { user.write(StandardCharsets.UTF_8.encode(message)).get(); } } } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } asc.read(bb, null, this); } @Override public void failed(Throwable throwable, Object o) { } } // 繼承ByteArrayOutputStream 實作 backspace 的功能 private static class MyByteArrayOutputStream extends ByteArrayOutputStream { public void backspace() { if (count > 0) count--; } } } // ✂--------------請沿虛線剪下-------------- 如此一來, 更輕量化的聊天室就完成了下一次我們會開始實作登入系統 -- ╔═◢ ◣═╦╦═════╦═════╗ ║ ◤◤◤ ◥ ╠╣飛鳥ももこ╠═╗ ║ ║ ▇ ▇ ║╚═════╝ ╚═╦═╣ ║ ▌ ● ● ▌ ║╔══════╗╔═╩═╣ ║◤ ◥ ︺█◤ ◥╠╣Momoko Asuka╠╝ ║ ╚◣◢ ▄▂▄ ◣◢╩╩══════╩════╝ -- ※ 發信站: 批踢踢實業坊(ptt.cc), 來自: 211.72.253.48 (臺灣) ※ 文章網址: https://www.ptt.cc/bbs/mud/M.1576749320.A.36F.html