(二)NIO之网络IO

概述和核心 API

前面在进行文件 IO 时用到的 FileChannel 并不支持非阻塞操作,学习 NIO 主要就是进行网络 IO,Java NIO 中的网络通道是非阻塞 IO 的实现,基于事件驱动,非常适用于服务器需要维持大量连接,但是数据交换量不大的情况,例如一些即时通信的服务等等
在 Java 中编写 Socket 服务器,通常有以下几种模式:

  • 一个客户端连接用一个线程,优点:程序编写简单;缺点:如果连接非常多,分配的线程也会非常多,服务器可能会因为资源耗尽而崩溃。
  • 把每一个客户端连接交给一个拥有固定数量线程的连接池,优点:程序编写相对简单,可以处理大量的连接。缺点:线程的开销非常大,连接如果非常多,排队现象会比较严重。
  • 使用 Java 的 NIO,用非阻塞的 IO 方式处理。这种模式可以用一个线程,处理大量的客
    户端连接。
  1. Selector(选择器),能够检测多个注册的通道上是否有事件发生,如果有事件发生,便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道,也就是管理多个连接。这样使得只有在连接真正有读写事件发生时,才会调用函数来进行读写,就大大地减少了系统开销,并且不必为每个连接都创建一个线程,不用去维护多个线程,并且避免了多线程之间的上下文切换导致的开销。

    该类的常用方法如下所示:
  • public static Selector open(),得到一个选择器对象
  • public int select(long timeout),监控所有注册的通道,当其中有 IO 操作可以进行时,将
    对应的 SelectionKey 加入到内部集合中并返回,参数用来设置超时时间
  • public Set selectedKeys(),从内部集合中得到所有的 SelectionKey
  1. SelectionKey,代表了 Selector 和网络通道的注册关系,一共四种:
  • int OP_ACCEPT:有新的网络连接可以 accept,值为 16
  • int OP_CONNECT:代表连接已经建立,值为 8
  • int OP_READ 和 int OP_WRITE:代表了读、写操作,值为 1 和 4
    该类的常用方法如下所示:
  • public abstract Selector selector(),得到与之关联的 Selector 对象
  • public abstract SelectableChannel channel(),得到与之关联的通道
  • public final Object attachment(),得到与之关联的共享数据
  • public abstract SelectionKey interestOps(int ops),设置或改变监听事件
  • public final boolean isAcceptable(),是否可以 accept
  • public final boolean isReadable(),是否可以读
  • public final boolean isWritable(),是否可以写
  1. ServerSocketChannel,用来在服务器端监听新的客户端 Socket 连接,常用方法如下所示:
  • public static ServerSocketChannel open(),得到一个 ServerSocketChannel 通道
  • public final ServerSocketChannel bind(SocketAddress local),设置服务器端端口号
    public final SelectableChannel configureBlocking(boolean block),设置阻塞或非阻塞模式,取值 false 表示采用非阻塞模式
  • public SocketChannel accept(),接受一个连接,返回代表这个连接的通道对象
  • public final SelectionKey register(Selector sel, int ops),注册一个选择器并设置监听事件
  1. SocketChannel,网络 IO 通道,具体负责进行读写操作。NIO 总是把缓冲区的数据写入通
    道,或者把通道里的数据读到缓冲区。常用方法如下所示:
  • public static SocketChannel open(),得到一个 SocketChannel 通道
  • public final SelectableChannel configureBlocking(boolean block),设置阻塞或非阻塞模式,取值 false 表示采用非阻塞模式
  • public boolean connect(SocketAddress remote),连接服务器
  • public boolean finishConnect(),如果上面的方法连接失败,接下来就要通过该方法完成
    连接操作
  • public int write(ByteBuffer src),往通道里写数据
  • public int read(ByteBuffer dst),从通道里读数据
  • public final SelectionKey register(Selector sel, int ops, Object att),注册一个选择器并设置
    监听事件,最后一个参数可以设置共享数据
  • public final void close(),关闭通道

实战演示

使用 NIO 开发一个入门案例,实现服务器端和客户端之间的数据通信(非阻塞):

客户端:

package com.bestqiang.nio.socket;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

/** * 网络客户端程序 * @author BestQiang */
public class NIOClient {
    public static void main(String[] args) throws Exception {
        // 文件操作是通过流得到通道,而网络IO操作是通过SocketChannel.open();
        // 1. 得到一个网络通道
        SocketChannel channel = SocketChannel.open();

        // 2. 得到一个阻塞方式
        channel.configureBlocking(false);

        // 3.提供服务器端的IP地址和端口号
        InetSocketAddress address = new InetSocketAddress("127.0.0.1",9999);

        // 4.连接服务器端
        if (!channel.connect(address)) {
            while (!channel.finishConnect()) { //nio作为非阻塞式的优势
                System.out.println("Client:连接服务器端的同时,我还可以干别的一些事情");
            }
        }

        // 5.得到一个缓冲区,并存入数据
        String msg = "Hello,Server!";
        ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());

        // 6.发送数据
        channel.write(buffer);

        System.in.read();
    }
}

服务端:

package com.bestqiang.nio.socket;

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

/** * 网络服务器端程序 * * @author BestQiang */
public class NIOServer {
    public static void main(String[] args) throws IOException {
        // 1.得到一个ServerSocketChannel对象
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

        // 2.得到一个Selector对象 间谍
        Selector selector = Selector.open();

        // 3.绑定一个端口号
        serverSocketChannel.bind(new InetSocketAddress(9999));

        // 4.设置非阻塞式
        serverSocketChannel.configureBlocking(false);

        // 5.把ServerSocket对象注册给selector对象
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        // 6.干活
        while (true) {
            // 6.1 监控客户端
            if (selector.select(2000) == 0) { //nio非阻塞式的优势
                System.out.println("Server:没有客户端搭理我,我就干点别的事");
                continue;
            }
            // 6.2 得到SelectionKey,判断通道里的事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            int i = 0;
            while (iterator.hasNext()) {
                i ++;
                SelectionKey key = iterator.next();
                if (key.isReadable()) { //读取客户端数据事件
                    SocketChannel socketChannel = (SocketChannel) key.channel();
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    socketChannel.read(buffer);
                    System.out.println("客户端发来数据:" + new String(buffer.array()));
                    System.out.println("attachment:" + i);
                }
                if (key.isAcceptable()) { //客户端连接事件
                    System.out.println("OP_ACCEPT");
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                    System.out.println("OP_ACCEPT:" + i);
                }


                // 6.3手动从集合中移除当前key,防止重复处理
                iterator.remove();
            }


        }
    }
}


也就是把ServerSocket注册到Selector,客户端连接服务器的时候取得SelectionKey进行识别,迭代遍历进行事件处理,由此可以完成通信,完成事件后,把SelectorKey移除即可。
网络聊天案例:
网络客户端程序:

package com.bestqiang.nio.socket;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

/** * 网络客户端程序 * @author BestQiang */
public class NIOClient {
    public static void main(String[] args) throws Exception {
        package com.bestqiang.nio.chat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;

//聊天程序客户端
public class ChatClient {
    private final String HOST = "127.0.0.1"; //服务器地址
    private int PORT = 9999; //服务器端口
    private SocketChannel socketChannel; //网络通道
    private String userName; //聊天用户名

    public ChatClient() throws IOException {
        //1. 得到一个网络通道
        socketChannel = SocketChannel.open();
        //2. 设置非阻塞方式
        socketChannel.configureBlocking(false);
        //3. 提供服务器端的IP地址和端口号
        InetSocketAddress address = new InetSocketAddress(HOST, PORT);
        //4. 连接服务器端
        if (!socketChannel.connect(address)) {
            while (!socketChannel.finishConnect()) {  //nio作为非阻塞式的优势
                System.out.println("Client:连接服务器端的同时,我还可以干别的一些事情");
            }
        }
        //5. 得到客户端IP地址和端口信息,作为聊天用户名使用
        userName = socketChannel.getLocalAddress().toString().substring(1);
        System.out.println("---------------Client(" + userName + ") is ready---------------");
    }

    //向服务器端发送数据

    public void sendMsg(String msg) throws Exception {
        if (msg.equalsIgnoreCase("bye")) {
            socketChannel.close();
            return;
        }
        msg = userName + "说:" + msg;
        ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
        socketChannel.write(buffer);
    }

    //从服务器端接收数据
    public void receiveMsg() throws Exception {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        int size = socketChannel.read(buffer);
        if (size > 0) {
            String msg = new String(buffer.array());
            System.out.println(msg.trim());
        }
    }

}


聊天程序服务器端:

package com.bestqiang.nio.chat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.text.SimpleDateFormat;
import java.util.*;

//聊天程序服务器端
public class ChatServer {
    private ServerSocketChannel listenerChannel; //监听通道 老大
    private Selector selector;//选择器对象 间谍
    private static final int PORT = 9999; //服务器端口

    public ChatServer() {
        try {
            // 1. 得到监听通道 老大
            listenerChannel = ServerSocketChannel.open();
            // 2. 得到选择器 间谍
            selector = Selector.open();
            // 3. 绑定端口
            listenerChannel.bind(new InetSocketAddress(PORT));
            // 4. 设置为非阻塞模式
            listenerChannel.configureBlocking(false);
            // 5. 将选择器绑定到监听通道并监听accept事件
            listenerChannel.register(selector, SelectionKey.OP_ACCEPT);
            printInfo("Chat Server is ready.......");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //6. 干活儿
    public void start() throws  Exception{
        try {
            while (true) { //不停监控
                if (selector.select(2000) == 0) {
                    System.out.println("Server:没有客户端找我, 我就干别的事情");
                    continue;
                }
                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    if (key.isAcceptable()) { //连接请求事件
                        SocketChannel sc=listenerChannel.accept();
                        sc.configureBlocking(false);
                        sc.register(selector,SelectionKey.OP_READ);
                        System.out.println(sc.getRemoteAddress().toString().substring(1)+"上线了...");
                    }
                    if (key.isReadable()) { //读取数据事件
                        readMsg(key);
                    }
                    //一定要把当前key删掉,防止重复处理
                    iterator.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //读取客户端发来的消息并广播出去
    public void readMsg(SelectionKey key) throws Exception{
        SocketChannel channel=(SocketChannel) key.channel();
        ByteBuffer buffer=ByteBuffer.allocate(1024);
        int count=channel.read(buffer);
        if(count>0){
            String msg=new String(buffer.array());
            printInfo(msg);
            //发广播
            broadCast(channel,msg);
        }
    }

    //给所有的客户端发广播
    public void broadCast(SocketChannel except,String msg) throws Exception{
        System.out.println("服务器发送了广播...");
        for(SelectionKey key:selector.keys()){
            Channel targetChannel=key.channel();
            if(targetChannel instanceof SocketChannel && targetChannel!=except){
                SocketChannel destChannel=(SocketChannel)targetChannel;
                ByteBuffer buffer=ByteBuffer.wrap(msg.getBytes());
                destChannel.write(buffer);
            }
        }
    }

    private void printInfo(String str) { //往控制台打印消息
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        System.out.println("[" + sdf.format(new Date()) + "] -> " + str);
    }

    public static void main(String[] args) throws Exception {
        new ChatServer().start();
    }
}

启动聊天程序客户端:

package com.bestqiang.nio.chat;

import java.util.Scanner;

//启动聊天程序客户端
public class TestChat {
    public static void main(String[] args) throws Exception {
        final ChatClient chatClient=new ChatClient();

        new Thread(){
            public void run(){
                while(true){
                    try {
                        chatClient.receiveMsg();
                        Thread.sleep(2000);
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            }
        }.start();

        Scanner scanner=new Scanner(System.in);
        while (scanner.hasNextLine()){
            String msg=scanner.nextLine();
            chatClient.sendMsg(msg);
        }

    }
}

全部评论

相关推荐

点赞 收藏 评论
分享
牛客网
牛客企业服务