java nio之selector

Posted 2020-10-16 小不点丶

　　一、selector简介：选择器提供选择执行已经就绪的任务的能力.从底层来看，Selector提供了询问通道是否已经准备好执行每个I/O操作的能力。Selector 允许一个单一的线程来操作多个 Channel。仅用单个线程来处理多个Channels的好处是，只需要更少的线程来处理通道。事实上，可以只用一个线程处理所有的通道，这样会大量的减少线程之间上下文切换的开销。

　　二、选择器的创建以及使用

　　1）创建　

Selector selector = Selector.open();

　　2）注册选择器（Channel这里不介绍）

socketChannel.configureBlocking(false);

socketChannel.register(selector, SelectionKey.OP_READ)

　　注意：一个通道注册到选择器中，必须是非阻塞的。

　　3）注册模式有4种

SelectionKey.OP_CONNECT 
SelectionKey.OP_ACCEPT 
SelectionKey.OP_READ 
SelectionKey.OP_WRITE

　　4）SelectionKey的使用

　　在选择其中会存在多个选择键SelectionKey，每一个选择键的类型可能不一样，所以我们这里需要判定是哪一种类型

selector.selectedKeys() //获取所有选择键
selectionKey.isConnectable() //是否是连接选择键
selectionKey.isReadable() //读取
selectionKey.isWritable() //写入
selectionKey.isAcceptable() //接收

　　获取对应的选择键过后可以强转成对应的通信管道。（示例）

SocketChannel channel = (SocketChannel) selectionKey.channel();

　　三、聊天室的基本写法（基本使用都在里面）

　　1）客户端

package com.troy.nio.application;

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

public class Client {

    //选择器
    private static Selector selector;
    //通信管道
    private static SocketChannel socketChannel;
    public static void main(String[] args) {
        try {
            clientInit();
            listen();
            //发送数据
            while (true) {
                Thread.sleep(1000);
                socketChannel.write(ByteBuffer.wrap(("hello server!").getBytes()));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //初始化选择器和发送数据
    private static void clientInit() throws Exception {
        //打开一个通道管理器
        selector = Selector.open();
        //获取一个通信管道
        socketChannel = SocketChannel.open();
        //设置对应的发送地址和端口
        socketChannel.connect(new InetSocketAddress("localhost",9000));
        //设置非阻塞
        socketChannel.configureBlocking(false);
        //注册一个写入事件
        socketChannel.register(selector, SelectionKey.OP_READ);
    }

    //监听服务器返回的数据
    private static void listen() throws Exception {
        Runnable runnable = () -> {
            while (true) {
                try {
                    //这里会一直阻塞，直到事件过来
                    selector.select();
                    //在选择器中获取对应的注册事件
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                    while (iterator.hasNext()) {
                        //注册事件
                        SelectionKey selectionKey = iterator.next();
                        iterator.remove();
                        //判断是否是读事件
                        if (selectionKey.isReadable()) {
                            //获取对应通信管道，并处理层数据
                            SocketChannel channel = (SocketChannel) selectionKey.channel();
                            //一次性读取数据量，这里应该做循环，我这里方便没有做
                            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                            channel.read(byteBuffer);
                            byteBuffer.flip();
                            System.out.println(new String(byteBuffer.array()).trim());
                        }
                    }
                } catch (Exception e) {
                    throw new RuntimeException(e.getMessage());
                }
            }
        };
        new Thread(runnable).start();
    }
}

　　2）服务端

package com.troy.nio.application;


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;

public class Server {

    //选择器
    private static Selector selector;
    //服务端通信管道
    private static ServerSocketChannel channel;
    public static void main(String[] args) {
        try {
            serverInit();
            listen();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    //初始化
    private static void serverInit() throws IOException {
        //打开一个选择器
        selector = Selector.open();
        //打开一个服务端通信管道
        channel = ServerSocketChannel.open();
        //设置接收端口
        channel.socket().bind(new InetSocketAddress(9000));
        //设置非阻塞
        channel.configureBlocking(false);
        //注册接收事件
        channel.register(selector, SelectionKey.OP_ACCEPT);
    }

    //监听
    private static void listen() throws IOException {
        while (true) {
            //形成阻塞事件，接口完成后进行下一步
            selector.select();
            //获取选择器中的事件
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey selectionKey = iterator.next();
                iterator.remove();
                //判断是否是接受事件
                if (selectionKey.isAcceptable()) {
                    SocketChannel socketChannel = channel.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector,SelectionKey.OP_READ);
                }
                //是否是可读事件
                if (selectionKey.isReadable()) {
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                    //这里的目的是当这个服务端一直存在，因为读取数据存在异常，直接处理掉，下一个客户端景来可以继续接受
                    try {
                        socketChannel.read(byteBuffer);
                    } catch (Exception e) {
                        continue;
                    }
                    byteBuffer.flip();
                    System.out.println(new String(byteBuffer.array()).trim());
                    socketChannel.write(ByteBuffer.wrap("hello client!".getBytes()));
                }
            }
        }
    }
}