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演示

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

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 218,204评论 6赞 506
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 93,091评论 3赞 395
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 164,548评论 0赞 354
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 58,657评论 1赞 293
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 67,689评论 6赞 392
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 51,554评论 1赞 305
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 40,302评论 3赞 418
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 39,216评论 0赞 276
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 45,661评论 1赞 314
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 37,851评论 3赞 336
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 39,977评论 1赞 348
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 35,697评论 5赞 347
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 41,306评论 3赞 330
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 31,898评论 0赞 22
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 33,019评论 1赞 270
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 48,138评论 3赞 370
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 44,927评论 2赞 355

NIO消息群发Demo

NIO

代码实现

推荐阅读更多精彩内容