Neety学习----Java_NIO_知识点（二）

2022-11-22 14:36:06

Java_NIO_知识点二

一.Selector(选择器)
二. Selector 类相关方法
三. NIO 非阻塞网络编程原理分析图
四. 实现服务器端和客户端之间的数据简单通讯（非阻塞）
五. SelectionKey
六. ServerSocketChannel
七. SocketChannel
八. NIO 网络编程应用实例-群聊系统

一.Selector(选择器)

Java 的 NIO，用非阻塞的 IO 方式。可以用一个线程，处理多个的客户端连接，就会使用到 Selector(选择器)
Selector 能够检测多个注册的通道上是否有事件发生(注意:多个 Channel 以事件的方式可以注册到同一个Selector)，如果有事件发生，便获取事件然后针对每个事件进行相应的处理。这样就可以只用一个单线程去管理多个通道，也就是管理多个连接和请求
只有在连接/通道真正有读写事件发生时，才会进行读写，就大大地减少了系统开销，并且不必为每个连接都创建一个线程，不用去维护多个线程
避免了多线程之间的上下文切换导致的开销
Netty 的 IO 线程 NioEventLoop 聚合了 Selector(选择器，也叫多路复用器)，可以同时并发处理成百上千个客户端连接。
当线程从某客户端 Socket 通道进行读写数据时，若没有数据可用时，该线程可以进行其他任务。
线程通常将非阻塞 IO 的空闲时间用于在其他通道上执行 IO 操作，所以单独的线程可以管理多个输入和输出通道。
由于读写操作都是非阻塞的，这就可以充分提升 IO 线程的运行效率，避免由于频繁 I/O 阻塞导致的线程挂起。
一个 I/O 线程可以并发处理 N 个客户端连接和读写操作，这从根本上解决了传统同步阻塞 I/O 一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

二. Selector 类相关方法

Selector 类是一个抽象类, 常用方法和说明如下:

selector 相关方法说明
selector.select()//阻塞
selector.select(1000);//阻塞 1000 毫秒，在 1000 毫秒后返回
selector.wakeup();//唤醒 selector
selector.selectNow();//不阻塞，立马返还

三. NIO 非阻塞网络编程原理分析图

NIO 非阻塞网络编程相关的(Selector、SelectionKey、ServerScoketChannel 和 SocketChannel)

对上图的说明:

当客户端连接时，会通过 ServerSocketChannel 得到 SocketChannel
Selector 进行监听 select 方法, 返回有事件发生的通道的个数.
将 socketChannel 注册到 Selector 上, register(Selector sel, int ops), 一个 selector 上可以注册多个 SocketChannel
注册后返回一个 SelectionKey, 会和该 Selector 关联(集合)
进一步得到各个 SelectionKey (有事件发生)
在通过 SelectionKey 反向获取 SocketChannel , 方法 channel()
可以通过得到的 channel , 完成业务处理

四. 实现服务器端和客户端之间的数据简单通讯（非阻塞）

NIOClient_客户端



public class NIOClient_01 {
    public static void main(String[] args) throws IOException {
        //得到一个网络通道
        SocketChannel socketChannel = SocketChannel.open();
        //设置非阻塞模式
        socketChannel.configureBlocking(false);
        //提供服务器端IP 端口
        InetSocketAddress inetSocketAddress = new InetSocketAddress("127.0.0.1", 6666);
        //连接服务器
        if(!socketChannel.connect(inetSocketAddress)){
            while (!socketChannel.finishConnect()){
                System.out.println("连接需要时间，客户端可以做其他工作");
            }
        }
        //连接成功 发送数据
        String str="戏子66666";
        //包裹一个字节数组到buffer里面去
        ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
        //发送数据 将buffer写入channel
        socketChannel.write(buffer);
        System.in.read();
    }
}

NIOServer_服务器端


/*
    NIO非阻塞
 */
public class NIOServer_01 {
    public static void main(String[] args) throws IOException {
        //创建ServerSocketChannel-> ServerSocket
        ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();

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

        //绑定一个端口 在服务器端监听
        serverSocketChannel.socket().bind(new InetSocketAddress(6666));
        //设置位非阻塞
        serverSocketChannel.configureBlocking(false);

        //把 serverSocketChannel 注册到 selector 关心事件 OP_ACCEPT
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

        System.out.println("注册后的selectionkey 数量="+selector.keys().size());
        //等待客户端连接
        while(true){
            if(selector.select(1000)==0){ //没有时间发生
                System.out.println("服务器等待了一秒，没有连接");
                continue;
            }
            //获取到相关的 selectionkey集合
            //1. 返回大于0，表示已经获取到关注的事件
            //2. selector.selectedKeys() 返回关注事件的集合
            Set<SelectionKey> selectionKeys = selector.selectedKeys();
            System.out.println("selectionKeys 数量="+selectionKeys.size());
            //通过 selectionKeys  反向获取通道
            Iterator<SelectionKey> keyIterator = selectionKeys.iterator();
            while(keyIterator.hasNext()){
                //获取到selectionKey
                SelectionKey key = keyIterator.next();
                //根据key 对应的通道发生的事件做相应的处理
                if (key.isAcceptable()){ // 若是OP_ACCEPT 有新的客户端连接
                    //该客户端生成一个 socketChannel
                    SocketChannel socketChannel = serverSocketChannel.accept();
                    System.out.println("客户端连接成功 生成了一个新的socketChannel "+socketChannel.hashCode());
                    //将socketChannel设置为非阻塞
                    socketChannel.configureBlocking(false);
                    //将socketChannel 注册到selector  关注事件为OP_READ
                    //同时给将socketChannel关联一个buffer
                    socketChannel.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                    System.out.println("客户端连接后，注册的selectionkey 数量="+selector.keys().size());

                }
                if (key.isReadable()){ //发生 OP_READ
                    //通过key 反向获取对应的channel
                    SocketChannel channel = (SocketChannel) key.channel();
                    //获取到channel 关联的buffer
                    ByteBuffer buffer = (ByteBuffer) key.attachment();
                    channel.read(buffer);
                    System.out.println("客户端发送信息："+new String(buffer.array()));
                }

                //手动从集合中移除当前的selectionKey,防止重复操作
                keyIterator.remove();
            }
        }
    }
}

五. SelectionKey

SelectionKey，表示 Selector 和网络通道的注册关系, 共四种:

	int OP_ACCEPT：有新的网络连接可以 accept，值为 16
	int OP_CONNECT：代表连接已经建立，值为 8
	int OP_READ：代表读操作，值为 1
	int OP_WRITE：代表写操作，值为 4

源码中

public static final int OP_READ = 1 << 0;
public static final int OP_WRITE = 1 << 2;
public static final int OP_CONNECT = 1 << 3;
public static final int OP_ACCEPT = 1 << 4;

SelectionKey 相关方法

六. ServerSocketChannel

NIO 中的 ServerSocketChannel 功能类似 ServerSocket，SocketChannel 功能类似 Socket
ServerSocketChannel 在服务器端监听新的客户端 Socket 连接

相关方法如下

七. SocketChannel

SocketChannel，网络 IO 通道，具体负责进行读写操作。NIO 把缓冲区的数据写入通道，或者把通道里的数据读到缓冲区。

相关方法如下

八. NIO 网络编程应用实例-群聊系统

实例要求:

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

GroupChatServer _服务端

package com.xizi.nio_chat;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Set;

public class GroupChatServer {
    //定义属性
    private Selector selector;
    private ServerSocketChannel listenChannel;
    private static final int  PORT=6667;


    public GroupChatServer() {
        try {
            //得到选择器
            selector=Selector.open();
            //得到ServerSocketChannel
            listenChannel = ServerSocketChannel.open();
            //绑定端口
            listenChannel.socket().bind(new InetSocketAddress(PORT));
            //将listenChannel设置为非阻塞模式
            listenChannel.configureBlocking(false);
            //将将listenChannel 注册到selector
            listenChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //监听
    public void listen(){
        while (true){
            try {
                //使用选择器监听
                int count = selector.select();
                if(count>0){ //有事件处理
                    //遍历得到的selectionKey 集合
                    Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                    while(keyIterator.hasNext()){{
                        //取出selectionkey
                        SelectionKey key = keyIterator.next();
                        //监听到accept事件
                        if(key.isAcceptable()){
                            SocketChannel socketChannel = listenChannel.accept();
                            //设置SocketChannel为非阻塞
                            socketChannel.configureBlocking(false);
                            //将socketChannel注册到selector
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            //提示
                            System.out.println(socketChannel.getRemoteAddress()+" 上线");
                        }
                        //监听到read事件 通道可读
                        if (key.isReadable()){
                            //处理读
                            readData(key);
                        }

                        //当前的key 删除 防止重复处理
                        //手动从集合中移除当前的selectionKey,防止重复操作
                        keyIterator.remove();
                    }}

                }else {
                    System.out.println("等待.............");
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {

            }
        }
    }

    //读取客户端消息
    public  void  readData(SelectionKey key){
        //定义一个SocketChannel
        SocketChannel channel=null;
        try {
             channel =(SocketChannel)key.channel();
             //创建buffer
            ByteBuffer buffer = ByteBuffer.allocate(1024);
            //将通道的信息读取到缓冲区
            int count = channel.read(buffer);
            if(count>0){
                //把缓冲区数据转成字符串
                String msg = new String(buffer.array());
                System.out.println("客户端发送的信息是："+msg);
                //向其他客户端转发消息(去掉自己)
                sendInfoToOTHERClient(msg, channel);
            }
        } catch (Exception e) {
            try {
                System.out.println(channel.getRemoteAddress()+" 离线了");
                //取消注册
                key.cancel();
                //关闭通道
                channel.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
        }
    }

    //向其他客户端转发消息（给通道发）
    public void sendInfoToOTHERClient(String msg,SocketChannel self) throws IOException {
        System.out.println("服务器转发消息中.......");
        //遍历 所有注册到selector 上的SocketChannel 排除本身
        for (SelectionKey key:selector.keys()) {
            //通过key 取出对应的SocketChannel
            Channel targetChannel = key.channel();
            //排除自己
            if (targetChannel instanceof SocketChannel && targetChannel!=self){
                SocketChannel dest=(SocketChannel)targetChannel;
                //将msg存储到buffer
                ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
                dest.write(buffer);
            }

        }
    }

    public static void main(String[] args) {
        //创建一个服务器对象
        GroupChatServer groupChatServer = new GroupChatServer();
        groupChatServer.listen();
    }

}

GroupChatClient _客户端

package com.xizi.nio_chat;

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

public class GroupChatClient {

    //定义相关的属性
    private final String HOST="127.0.0.1";
    private final  int PORT=6667;
    private Selector selector;
    private SocketChannel socketChannel;
    private  String username;

    //构造器
    public GroupChatClient() throws IOException {
        selector=Selector.open();
        //连接服务器
        socketChannel=SocketChannel.open(new InetSocketAddress(HOST,PORT));
        //设置非阻塞
        socketChannel.configureBlocking(false);
        //将channel 注册到selector
        socketChannel.register(selector, SelectionKey.OP_READ);
        //得到username
         username = socketChannel.getLocalAddress().toString().substring(1);
        System.out.println(username+" 客户端启动成功.....");

    }

    //向服务器发送消息
    public void sendInfo(String info){
        info=username+"说： "+info;
        try {
            socketChannel.write(ByteBuffer.wrap(info.getBytes()));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    //读取从服务器端回复的消息
    public  void readInfo(){
        try {
            int readChannel = selector.select();
            if(readChannel>0){
                //有可以用的通道
                Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
                while(keyIterator.hasNext()){
                    SelectionKey key = keyIterator.next();
                    if(key.isReadable()){
                        //得到相关的通道
                        SocketChannel socketChannel = (SocketChannel) key.channel();
                        //设置SocketChannel为非阻塞
                        socketChannel.configureBlocking(false);
                        //读取数据到缓冲区
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
                        socketChannel.read(buffer);
                        //把缓冲区的数据转成字符串
                        String msg = new String(buffer.array());
                        System.out.println("服务器回复的数据： "+msg.trim());
                    }
                }
                //删除当前的selectionKey 防止重复操作
                keyIterator.remove();
            }else{
//                System.out.println("没有可用的通道...");
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void main(String[] args) throws IOException {
        //q启动客户端
        GroupChatClient groupChatClient = new GroupChatClient();
        //启动一个线程 每隔3秒 读取服务器发送的信息
        new Thread(){
            @Override
            public void run() {
                while(true){
                    groupChatClient.readInfo();
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }.start();


        //客户端发送数据给服务端
        Scanner scanner = new Scanner(System.in);
        System.out.println("请发送信息: ");
        while(scanner.hasNextLine()){
            System.out.println("请发送信息: ");
            String s = scanner.nextLine();
            groupChatClient.sendInfo(s);
        }
    }
}

码农公寓

Java_NIO_知识点二

一.Selector(选择器)

二. Selector 类相关方法

三. NIO 非阻塞 网络编程原理分析图

四. 实现服务器端和客户端之间的数据简单通讯（非阻塞）

五. SelectionKey

六. ServerSocketChannel

七. SocketChannel

八. NIO 网络编程应用实例-群聊系统

相关文章

三. NIO 非阻塞网络编程原理分析图