NIO消息群发Demo

NIO

NIO-Snipaste_2019-11-30_19-05-41.png

代码实现

服务端代码启动服务，并处理连接与消息

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;

public class NioServer {
  public static void main(String[] args) {
    try {
      new NioServer().start();

    } catch (IOException e) {
      e.printStackTrace();
    }
  }

  public void start() throws IOException {
    /**
     * 创建一个selector
     */
    Selector selector = Selector.open();

    /**
     * 通过serversocketchannel 创建channel通道
     */
    ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

    /**
     * 为channel 通道绑定监听端口
     */
    serverSocketChannel.bind(new InetSocketAddress(8000));

    /**
     * 设置channel为非阻塞模式
     */
    serverSocketChannel.configureBlocking(false);

    /**
     * 将channel 注册到selector上，监听链接事件
     */

    serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    System.out.println("server started seccessful");
    /**
     * 循环等待新连接
     */

    while (true) {
      /**
       * TODO: 获取可用channel数量
       */
      int readyChannels = selector.select();

      /**
       * TODO: 为什么要这样？
       */
      if (readyChannels == 0)
        continue;

      /**
       * 获取可用channel的集合
       */
      Set<SelectionKey> selectionKeys = selector.selectedKeys();

      Iterator<SelectionKey> iterator = selectionKeys.iterator();

      while (iterator.hasNext()) {
        /**
         * selectionKey实例
         */

        SelectionKey selectionKey = (SelectionKey) iterator.next();

        /**
         * [!] 移除set中的当前的selectionKey
         */
        iterator.remove();

        /**
         * 根据就绪状态来判断相应的逻辑
         */

        /**
         * 如果是接入事件
         */
        if (selectionKey.isAcceptable()) {
          acceptHandler(serverSocketChannel, selector);
        }

        /**
         * 如果是可读事件
         */
        if (selectionKey.isReadable()) {
          readHandler(selectionKey, selector);
        }

      }
    }

    /**
     * 
     */
  }

  public void acceptHandler(ServerSocketChannel serverSocketChannel, Selector selector) throws IOException {
    /**
     * 如果是接入事件，创建socketchannel
     */
    SocketChannel socketChannel = serverSocketChannel.accept();

    /**
     * 将socketChannel设置为非阻塞工作模式
     */

    socketChannel.configureBlocking(false);
    /**
     * 将channel注册到selector上，监听可读事件
     */
    socketChannel.register(selector, SelectionKey.OP_READ);

    /**
     * 回复客户端提示信息
     */
    socketChannel.write(Charset.forName("UTF-8").encode("请注意隐私安全"));
  }

  public void readHandler(SelectionKey selectionKey, Selector selector) {
    /**
     * 要从selectionKey中获取到已经就绪的channel
     */
    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();

    /**
     * 创建buffer
     */
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    /**
     * 循环读取客户端请求信息
     */
    String request = "";
    try {

      while (socketChannel.read(byteBuffer) > 0) {
        /**
         * 切换buffer为读模式
         */
        byteBuffer.flip();

        /**
         * 读取buffer中的内容
         */

        request += Charset.forName("UTF-8").decode(byteBuffer);

      }
    } catch (IOException ioe) {
      ioe.printStackTrace();
      return;
    }

    /**
     * 将channel再次注册到selector上，监听它的可读事件
     */
    try {

      socketChannel.register(selector, SelectionKey.OP_READ);
    } catch (ClosedChannelException cce) {
      cce.printStackTrace();
      return;
    }

    /**
     * 将客户端发送的请求信息，广播给其他客户端
     */
    if (request.length() > 0) {
      // 广播给其他客户端
      System.out.println("::" + request);
      this.broadcast(selector, socketChannel, request);
    }

  }

  private void broadcast(Selector selector, SocketChannel socketChannel, String request) {
    /**
     * 获取所有已介入的客户端channel
     */
    Set<SelectionKey> selectionKeys = selector.keys();

    selectionKeys.forEach(selectionKey -> {
      Channel targetChannel = selectionKey.channel();

      // 剔除发消息的客户端
      if (targetChannel instanceof SocketChannel && targetChannel != socketChannel) {
        try {
          ((SocketChannel) targetChannel).write(Charset.forName("UTF-8").encode(request));

        } catch (IOException ioe) {
          ioe.printStackTrace();
        }
      }
    });

  }

}

客户端响应处理线程类

import java.nio.channels.Selector;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;
import java.util.Set;

public class ResponseHandler implements Runnable {
  private Selector selector;

  public ResponseHandler(Selector selector) {
    this.selector = selector;
  }

  @Override
  public void run() {
    try {
      /**
       * 循环等待新连接
       */

      while (true) {
        /**
         * TODO: 获取可用channel数量
         */
        int readyChannels = selector.select();

        /**
         * TODO: 为什么要这样？
         */
        if (readyChannels == 0)
          continue;

        /**
         * 获取可用channel的集合
         */
        Set<SelectionKey> selectionKeys = selector.selectedKeys();

        Iterator iterator = selectionKeys.iterator();

        while (iterator.hasNext()) {
          /**
           * selectionKey实例
           */

          SelectionKey selectionKey = (SelectionKey) iterator.next();

          /**
           * [!] 移除set中的当前的selectionKey
           */
          iterator.remove();

          /**
           * 根据就绪状态来判断相应的逻辑
           */


          /**
           * 如果是可读事件
           */
          if (selectionKey.isReadable()) {
            readHandler(selectionKey, selector);
          }
        }
      }
    } catch (Exception e) {
      // TODO: handle exception
    }
  }

  public void readHandler(SelectionKey selectionKey, Selector selector) {
    /**
     * 要从selectionKey中获取到已经就绪的channel
     */
    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();

    /**
     * 创建buffer
     */
    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);

    /**
     * 循环读取客户端请求信息
     */
    String response = "";
    try {

      while (socketChannel.read(byteBuffer) > 0) {
        /**
         * 切换buffer为读模式
         */
        byteBuffer.flip();

        /**
         * 读取buffer中的内容
         */

        response += Charset.forName("UTF-8").decode(byteBuffer);

      }
    } catch (IOException ioe) {
      ioe.printStackTrace();
      return;
    }

    /**
     * 将channel再次注册到selector上，监听它的可读事件
     */
    try {

      socketChannel.register(selector, SelectionKey.OP_READ);
    } catch (ClosedChannelException cce) {
      cce.printStackTrace();
      return;
    }

    /**
     * 将客户端发送的请求信息，广播给其他客户端
     */
    if (response.length() > 0) {
      // 广播给其他客户端
      System.out.println(response);
    }

  }
}

客户端，接收并展示消息


import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Scanner;

public class NioClient {
  public void start() throws IOException {
    /**
     * 连接服务器端
     */

    SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8000));
    /**
     * 向服务器端发送数据
     */
    Selector selector = Selector.open();
    socketChannel.configureBlocking(false);
    socketChannel.register(selector, SelectionKey.OP_READ);
    new Thread(new ResponseHandler(selector)).start();

    Scanner scanner = new Scanner(System.in);

    while (scanner.hasNextLine()) {
      String request = scanner.nextLine();

      if (request != null && request.length() > 0) {
        socketChannel.write(Charset.forName("UTF-8").encode(request));
      }

    }

    scanner.close();
    /**
     * 接受服务器响应
     */

  }

  public static void main(String[] args) {
    try {
      new NioClient().start();

    } catch (IOException e) {
      e.printStackTrace();
    }
  }

}

运行演示

消息群发demo演示

启动一个服务端，可以启动多个客户端连接到同一服务，实现消息群发。

©著作权归作者所有,转载或内容合作请联系作者
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。

推荐阅读更多精彩内容

  • Android面试准备
    一、简历准备 1、个人技能 (1)自定义控件、UI设计、常用动画特效 自定义控件 ①为什么要自定义控件？ Andr...
    lucas777阅读 5,266评论 2赞 54
  • Java NIO与IO详解
    概述 NIO主要有三大核心部分：Channel(通道)，Buffer(缓冲区),Selector。 传统IO基于...
    时之令阅读 3,750评论 0赞 8
  • IO/NIO & Netty
    IO编程 传统的 IO 编程，相信大家都不陌生，它能够实现客户端和服务端的通信，但是却是阻塞 IO。我们来通过 I...
    habit_learning阅读 4,360评论 0赞 18
  • 最难看透的是人心
    生活中总有那么些人，他们打着各种各样的幌子来跟你说他们对你有多好，你应该要感谢他们而不是做一些让他们不开心的事，说...
    什么鬼都一样阅读 593评论 4赞 1
  • 2019-02-27
    今天忙了一天，每次都提醒自己不要忘了写作这段时间！ 可是还是在不断地被其他事情耽误了，晚上在桂林鼓楼街收获了不少的...
    倾听_我是琪琪阅读 223评论 0赞 1