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

[gasbomb (虛空雷神獸)]

之前實作的聊天室由於使用了阻塞式的 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
→ laechan     : 所以實際上底層還是走線程池
→ laechan     : 以AsynchronousServerSocketChannel
→ laechan     : 來包裹它的一些應用
推 outshaker   : 推 期待更新
推 nfsong      : 推
推 jameslong   : 等候更新