在讲解利用NIO实现通信架构之前,我们需要先来了解一下NIO的基本特点和使用。
NIO | BIO |
---|---|
面向缓冲区(Buffer) | 面向流(Stream) |
非阻塞(Non Blocking IO) | 阻塞IO(Blocking IO) |
选择器(Selectors) |
NIO 有三大核心部分:Channel( 通道) ,Buffer( 缓冲区), Selector( 选择器)
缓冲区本质上是一块可以写入数据,然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象,并提供了一组方法,用来方便的访问该块内存。相比较直接对数组的操作,Buffer API更加容易操作和管理。
Java NIO的通道类似流,但又有些不同:既可以从通道中读取数据,又可以写数据到通道。但流的(input或output)读写通常是单向的。 通道可以非阻塞读取和写入通道,通道可以支持读取或写入缓冲区,也支持异步地读写。
Selector是 一个Java NIO组件,可以能够检查一个或多个 NIO 通道,并确定哪些通道已经准备好进行读取或写入。这样,一个单独的线程可以管理多个channel,从而管理多个网络连接,提高效率
一个用于特定基本数据类 型的容器。由 java.nio 包定义的,所有缓冲区 都是 Buffer 抽象类的子类.。Java NIO 中的 Buffer 主要用于与 NIO 通道进行 交互,数据是从通道读入缓冲区,从缓冲区写入通道中的
Buffer 就像一个数组,可以保存多个相同类型的数据。根 据数据类型不同 ,有以下 Buffer 常用子类:
上述 Buffer 类 他们都采用相似的方法进行管理数据,只是各自 管理的数据类型不同而已。都是通过如下方法获取一个 Buffer 对象:
static XxxBuffer allocate(int capacity) : 创建一个容量为capacity 的 XxxBuffer 对象
Buffer 中的重要概念:
Buffer clear() 清空缓冲区并返回对缓冲区的引用
Buffer flip() 为 将缓冲区的界限设置为当前位置,并将当前位置充值为 0
int capacity() 返回 Buffer 的 capacity 大小
boolean hasRemaining() 判断缓冲区中是否还有元素
int limit() 返回 Buffer 的界限(limit) 的位置
Buffer limit(int n) 将设置缓冲区界限为 n, 并返回一个具有新 limit 的缓冲区对象
Buffer mark() 对缓冲区设置标记
int position() 返回缓冲区的当前位置 position
Buffer position(int n) 将设置缓冲区的当前位置为 n , 并返回修改后的 Buffer 对象
int remaining() 返回 position 和 limit 之间的元素个数
Buffer reset() 将位置 position 转到以前设置的 mark 所在的位置
Buffer rewind() 将位置设为为 0, 取消设置的 mark
Buffer 所有子类提供了两个用于数据操作的方法:get()put() 方法
取获取 Buffer中的数据
get() :读取单个字节
get(byte[] dst):批量读取多个字节到 dst 中
get(int index):读取指定索引位置的字节(不会移动 position)
放到 入数据到 Buffer 中 中
put(byte b):将给定单个字节写入缓冲区的当前位置
put(byte[] src):将 src 中的字节写入缓冲区的当前位置
put(int index, byte b):将指定字节写入缓冲区的索引位置(不会移动 position)
使用Buffer读写数据一般遵循以下四个步骤:
public class TestBuffer {
@Test
public void test3(){
//分配直接缓冲区
ByteBuffer buf = ByteBuffer.allocateDirect(1024);
System.out.println(buf.isDirect());
}
@Test
public void test2(){
String str = "itheima";
ByteBuffer buf = ByteBuffer.allocate(1024);
buf.put(str.getBytes());
buf.flip();
byte[] dst = new byte[buf.limit()];
buf.get(dst, 0, 2);
System.out.println(new String(dst, 0, 2));
System.out.println(buf.position());
//mark() : 标记
buf.mark();
buf.get(dst, 2, 2);
System.out.println(new String(dst, 2, 2));
System.out.println(buf.position());
//reset() : 恢复到 mark 的位置
buf.reset();
System.out.println(buf.position());
//判断缓冲区中是否还有剩余数据
if(buf.hasRemaining()){
//获取缓冲区中可以操作的数量
System.out.println(buf.remaining());
}
}
@Test
public void test1(){
String str = "itheima";
//1. 分配一个指定大小的缓冲区
ByteBuffer buf = ByteBuffer.allocate(1024);
System.out.println("-----------------allocate()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
//2. 利用 put() 存入数据到缓冲区中
buf.put(str.getBytes());
System.out.println("-----------------put()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
//3. 切换读取数据模式
buf.flip();
System.out.println("-----------------flip()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
//4. 利用 get() 读取缓冲区中的数据
byte[] dst = new byte[buf.limit()];
buf.get(dst);
System.out.println(new String(dst, 0, dst.length));
System.out.println("-----------------get()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
//5. rewind() : 可重复读
buf.rewind();
System.out.println("-----------------rewind()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
//6. clear() : 清空缓冲区. 但是缓冲区中的数据依然存在,但是处于“被遗忘”状态
buf.clear();
System.out.println("-----------------clear()----------------");
System.out.println(buf.position());
System.out.println(buf.limit());
System.out.println(buf.capacity());
System.out.println((char)buf.get());
}
}
什么是直接内存与非直接内存
根据官方文档的描述:
byte byffer
可以是两种类型,一种是基于直接内存(也就是非堆内存);另一种是非直接内存(也就是堆内存)。对于直接内存来说,JVM将会在IO操作上具有更高的性能,因为它直接作用于本地系统的IO操作。而非直接内存,也就是堆内存中的数据,如果要作IO操作,会先从本进程内存复制到直接内存,再利用本地IO处理。
从数据流的角度,非直接内存是下面这样的作用链:
本地IO-->直接内存-->非直接内存-->直接内存-->本地IO
而直接内存是:
本地IO-->直接内存-->本地IO
很明显,在做IO处理时,比如网络发送大量数据时,直接内存会具有更高的效率。直接内存使用allocateDirect创建,但是它比申请普通的堆内存需要耗费更高的性能。不过,这部分的数据是在JVM之外的,因此它不会占用应用的内存。所以呢,当你有很大的数据要缓存,并且它的生命周期又很长,那么就比较适合使用直接内存。只是一般来说,如果不是能带来很明显的性能提升,还是推荐直接使用堆内存。字节缓冲区是直接缓冲区还是非直接缓冲区可通过调用其 isDirect() 方法来确定。
使用场景
定义
直接内存
:NIO的Buffer提供了一个可以不经过JVM内存直接访问系统物理内存的类——DirectBuffer。 DirectBuffer类继承自ByteBuffer,但和普通的ByteBuffer不同,普通的ByteBuffer仍在JVM堆上分配内存,其最大内存受到最大堆内存的限制;而DirectBuffer直接分配在物理内存中,并不占用堆空间,其可申请的最大内存受操作系统限制。
直接内存有时候也被称为堆外内存:
堆内存
:Java 虚拟机具有一个堆,堆是运行时数据区域,所有类实例和数组的内存均从此处分配。堆是在 Java 虚拟机启动时创建的。对象的堆内存由称为垃圾回收器的自动内存管理系统回收。
区别和应用场景
通道(Channel):由 java.nio.channels 包定义 的。Channel 表示 IO 源与目标打开的连接。 Channel 类似于传统的“流”。只不过 Channel 本身不能直接访问数据,Channel 只能与 Buffer 进行交互。
1、 NIO 的通道类似于流,但有些区别如下:
2、BIO 中的 stream 是单向的,例如 FileInputStream 对象只能进行读取数据的操作,而 NIO 中的通道(Channel) 是双向的,可以读操作,也可以写操作。
3、Channel 在 NIO 中是一个接口
public interface Channel extends Closeable{}
获取通道的一种方式是对支持通道的对象调用getChannel() 方法。支持通道的类如下:
int read(ByteBuffer dst) 从 从 Channel 到 中读取数据到 ByteBuffer
long read(ByteBuffer[] dsts) 将 将 Channel 到 中的数据“分散”到 ByteBuffer[]
int write(ByteBuffer src) 将 将 ByteBuffer 到 中的数据写入到 Channel
long write(ByteBuffer[] srcs) 将 将 ByteBuffer[] 到 中的数据“聚集”到 Channel
long position() 返回此通道的文件位置
FileChannel position(long p) 设置此通道的文件位置
long size() 返回此通道的文件的当前大小
FileChannel truncate(long s) 将此通道的文件截取为给定大小
void force(boolean metaData) 强制将所有对此通道的文件更新写入到存储设备中
需求:使用前面学习后的 ByteBuffer
(缓冲) 和 FileChannel
(通道), 将 “你好,大忽悠” 写入到 data.txt 中.
public class ChannelTest {
@Test
public void write(){
try {
// 1、字节输出流通向目标文件
FileOutputStream fos = new FileOutputStream("dhy.txt");
// 2、得到字节输出流对应的通道Channel
FileChannel channel = fos.getChannel();
// 3、分配缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put("你好,大忽悠!".getBytes());
// 4、把缓冲区切换成写出模式
buffer.flip();
channel.write(buffer);
channel.close();
System.out.println("写数据到文件中!");
} catch (Exception e) {
e.printStackTrace();
}
}
}
相对路径是相对于当前模块而言
需求:使用前面学习后的 ByteBuffer(缓冲) 和 FileChannel(通道), 将 data01.txt 中的数据读入到程序,并显示在控制台屏幕
public class ChannelTest {
@Test
public void read() throws Exception {
// 1、定义一个文件字节输入流与源文件接通
FileInputStream is = new FileInputStream("data01.txt");
// FileOutputStream fos = new FileOutputStream()
// 2、需要得到文件字节输入流的文件通道
FileChannel channel = is.getChannel();
// 3、定义一个缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
// 4、读取数据到缓冲区
channel.read(buffer);
buffer.flip();
// 5、读取出缓冲区中的数据并输出即可
String rs = new String(buffer.array(),0,buffer.remaining());
System.out.println(rs);
}
因为channel是双向的,既可以读也可以写,因此上面也可以通过文件输出流得到通道后,进行通道内数据的读取操作
使用 FileChannel(通道) ,完成文件的拷贝。
@Test
public void copy() throws Exception {
// 源文件
File srcFile = new File("C:\\Users\\dlei\\Desktop\\BIO,NIO,AIO\\文件\\壁纸.jpg");
File destFile = new File("C:\\Users\\dlei\\Desktop\\BIO,NIO,AIO\\文件\\壁纸new.jpg");
// 得到一个字节字节输入流
FileInputStream fis = new FileInputStream(srcFile);
// 得到一个字节输出流
FileOutputStream fos = new FileOutputStream(destFile);
// 得到的是文件通道
FileChannel isChannel = fis.getChannel();
FileChannel osChannel = fos.getChannel();
// 分配缓冲区
ByteBuffer buffer = ByteBuffer.allocate(1024);
while(true){
// 必须先清空缓冲然后再写入数据到缓冲区
buffer.clear();
// 开始读取一次数据
int flag = isChannel.read(buffer);
if(flag == -1){
break;
}
// 已经读取了数据 ,把缓冲区的模式切换成可读模式
buffer.flip();
// 把数据写出到
osChannel.write(buffer);
}
isChannel.close();
osChannel.close();
System.out.println("复制完成!");
}
下面简化写法:
try(FileInputStream inputStream=new FileInputStream("dst.txt");
FileOutputStream outputStream=new FileOutputStream(UUID.randomUUID().toString().substring(0,5) +".txt");
FileChannel in=inputStream.getChannel();
FileChannel out=outputStream.getChannel();) {
ByteBuffer buffer=ByteBuffer.allocate(1024);
while(in.read(buffer)!=-1)
{
buffer.flip();
out.write(buffer);
buffer.clear();
}
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
分散读取(Scatter ):是指把Channel通道的数据读入到多个缓冲区中去
聚集写入(Gathering )是指将多个 Buffer 中的数据“聚集”到 Channel。
public class TestBuffer
{
//分散和聚集
@Test
public void test() throws IOException{
RandomAccessFile raf1 = new RandomAccessFile("1.txt", "rw");
//1. 获取通道
FileChannel channel1 = raf1.getChannel();
//2. 分配指定大小的缓冲区
ByteBuffer buf1 = ByteBuffer.allocate(100);
ByteBuffer buf2 = ByteBuffer.allocate(1024);
//3. 分散读取
ByteBuffer[] bufs = {buf1, buf2};
channel1.read(bufs);
for (ByteBuffer byteBuffer : bufs)
{
System.out.println("反转前: ");
System.out.println("position: "+byteBuffer.position());
System.out.println("limit: "+byteBuffer.limit());
System.out.println("capacity: "+byteBuffer.capacity());
byteBuffer.flip();
System.out.println("反转后: ");
System.out.println("position: "+byteBuffer.position());
System.out.println("limit: "+byteBuffer.limit());
System.out.println("capacity: "+byteBuffer.capacity());
}
System.out.println(new String(bufs[0].array(), 0, bufs[0].limit()));
System.out.println("-----------------");
System.out.println(new String(bufs[1].array(), 0, bufs[1].limit()));
//4. 聚集写入
RandomAccessFile raf2 = new RandomAccessFile("2.txt", "rw");
FileChannel channel2 = raf2.getChannel();
channel2.write(bufs);
}
}
从目标通道中去复制原通道数据
@Test
public void test02() throws Exception {
// 1、字节输入管道
FileInputStream is = new FileInputStream("data01.txt");
FileChannel isChannel = is.getChannel();
// 2、字节输出流管道
FileOutputStream fos = new FileOutputStream("data03.txt");
FileChannel osChannel = fos.getChannel();
// 3、复制
osChannel.transferFrom(isChannel,isChannel.position(),isChannel.size());
isChannel.close();
osChannel.close();
}
把原通道数据复制到目标通道
@Test
public void test02() throws Exception {
// 1、字节输入管道
FileInputStream is = new FileInputStream("data01.txt");
FileChannel isChannel = is.getChannel();
// 2、字节输出流管道
FileOutputStream fos = new FileOutputStream("data04.txt");
FileChannel osChannel = fos.getChannel();
// 3、复制
isChannel.transferTo(isChannel.position() , isChannel.size() , osChannel);
isChannel.close();
osChannel.close();
}
MappedByteBuffer map(FileChannel.MapMode mode,long position,long size)
方法的作用是将此通道的文件区域直接映射到内存中。
1)mode
:根据只读、读取/写入或专用(写入时复制)来映射文件,分别为FileChannel.MapMode
类中所定义的READ_ONLY、READ_WRITE和PRIVATE;
2)position
:文件中的位置,映射区域从此位置开始;必须为非负数。
3)size
:要映射的区域大小;必须为非负数且不大于Integer.MAX_VALUE。
可以通过下列3种模式将文件区域映射到内存中。
1)只读:试图修改得到的缓冲区将导致抛出ReadOnlyBufferException
异常。(MapMode.READ_ONLY
)
2)读取/写入:对得到的缓冲区的更改最终将传播到文件;该更改对映射到同一文件的其他程序不一定是可见的。(MapMode.READ_WRITE
)
3)专用:对得到的缓冲区的更改不会传播到文件,并且该更改对映射到同一文件的其他程序也不是可见的;相反,会创建缓冲区已修改部分的专用副本。(MapMode.PRIVATE
)
总结:
1、对于只读映射关系,此通道必须可以进行读取操作;对于读取/写入或专用映射关系,此通道必须可以进行读取和写入操作。
2、此方法返回的已映射字节缓冲区位置为零,限制和容量为size;其标记是不确定的。在缓冲区本身被作为垃圾回收之前,该缓冲区及其表示的映射关系都是有效的。
3、映射关系一经创建,就不再依赖于创建它时所用的文件通道。特别是关闭该通道对映射关系的有效性没有任何影响。
4、对于大多数操作系统而言,与通过普通的read()和write()
方法读取或写入数千字节的数据相比,将文件映射到内存中开销更大。从性能的观点来看,通常将相对较大的文件映射到内存中才是值得的。
MappedByteBuffer的简单介绍:
它是直接字节缓冲区
,其内容是文件的内存映射区域。映射的字节缓冲区是通过FileChannel.map()方法创建的。此类用特定于内存映射文件区域的操作扩展ByteBuffer类。
public abstract class MappedByteBuffer extends ByteBuffer
作为ByteBuffer的子类,除了具有父类的方法外,还新增了
force()
将此缓冲区所做的内容更改强制写入包含映射文件的存储设备中。load()
将此缓冲区内容加载到物理内存中。isLoaded()
判断次缓冲区的内容是否位于物理内存中。FileChannel类或MappedByteBuffer类对文件进行操作时,在大部分情况下,它们的效率并不比使用InputStream或OutputStream高很多,这是因为NIO的出现是为了解决操作I/O线程阻塞的问题,使用NIO就把线程变成了非阻塞,这样就提高了运行效率。
NIO真正的优势:非阻塞。
使用演示:
@Test
public void userMap() throws Exception
{
//源文件通道
FileChannel fileChannel = new FileInputStream("1.txt").getChannel();
//以当前模块为相对路径
RandomAccessFile desc=new RandomAccessFile("desc.txt","rw");
//将1.txt映射到直接字节缓冲区
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, fileChannel.size());
//将缓冲区数据输出到目的文件中
desc.getChannel().write(mappedByteBuffer);
//转换读写模式,将缓冲区内的数据打印到控制台上
mappedByteBuffer.flip();
//通过字符集对象进行解码,得到对应解码后的字符缓冲区
Charset charset=Charset.defaultCharset();
CharBuffer charBuffer = charset.decode(mappedByteBuffer);
System.out.println(charBuffer);
}
选择器(Selector) 是 SelectableChannle 对象的多路复用器,Selector 可以同时监控多个 SelectableChannel 的 IO 状况,也就是说,利用 Selector可使一个单独的线程管理多个 Channel。Selector 是非阻塞 IO 的核心
创建 Selector :通过调用 Selector.open() 方法创建一个 Selector。
Selector selector = Selector.open();
向选择器注册通道:SelectableChannel.register(Selector sel, int ops)
//1. 获取通道
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2. 切换非阻塞模式
ssChannel.configureBlocking(false);
//3. 绑定连接
ssChannel.bind(new InetSocketAddress(9898));
//4. 获取选择器
Selector selector = Selector.open();
//5. 将通道注册到选择器上, 并且指定“监听接收事件”
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
FileChannel是阻塞的,不能够设置为非阻塞,因此不能注册到选择器上面
当调用 register(Selector sel, int ops) 将通道注册选择器时,选择器对通道的监听事件,需要通过第二个参数 ops 指定。可以监听的事件类型(用 可使用 SelectionKey 的四个常量 表示):
int interestSet = SelectionKey.OP_READ|SelectionKey.OP_WRITE
Selector可以实现: 一个 I/O 线程可以并发处理 N 个客户端连接和读写操作,这从根本上解决了传统同步阻塞 I/O 一连接一线程模型,架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。
1、当客户端连接服务端时,服务端会通过 ServerSocketChannel 得到 SocketChannel: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<SelectionKey> 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();
}
}
}
为什么最后要从集合中移除选择键
每个Selector对象都需要维护以下三个集合:
这里我们移除的是已选择键的集合,因为如果处理完了一个已选择键,而不进行移除,那么下一轮循环这个键依然存在,就会造成重复处理
SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9999));
sChannel.configureBlocking(false);
ByteBuffer buf = ByteBuffer.allocate(1024);
Scanner scan = new Scanner(System.in);
while(scan.hasNext()){
String str = scan.nextLine();
buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
+ "\n" + str).getBytes());
buf.flip();
sChannel.write(buf);
buf.clear();
}
//关闭通道
sChannel.close();
需求:服务端接收客户端的连接请求,并接收多个客户端发送过来的事件。
客户端:
/**
客户端
*/
public class Client {
public static void main(String[] args) throws Exception {
//1. 获取通道
//写法一:
//SocketChannel sChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", 9999));
//写法二:
SocketChannel sChannel=SocketChannel.open();
//规定客户端的端口--一定要在连接服务器之前规定,否则会随机分配一个端口,再bind就会抛出已绑定的异常
sChannel.bind(new InetSocketAddress(8082));
//连接服务器
sChannel.connect(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 str = scan.nextLine();
buf.put((new SimpleDateFormat("yyyy/MM/dd HH:mm:ss").format(System.currentTimeMillis())
+ "\n" + str).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) {
//7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
while (it.hasNext()) {
//8. 获取准备“就绪”的是事件
SelectionKey sk = it.next();
//9. 判断具体是什么事件准备就绪
if (sk.isAcceptable())
{
//10. 若“接收就绪”,获取客户端连接
SocketChannel sChannel = ssChannel.accept();
InetSocketAddress remoteAddress = (InetSocketAddress)sChannel.getRemoteAddress();
InetSocketAddress localAddress = (InetSocketAddress)sChannel.getLocalAddress();
System.out.println("客户端连接事件: ");
System.out.println("客户端的HostName: "+remoteAddress.getHostName());
System.out.println("客户端的HostString: "+remoteAddress.getHostString());
System.out.println("客户端的Address: "+remoteAddress.getAddress());
System.out.println("客户端的Port: "+remoteAddress.getPort());
//11. 切换非阻塞模式
sChannel.configureBlocking(false);
//12. 将该通道注册到选择器上
sChannel.register(selector, SelectionKey.OP_READ);
} else if (sk.isReadable()) {
System.out.println("客户端发送数据事件: ");
//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();
}
}
}
}
输出:
客户端连接事件:
客户端的HostName: config-3344.com
客户端的HostString: config-3344.com
客户端的Address: config-3344.com/127.0.0.1
客户端的Port: 8081
客户端连接事件:
客户端的HostName: config-3344.com
客户端的HostString: config-3344.com
客户端的Address: config-3344.com/127.0.0.1
客户端的Port: 8082
客户端发送数据事件:
2022/01/15 22:30:49
你好
客户端发送数据事件:
2022/01/15 22:30:54
你好
需求:进一步理解 NIO 非阻塞网络编程机制,实现多人群聊
package dhy.com;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
public class Server {
//定义属性
private Selector selector;
private ServerSocketChannel ssChannel;
private static final int PORT = 80;
//构造器
//初始化工作
public Server() {
try {
// 1、获取通道
ssChannel = ServerSocketChannel.open();
// 2、切换为非阻塞模式
ssChannel.configureBlocking(false);
// 3、绑定连接的端口
ssChannel.bind(new InetSocketAddress(PORT));
// 4、获取选择器Selector
selector = Selector.open();
// 5、将通道都注册到选择器上去,并且开始指定监听接收事件
ssChannel.register(selector , SelectionKey.OP_ACCEPT);
}catch (IOException e) {
e.printStackTrace();
}
}
//监听
public void listen() {
System.out.println("监听线程: " + Thread.currentThread().getName());
try {
//select在没有事件触发的情况下,会阻塞住
while (selector.select() > 0){
System.out.println("开始一轮事件处理~~~");
// 7、获取选择器中的所有注册的通道中已经就绪好的事件
Iterator<SelectionKey> it = selector.selectedKeys().iterator();
// 8、开始遍历这些准备好的事件
while (it.hasNext()){
// 提取当前这个事件
SelectionKey sk = it.next();
// 9、判断这个事件具体是什么
if(sk.isAcceptable()){
// 10、直接获取当前接入的客户端通道
SocketChannel schannel = ssChannel.accept();
// 11 、切换成非阻塞模式
schannel.configureBlocking(false);
// 12、将本客户端通道注册到选择器
System.out.println(schannel.getRemoteAddress() + " 上线 ");
schannel.register(selector , SelectionKey.OP_READ);
//提示
}
//当远程客户端断开连接的时候,也会触发OP_READ事件
else if(sk.isReadable()){
//处理读 (专门写方法..)
readData(sk);
}
it.remove(); // 处理完毕之后需要移除当前事件
}
}
}catch (Exception e) {
e.printStackTrace();
}finally {
//发生异常处理....
System.out.println("监听结束....");
}
}
//读取客户端消息
private void readData(SelectionKey key) {
//取到关联的channle
SocketChannel channel = null;
try {
//得到channel
channel = (SocketChannel) key.channel();
//创建buffer
ByteBuffer buffer = ByteBuffer.allocate(1024);
//count为读取到的字节数
int count = channel.read(buffer);
//根据count的值做处理
if(count > 0) {
//把缓存区的数据转成字符串
String msg = new String(buffer.array(),0,buffer.remaining());
//输出该消息
System.out.println("form 客户端: " + msg);
//向其它的客户端转发消息(去掉自己), 专门写一个方法来处理
sendInfoToOtherClients(msg, channel);
}
}catch (IOException e) {
try {
System.out.println(channel.getRemoteAddress() + " 离线了..");
e.printStackTrace();
//取消注册
key.cancel();
//关闭通道
channel.close();
}catch (IOException e2) {
e2.printStackTrace();;
}
}
}
//转发消息给其它客户(通道)
private void sendInfoToOtherClients(String msg, SocketChannel self ) throws IOException{
System.out.println("服务器转发消息中...");
System.out.println("服务器转发数据给客户端线程: " + Thread.currentThread().getName());
//遍历 所有注册到selector 上的 SocketChannel,并排除 self
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());
//将buffer 的数据写入 通道
dest.write(buffer);
}
}
}
public static void main(String[] args) {
//创建服务器对象
Server groupChatServer = new Server();
groupChatServer.listen();
}
}
package dhy.com;
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.SocketChannel;
import java.util.Iterator;
import java.util.Scanner;
public class Client {
//定义相关的属性
private final String HOST = "127.0.0.1"; // 服务器的ip
private final int PORT = 80; //服务器端口
private Selector selector;
private SocketChannel socketChannel;
private String username;
//构造器, 完成初始化工作
public Client() throws IOException {
selector = Selector.open();
//连接服务器
socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1", PORT));
//设置非阻塞
socketChannel.configureBlocking(false);
//将channel 注册到selector
socketChannel.register(selector, SelectionKey.OP_READ);
//得到username
username = socketChannel.getLocalAddress().toString().substring(1);
System.out.println(username + " is ok...");
}
//向服务器发送消息
public void sendInfo(String info) {
info = username + " 说:" + info;
try {
socketChannel.write(ByteBuffer.wrap(info.getBytes()));
}catch (IOException e) {
e.printStackTrace();
}
}
//读取从服务器端回复的消息
public void readInfo() {
try {
//感兴趣的事件触发
while(selector.select()>0)
{
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
if(key.isReadable()) {
//得到相关的通道
SocketChannel sc = (SocketChannel) key.channel();
//得到一个Buffer
ByteBuffer buffer = ByteBuffer.allocate(1024);
//读取
sc.read(buffer);
//把读到的缓冲区的数据转成字符串
String msg = new String(buffer.array());
System.out.println(msg.trim());
}
}
iterator.remove(); //删除当前的selectionKey, 防止重复操作
}
}catch (Exception e) {
e.printStackTrace();
}
}
public static void main(String[] args) throws Exception {
//启动我们客户端
Client chatClient = new Client();
//启动一个线程,读取从服务器发送数据
new Thread() {
@Override
public void run() {
chatClient.readInfo();
}
}.start();
//发送数据给服务器端
Scanner scanner = new Scanner(System.in);
while (scanner.hasNextLine()) {
String s = scanner.nextLine();
chatClient.sendInfo(s);
}
}
}
AIO
异步非阻塞,基于NIO的,可以称之为NIO2.0
BIO NIO AIO
Socket SocketChannel AsynchronousSocketChannel
ServerSocket ServerSocketChannel AsynchronousServerSocketChannel
与NIO不同,当进行读写操作时,只须直接调用API的read或write方法即可, 这两种方法均为异步的,对于读操作而言,当有流可读取时,操作系统会将可读的流传入read方法的缓冲区,对于写操作而言,当操作系统将write方法传递的流写入完毕时,操作系统主动通知应用程序
即可以理解为,read/write方法都是异步的,完成后会主动调用回调函数。在JDK1.7中,这部分内容被称作NIO.2,主要在Java.nio.channels包下增加了下面四个异步通道:
AsynchronousSocketChannel
AsynchronousServerSocketChannel
AsynchronousFileChannel
AsynchronousDatagramChannel
BIO、NIO、AIO:
BIO、NIO、AIO适用场景分析: