9.NIO非阻塞式网络通信入门

开发运维 2023-07-24 向阳逐梦手机阅读

Selector 示意图和特点说明

一个 I/O 线程可以并发处理 N 个客户端连接和读写操作，这从根本上解决了传统同步阻塞 I/O 一连接一线程模型。架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

服务端流程

1、当客户端连接服务端时，服务端会通过 ServerSocketChannel 得到 SocketChannel:获取通道
```
 ServerSocketChannel ssChannel = ServerSocketChannel.open();
```
2、切换非阻塞模式
```
 ssChannel.configureBlocking(false);
```

3、绑定连接

 ssChannel.bind(new InetSocketAddress(9999));

4、获取选择器
```
Selector selector = Selector.open();
```
5、将通道注册到选择器上, 并且指定“监听接收事件”
```
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
```
轮询式的获取选择器上已经“准备就绪”的事件

 //轮询式的获取选择器上已经“准备就绪”的事件,大于0 说明存在 准备就绪的事件
 while (selector.select() > 0) {
        System.out.println("轮一轮");
        //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
        Iterator it = selector.selectedKeys().iterator();
        while (it.hasNext()) {
            //8. 获取准备“就绪”的是事件
            SelectionKey sk = it.next();
            //9. 判断具体是什么事件准备就绪
            if (sk.isAcceptable()) {
                //10. 若“接收就绪”，获取客户端连接
                SocketChannel sChannel = ssChannel.accept();
                //11. 切换非阻塞模式
                sChannel.configureBlocking(false);
                //12. 将该通道注册到选择器上并修改注册事件为read
                sChannel.register(selector, SelectionKey.OP_READ);
            } else if (sk.isReadable()) {
                //13. 获取当前选择器上“读就绪”状态的通道
                SocketChannel sChannel = (SocketChannel) sk.channel();
                //14. 读取数据
                ByteBuffer buf = ByteBuffer.allocate(1024);
                int len = 0;
                while ((len = sChannel.read(buf)) > 0) {
                    buf.flip();
                    System.out.println(new String(buf.array(), 0, len));
                    buf.clear();
                }
            }
            //15. 处理完毕 移除选择键 SelectionKey
            it.remove();
        }
    }
}

客户端流程

1.获取通道

SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9999));

2.切换非阻塞模式

sChannel.configureBlocking(false);

3.分配指定大小的缓冲区

ByteBuffer buf = ByteBuffer.allocate(1024);

4.发送数据给服务端

 Scanner scan = new Scanner(System.in);
while(scan.hasNext()){
    String message = scan.nextLine();
    buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
            + "n" + message).getBytes());
    buf.flip();
    sChannel.write(buf);
    buf.clear();
}
//关闭通道
sChannel.close();

NIO非阻塞式网络通信入门案例

需求：服务端接收客户端的连接请求，并接收多个客户端发送过来的事件。

代码案例

 /**
  客户端
 */
public class Client {
    public static void main(String[] args) throws Exception {
        //1. 获取通道 - SelectionKey.OP_ACCEPT 对应监听接收事件
        SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9999));
        //2. 切换非阻塞模式
        sChannel.configureBlocking(false);
        //3. 分配指定大小的缓冲区
        ByteBuffer buf = ByteBuffer.allocate(1024);
        //4. 发送数据给服务端
        Scanner scan = new Scanner(System.in);
        
        while(scan.hasNext()){
            String message = scan.nextLine();
            buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
                    + "n" + message).getBytes());
            buf.flip();
            //客户端写对应服务器端读就绪
            sChannel.write(buf);
            buf.clear();
        }
        //5. 关闭通道
        sChannel.close();
    }
}
/**
 服务端
 */
public class Server {
    public static void main(String[] args) throws IOException {
        //1. 获取通道
        ServerSocketChannel ssChannel = ServerSocketChannel.open();
        //2. 切换非阻塞模式
        ssChannel.configureBlocking(false);
        //3. 绑定连接
        ssChannel.bind(new InetSocketAddress(9999));
        //4. 获取选择器
        Selector selector = Selector.open();
        //5. 将通道注册到选择器上, 并且指定“监听接收事件”
        ssChannel.register(selector, SelectionKey.OP_ACCEPT);
        //6. 轮询式的获取选择器上已经“准备就绪”的事件
        while (selector.select() > 0) {
            System.out.println("轮一轮");
            //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
            Iterator it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                //8. 获取准备“就绪”的是事件
                SelectionKey sk = it.next();
                //9. 判断具体是什么事件准备就绪
                if (sk.isAcceptable()) {
                    //10. 若“接收就绪”，获取客户端连接
                    SocketChannel sChannel = ssChannel.accept();
                    //11. 切换非阻塞模式
                    sChannel.configureBlocking(false);
                    //12. 将该通道注册到选择器上 这里可以把缓存指定上
                    sChannel.register(selector, SelectionKey.OP_READ);
                } else if (sk.isReadable()) {
                    //13. 获取当前选择器上“读就绪”状态的通道
                    SocketChannel sChannel = (SocketChannel) sk.channel();
                    //14. 读取数据
                    ByteBuffer buf = ByteBuffer.allocate(1024);
                    int len = 0;
                    while ((len = sChannel.read(buf)) > 0) {
                        buf.flip();
                        System.out.println(new String(buf.array(), 0, len));
                        buf.clear();
                    }
                }
                //15. 取消选择键 SelectionKey
                it.remove();
            }
        }
    }
}

尚硅谷代码案例

 package com.atguigu.nio;
 
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 NIOServer {
    public static void main(String[] args) throws  Exception{
        //1. 得到一个ServerSocketChannel对象
        ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
        //2. 得到一个Selector对象
        Selector selector=Selector.open();
        //3. 绑定一个端口号, 在服务器的6666监听 2个方式有什么区别
        //serverSocketChannel.bind(new InetSocketAddress(6666));
        serverSocketChannel.socket().bind(new InetSocketAddress(6666));
        //4. 设置非阻塞方式
        serverSocketChannel.configureBlocking(false);
        //5. 把ServerSocketChannel对象注册给Selector对象
        serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        //6. 干活
        while(true){
            //6.1 监控客户端
            //如果使用 selector.select() 就会阻塞在这里的
            if(selector.select(1000)==0){  //nio非阻塞式的优势
                System.out.println("Server:等待了1秒，无客户端连接");
                continue;
            }
            //6.2 得到SelectionKey,判断通道里的事件
            Iterator keyIterator=selector.selectedKeys().iterator();
            while(keyIterator.hasNext()){
                SelectionKey key=keyIterator.next();
                if(key.isAcceptable()){  //客户端连接请求事件
                    SocketChannel socketChannel=serverSocketChannel.accept();
                    socketChannel.configureBlocking(false);
                    socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                }
                if(key.isReadable()){  //读取客户端数据事件
                    SocketChannel channel=(SocketChannel) key.channel();
                    ByteBuffer buffer=(ByteBuffer) key.attachment();
                    channel.read(buffer);
                    System.out.println("接收到客户端数据："+new String(buffer.array()));
                }
                // 6.3 手动从集合中移除当前key,防止重复处理
                keyIterator.remove();
            }
        }
    }
 
}

 package com.atguigu.nio;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
public class NIOClient {
    public static void main(String[] args) throws Exception{
        //得到一个网络通道
        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 = "hello, 尚硅谷~";
        //Wraps a byte array into a buffer
        ByteBuffer buffer = ByteBuffer.wrap(str.getBytes());
        // 发送数据，将 buffer 数据写入 channel
        socketChannel.write(buffer);
        System.in.read();
    }
}

使用6666端口有个坑爹的地方,端口号被占用。

解决:www.cnblogs.com/jf-67/p/842…

原因:blog.csdn.net/hi_pig2003/…

	//轮询式的获取选择器上已经“准备就绪”的事件,大于0 说明存在准备就绪的事件
	while (selector.select() > 0) {
	System.out.println("轮一轮");
	//7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
	Iterator it = selector.selectedKeys().iterator();
	while (it.hasNext()) {
	//8. 获取准备“就绪”的是事件
	SelectionKey sk = it.next();
	//9. 判断具体是什么事件准备就绪
	if (sk.isAcceptable()) {
	//10. 若“接收就绪”，获取客户端连接
	SocketChannel sChannel = ssChannel.accept();
	//11. 切换非阻塞模式
	sChannel.configureBlocking(false);
	//12. 将该通道注册到选择器上并修改注册事件为read
	sChannel.register(selector, SelectionKey.OP_READ);
	} else if (sk.isReadable()) {
	//13. 获取当前选择器上“读就绪”状态的通道
	SocketChannel sChannel = (SocketChannel) sk.channel();
	//14. 读取数据
	ByteBuffer buf = ByteBuffer.allocate(1024);
	int len = 0;
	while ((len = sChannel.read(buf)) > 0) {
	buf.flip();
	System.out.println(new String(buf.array(), 0, len));
	buf.clear();
	}
	}
	//15. 处理完毕移除选择键 SelectionKey
	it.remove();
	}
	}
	}

	Scanner scan = new Scanner(System.in);
	while(scan.hasNext()){
	String message = scan.nextLine();
	buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
	+ "n" + message).getBytes());
	buf.flip();
	sChannel.write(buf);
	buf.clear();
	}
	//关闭通道
	sChannel.close();

	/**
	客户端
	*/
	public class Client {

	public static void main(String[] args) throws Exception {
	//1. 获取通道 - SelectionKey.OP_ACCEPT 对应监听接收事件
	SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9999));
	//2. 切换非阻塞模式
	sChannel.configureBlocking(false);
	//3. 分配指定大小的缓冲区
	ByteBuffer buf = ByteBuffer.allocate(1024);
	//4. 发送数据给服务端
	Scanner scan = new Scanner(System.in);

	while(scan.hasNext()){
	String message = scan.nextLine();
	buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
	+ "n" + message).getBytes());
	buf.flip();
	//客户端写对应服务器端读就绪
	sChannel.write(buf);
	buf.clear();
	}
	//5. 关闭通道
	sChannel.close();
	}
	}

	/**
	服务端
	*/
	public class Server {
	public static void main(String[] args) throws IOException {
	//1. 获取通道
	ServerSocketChannel ssChannel = ServerSocketChannel.open();
	//2. 切换非阻塞模式
	ssChannel.configureBlocking(false);
	//3. 绑定连接
	ssChannel.bind(new InetSocketAddress(9999));
	//4. 获取选择器
	Selector selector = Selector.open();
	//5. 将通道注册到选择器上, 并且指定“监听接收事件”
	ssChannel.register(selector, SelectionKey.OP_ACCEPT);
	//6. 轮询式的获取选择器上已经“准备就绪”的事件
	while (selector.select() > 0) {
	System.out.println("轮一轮");
	//7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
	Iterator it = selector.selectedKeys().iterator();
	while (it.hasNext()) {
	//8. 获取准备“就绪”的是事件
	SelectionKey sk = it.next();
	//9. 判断具体是什么事件准备就绪
	if (sk.isAcceptable()) {
	//10. 若“接收就绪”，获取客户端连接
	SocketChannel sChannel = ssChannel.accept();
	//11. 切换非阻塞模式
	sChannel.configureBlocking(false);
	//12. 将该通道注册到选择器上这里可以把缓存指定上
	sChannel.register(selector, SelectionKey.OP_READ);
	} else if (sk.isReadable()) {
	//13. 获取当前选择器上“读就绪”状态的通道
	SocketChannel sChannel = (SocketChannel) sk.channel();
	//14. 读取数据
	ByteBuffer buf = ByteBuffer.allocate(1024);
	int len = 0;
	while ((len = sChannel.read(buf)) > 0) {
	buf.flip();
	System.out.println(new String(buf.array(), 0, len));
	buf.clear();
	}
	}
	//15. 取消选择键 SelectionKey
	it.remove();
	}
	}
	}
	}

	package com.atguigu.nio;

	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 NIOServer {
	public static void main(String[] args) throws Exception{
	//1. 得到一个ServerSocketChannel对象
	ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();
	//2. 得到一个Selector对象
	Selector selector=Selector.open();
	//3. 绑定一个端口号, 在服务器的6666监听 2个方式有什么区别
	//serverSocketChannel.bind(new InetSocketAddress(6666));
	serverSocketChannel.socket().bind(new InetSocketAddress(6666));
	//4. 设置非阻塞方式
	serverSocketChannel.configureBlocking(false);
	//5. 把ServerSocketChannel对象注册给Selector对象
	serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
	//6. 干活
	while(true){
	//6.1 监控客户端
	//如果使用 selector.select() 就会阻塞在这里的
	if(selector.select(1000)==0){ //nio非阻塞式的优势
	System.out.println("Server:等待了1秒，无客户端连接");
	continue;
	}
	//6.2 得到SelectionKey,判断通道里的事件
	Iterator keyIterator=selector.selectedKeys().iterator();
	while(keyIterator.hasNext()){
	SelectionKey key=keyIterator.next();
	if(key.isAcceptable()){ //客户端连接请求事件
	SocketChannel socketChannel=serverSocketChannel.accept();
	socketChannel.configureBlocking(false);
	socketChannel.register(selector,SelectionKey.OP_READ, ByteBuffer.allocate(1024));
	}
	if(key.isReadable()){ //读取客户端数据事件
	SocketChannel channel=(SocketChannel) key.channel();
	ByteBuffer buffer=(ByteBuffer) key.attachment();
	channel.read(buffer);
	System.out.println("接收到客户端数据："+new String(buffer.array()));
	}
	// 6.3 手动从集合中移除当前key,防止重复处理
	keyIterator.remove();
	}
	}
	}

	}

9.NIO非阻塞式网络通信入门

服务端流程

客户端流程

代码案例

尚硅谷代码案例

深入探讨：PHP为何无法支持多线程？

Rust async FFI

C/C++ 程序以找到数组乘积除以 n 的余数

Java开发表单数据的多用户协同编辑与同步功能

PHP异常处理指南：如何使用Throwable接口捕获和处理异常