一、案例1

要求

  1. 编写一个 NIO 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
  2. 实现多人群聊
  3. 服务器端：可以监测用户上线，离线，并实现消息转发功能
  4. 客户端：通过 Channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息（有服务器转发得到）

  5. 目的：进一步理解 NIO 非阻塞网络编程机制

    1、服务端

    1. public class GroupChatServer {
    2.  // 定义属性
    3.  private Selector selector;
    4.  private ServerSocketChannel listenChannel;
    5.  private static final int PORT = 6667;
    7.  // 构造器
    8.  // 初始化工作
    9.  public GroupChatServer() {
    10.      try {
    11.          // 得到选择器
    12.          selector = Selector.open();
    13.          // 获取 ServerSocketChannel
    14.          listenChannel = ServerSocketChannel.open();
    15.          // 绑定端口
    16.          listenChannel.socket().bind(new InetSocketAddress(PORT));
    17.          // 配置非阻塞模式
    18.          listenChannel.configureBlocking(false);
    19.          // 将该 listenChannel 注册到 selector
    20.          listenChannel.register(selector, SelectionKey.OP_ACCEPT);
    22.      } catch (IOException e) {
    23.          e.printStackTrace();
    24.      }
    25.  }
    27.  // 监听
    28.  public void listen() {
    29.      try {
    30.          while (true) {
    31. //                int count = selector.select(2000);
    32.              int count = selector.select();
    33.              if (count > 0) { // 有事件处理
    34.                  // 遍历得到的 SelectionKey 集合
    35.                  Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
    36.                  while (iterator.hasNext()) {
    37.                      // 取出 selectionKey
    38.                      SelectionKey key = iterator.next();
    40.                      // 监听到 accept
    41.                      if (key.isAcceptable()) {
    42.                          SocketChannel socketChannel = listenChannel.accept();
    43.                          socketChannel.configureBlocking(false);
    44.                          // 将该 socketChannel 注册到 Selector
    45.                          socketChannel.register(selector, SelectionKey.OP_READ);
    46.                          // 提示
    47.                          System.out.println(socketChannel.getRemoteAddress() + " 上线 ");
    48.                      }
    49.                      if (key.isReadable()) { //通道发送read事件，即通道可读
    50.                          // 处理读
    51.                          readData(key);
    52.                      }
    54.                      // 当前的key删除，防止重复处理
    55.                      iterator.remove();
    56.                  }
    57.              } else {
    58.                  System.out.println("等待。。。。。");
    59.              }
    60.          }
    61.      } catch (Exception e) {
    62.          e.printStackTrace();
    63.      } finally {
    65.      }
    66.  }
    68.  // 读取客户端信息
    69.  private void readData(SelectionKey key) {
    71.      SocketChannel channel = null;
    72.      try {
    73.          // 取到关联的 channel
    74.          channel = (SocketChannel) key.channel();
    75.          // 创建 buffer
    76.          ByteBuffer buffer = ByteBuffer.allocate(1024);
    77.          int count = channel.read(buffer);
    78.          // 根据count做处理
    79.          if (count > 0) {
    80.              String msg = new String(buffer.array());
    81.              System.out.println("from 客户端：" + msg);
    83.              // 向其他客户端转发消息(排除自己)
    84.              sendInfoToOtherClients(msg, channel);
    85.          }
    86.      } catch (Exception e) {
    87.          try {
    88.              System.out.println(channel.getRemoteAddress() + "离线");
    89.              // 取消注册
    90.              key.cancel();
    91.              // 关闭通道
    92.              channel.close();
    93.          } catch (IOException ioException) {
    94.              ioException.printStackTrace();
    95.          }
    96.      }
    97.  }
    99.  // 转发消息给其他客户
    100.  private void sendInfoToOtherClients(String msg, SocketChannel self) throws IOException {
    101.      System.out.println("服务器转发消息中...");
    102.      // 遍历 所有注册到 selector 上的 SocketChannel，并删除 self
    103.      for (SelectionKey key : selector.keys()) {
    104.          Channel targetChannel = key.channel();
    106.          // 排除自己
    107.          if (targetChannel instanceof SocketChannel && targetChannel != self) {
    108.              // 转型
    109.              SocketChannel dest = (SocketChannel) targetChannel;
    110.              // 将 msg 存储到buffer
    111.              ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes(StandardCharsets.UTF_8));
    112.              // 将 buffer 数据写入到 通道
    113.              System.out.println("转发给" + dest.getRemoteAddress().toString().substring(1) + ",内容：【" + msg + "]");
    114.              dest.write(buffer);
    115.          }
    116.      }
    117.  }
    119.  public static void main(String[] args) {
    120.      GroupChatServer groupChatServer = new GroupChatServer();
    121.      groupChatServer.listen();
    122.  }
    123. }

    2、客户端

    1. public class GroupChatClient {
    2.  private final static String HOST = "127.0.0.1";
    3.  private final static int PORT = 6667;
    4.  private Selector selector;
    5.  private SocketChannel socketChannel;
    6.  private String username;
    8.  // 构造器，完成初始化工作
    9.  public GroupChatClient() throws IOException {
    10.      selector = Selector.open();
    11.      // 连接服务器
    12.      socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
    13.      // 设置非阻塞
    14.      socketChannel.configureBlocking(false);
    15.      socketChannel.register(selector, SelectionKey.OP_READ);
    16.      // 得到 username
    17. //        username = socketChannel.getLocalAddress().toString().substring(1);
    18.      username = socketChannel.getLocalAddress().toString().substring(1);
    19.      System.out.println(username + " is ok...");
    20.  }
    22.  // 向服务器发送消息
    23.  public void sendInfo(String info) {
    24.      info = username + "说：" + info;
    25.      try {
    26.          socketChannel.write(ByteBuffer.wrap(info.getBytes(StandardCharsets.UTF_8)));
    27.      } catch (IOException e) {
    28.          e.printStackTrace();
    29.      }
    30.  }
    32.  // 读取从服务器端回复的消息
    33.  public void readInfo() {
    34.      try {
    35. //            int readChannels = selector.select(2000);
    36.          int readChannels = selector.select();
    38.          if (readChannels > 0) {
    39.              Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
    40.              while (iterator.hasNext()) {
    41.                  SelectionKey key = iterator.next();
    42.                  if (key.isReadable()) {
    43.                      // 得到相关的通道
    44.                      SocketChannel socketChannel = (SocketChannel) key.channel();
    45.                      // 得到一个buffer
    46.                      ByteBuffer buffer = ByteBuffer.allocate(1024);
    47.                      // 读取
    48.                      socketChannel.read(buffer);
    49.                      String msg = new String(buffer.array());
    50.                      System.out.println(msg);
    51.                  }
    52.              }
    53.              // 删除当前 SelectionKey 防止重复操作
    54.              iterator.remove();
    55.          } else {
    56. //                System.out.println("没有可用的通道");
    57.          }
    58.      } catch (IOException e) {
    59.          e.printStackTrace();
    60.      }
    61.  }
    63.  public static void main(String[] args) throws IOException {
    64.      // 启动客户端
    65.      GroupChatClient groupChatClient = new GroupChatClient();
    67.      // 启动一个线程,每隔三秒，从服务器端读取数据
    68.      new Thread(new Runnable() {
    69.          @Override
    70.          public void run() {
    71.              while (true) {
    72.                  groupChatClient.readInfo();
    73.                  try {
    74.                      Thread.sleep(3000);
    75.                  } catch (InterruptedException e) {
    76.                      e.printStackTrace();
    77.                  }
    78.              }
    79.          }
    80.      }).start();
    82.      // 发送数据给服务器
    83.      Scanner scanner = new Scanner(System.in);
    84.      while (scanner.hasNextLine()) {
    85.          String s = scanner.nextLine();
    86.          groupChatClient.sendInfo(s);
    87.      }
    88.  }
    89. }