NIO记录

曾经终败给现在 2021-09-24 16:20 203阅读 0赞

## NIO—NonBlocking IO(new IO) ##

1.  io面向流编程，只能作为输入或者输出流的一种，是同步阻塞的，每一个连接过来都要创建一个线程去处理，线程上下文切换开销很大，造成了很大的瓶颈
2.  于是有了线程池实现的伪阻塞IO，一定程度解决了线程创建过多的问题，但是没有从根本上解决阻塞的问题，并且线程过多而线程池过小时也会造成很大的瓶颈
3.  既然根本瓶颈原因是线程数和阻塞IO，那么我们有没有办法只用1个线程去处理多个客户端连接呢？这就是NIO出现的原因

NIO主要有**三个核心部分组成**：

*  **buffer缓冲区**
 *  **Channel管道**
 *  **Selector选择器**

> nio面向block块，buffer缓冲区编程，底层是数组，buffer提供数据访问，channel读写到buffer，buffer读写到channel，从buffer读取到程序channel是双向的
> 
> 理解NIO需要理解事件编程模型

NIO核心：

NIO由原来的阻塞读写（占用线程）变成了单线程轮询事件，找到可以进行读写的网络描述符进行读写。除了事件的轮询是阻塞的（没有可干的事情必须要阻塞），剩余的I/O操作都是纯CPU操作，没有必要开启多线程。

> 单线程处理I/O的效率确实非常高，没有线程切换，只是拼命的读、写、选择事件。

NIO带个我们：

1.  事件驱动模型—异步编程都离不开事件
2.  单线程处理多连接—多路复用使得处理更加高效
3.  非阻塞IO，只阻塞获取可操作事件
4.  基于block传输比基于流传输更加高效
5.  零拷贝—DirectBuffer

缺点：

NIO并没有完全屏蔽平台差异，它仍然是基于各个操作系统的I/O系统实现的，差异仍然存在。使用NIO做网络编程构建事件驱动模型并不容易，陷阱重重。

> 推荐使用NIO成熟框架Netty

### Buffer ###

**缓冲区本质上是一块可以写入数据，然后可以从中读取数据的内存。这块内存被包装成NIO Buffer对象，并提供了一组方法，用来方便的访问该块内存。**

Capacity、Position、Limit

![b7a5790209d76e451917b765e79bbcd5.png][] 0 `<=` *mark* `<=` *position* `<=` *limit* `<=` *capacity*

*  capacity

> 作为一个内存块，Buffer有一个固定的大小值，也叫“capacity”.你只能往里写capacity个byte、long，char等类型。一旦Buffer满了，需要将其清空（通过读数据或者清除数据）才能继续写数据往里写数据。

*  position

> 当你写数据到Buffer中时，position表示当前的位置。初始的position值为0.当一个byte、long等数据写到Buffer后， position会向前移动到下一个可插入数据的Buffer单元。position最大可为capacity – 1.

当读取数据时，也是从某个特定位置读。当将Buffer从写模式切换到读模式，position会被重置为0. 当从Buffer的position处读取数据时，position向前移动到下一个可读的位置。

*  limit

> 在写模式下，Buffer的limit表示你最多能往Buffer里写多少数据。 写模式下，limit等于Buffer的capacity。

当切换Buffer到读模式时， limit表示你最多能读到多少数据。因此，当切换Buffer到读模式时，limit会被设置成写模式下的position值。换句话说，你能读到之前写入的所有数据（limit被设置成已写数据的数量，这个值在写模式下就是position）

**同一个buffer可以存储不同数据类型的数据，但是获取的时候要指定类型获取**

1.  `ByteBuffer buffer = ByteBuffer.allocate(1024);`
2.  `buffer.putInt(1);`
3.  `buffer.putLong(387524875628742L);`
4.  `buffer.putChar('s');`
5.  `buffer.flip();`
6.  `System.out.println(buffer.getInt());`
7.  `System.out.println(buffer.getLong());`
8.  `System.out.println(buffer.getChar());`

> put方法只能放入byte型，不能放入int

flip、clear、rewind、mark

*  flip

**flip方法将Buffer从写模式切换到读模式。调用flip()方法会将position设回0，并将limit设置成之前position的值。**

1.  `public final Buffer flip() {`
2.  `this.limit = this.position;`
3.  `this.position = 0;`
4.  `this.mark = -1;`
5.  `return this;`
6.  `}`

*  clear

**position将被设回0，limit被设置成 capacity的值。换句话说，Buffer 被清空了。Buffer中的数据并未清除，只是这些标记告诉我们可以从哪里开始往Buffer里写数据。**

1.  `public final Buffer clear() {`
2.  `this.position = 0;`
3.  `this.limit = this.capacity;`
4.  `this.mark = -1;`
5.  `return this;`
6.  `}`

*  rewind

**Buffer.rewind()将position设回0，所以你可以重读Buffer中的所有数据。limit保持不变，仍然表示能从Buffer中读取多少个元素**

1.  `public final Buffer rewind() {`
2.  `this.position = 0;`
3.  `this.mark = -1;`
4.  `return this;`
5.  `}`

*  mark

**可以标记Buffer中的一个特定position。之后可以通过调用Buffer.reset()方法恢复到这个position。**

1.  `public final Buffer mark() {`
2.  `this.mark = this.position;`
3.  `return this;`
4.  `}`

*  slice分片

将buffer根据设置的position和limit分片一个buffer，有自己的position、limit和capacity，数据共用一个内存地址的buffer数据

1.  `public static void test2(){`
2.  `ByteBuffer buffer = ByteBuffer.allocate(1024);`
3.  `for(int i=0;i<buffer.capacity();i++){`
4.  `buffer.put((byte)i);`
5.  `}`
6.  `buffer.position(10);`
7.  `buffer.limit(20);`
8.  `ByteBuffer buffer1 = buffer.slice();//buffer分片`
9.  `for(int m=0;m<buffer1.capacity();m++){`
10. `byte b = buffer1.get();`
11. `System.out.print(b+" ");`
12. `}`
13. `}`
14.  
15. `输出：`
16. `10 11 12 13 14 15 16 17 18 19`

ReadOnlyBuffer

普通的Buffer(可读可写)可以随时转换为只读Buffer，但是只读Buffer不可以转换为普通Buffer

ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();

> 转换后的Buffer是一个新的只读Buffer，拥有独立的position、limit和capacity

DirectBuffer

`堆外内存buffer，本地JNI非JVM堆内存buffer，允许直接访问`

普通ByteBuffer由JVM管理，在JVM堆上分配内存

ByteBuffer buf = ByteBuffer.allocate(1024);

DirectBuffer会在本地内存中分配，脱离JVM的堆管理

ByteBuffer buf = ByteBuffer.allocateDirect(1024);

> 为什么要这样做呢？
> 
> \----------又是GC--------
> 
> 我们都知道JVM 在堆上的老年代中，GC时会采取`标记-整理`策略，会使得对象在堆内存中的地址发生变化，整理时会buffer太大时会很难gc整理

所以出现了`DirectBuffer`，它使用`unsafe.allocateMemory`分配内存，是一个native方法，由buffer的`address`变量记录这个内存的地址来提供访问

比较

*  DirectBuffer：本地方法分配内存显然没有JVM堆分配快，但是涉及`IO`和`网络IO`的话就是DirectBuffer比较快了

> DirectByteBuffer继承了MappedByteBuffer
> 
> 缓存的使用可以使用DirectByteBuffer和HeapByteBuffer。如果使用了DirectByteBuffer，一般来说可以减少一次系统空间到用户空间的拷贝。
> 
> 数据量比较小的中小应用情况下，可以考虑使用heapBuffer；反之可以用directBuffer

MappedByteBuffer

**映射到堆外内存的ByteBuffer，DirectByteBuffer继承此类实现堆外内存的分配**

通过下面方式映射buffer到堆外内存

MappedByteBuffer mappedByteBuffer = channel.map(MapMode.READ_WRITE, 0, channel.size());

使用拷贝文件：

1.  `RandomAccessFile in = new RandomAccessFile("nio/1.txt", "rw");`
2.  `RandomAccessFile out = new RandomAccessFile("nio/2.txt", "rw");`
3.  `FileChannel inChannel = in.getChannel();`
4.  `FileChannel outChannel = out.getChannel();`
5.  `MappedByteBuffer inputData = inChannel.map(FileChannel.MapMode.READ_ONLY,0,new File("nio/1.txt").length());`
6.  `Charset charset = Charset.forName("utf-8");//编码`
7.  `CharsetDecoder decoder = charset.newDecoder();`
8.  `CharsetEncoder encoder = charset.newEncoder();`
9.  `CharBuffer charBuffer = decoder.decode(inputData);`
10. `ByteBuffer buffer = encoder.encode(charBuffer);`
11. `outChannel.write(buffer);`
12. `in.close();out.close();`

### Channel—通道 ###

FileChannel

NIO提供的一种连接到文件的通道，用于文件的读写

**在使用FileChannel时，需要从`输入输出流或者RandomAccessFile`中获取FIleChannel**

*  如果要向FileChannel中读取数据，需要申请一个ByteBuffer，将数据从FileChannel中读取到缓冲区ByteBuffer，`read()`返回多少个字节被读取，如果返回-1说明文件已经到达末尾
 *  如果要向FileChannel中写入数据，需要先将数据写入到ByteBuffer中，在从ByteBuffer中写入到FileChannel中，调用`write()`方法

> 注意读写之间需要Buffer.flip();

例子：

1.读取文件数据并打印

1.  `FileInputStream fileInputStream = new FileInputStream("1.log");`
2.  `FileChannel channel = fileInputStream.getChannel();`
3.  `ByteBuffer byteBuffer = ByteBuffer.allocate(512);;`
4.  `channel.read(byteBuffer);`
5.  `byteBuffer.flip();`
6.  `while(byteBuffer.remaining()>0){`
7.  `byte b = byteBuffer.get();`
8.  `System.out.println((char) b);`
9.  `}`
10. `fileInputStream.close();`

2.把1.txt数据写入2.txt

1.  `FileInputStream inputStream = new FileInputStream("1.txt");`
2.  `FileChannel in = inputStream.getChannel();`
3.  `FileOutputStream outputStream = new FileOutputStream("2.txt");`
4.  `FileChannel out = outputStream.getChannel();`
5.  `ByteBuffer byteBuffer = ByteBuffer.allocate(1024);`
6.  `while(true){`
7.  `byteBuffer.clear();//没有的话会一直读取`
8.  `int read = in.read(byteBuffer);`
9.  `System.out.println("read:"+read);`
10. `if(read==-1){`
11. `break;//为-1表示文件结束 返回`
12. `}`
13. `byteBuffer.flip();`
14. `out.write(byteBuffer);`
15. `}`
16. `inputStream.close();`
17. `outputStream.close();`

ServerSockerChannel

NIO提供了一种可以监听新进入的TCP连接的通道，就是`ServerSocketChannel`，对应IO中`ServerSocket`

*  打开监听通道

1.  `ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();`
2.  `serverSocketChannel.socket().bind(new InetSocketAddress(9999));`
3.  `while(true){`
4.  `SocketChannel socketChannel = serverSocketChannel.accept();`
5.  `//do something with socketChannel...`
6.  `}`

SocketChannel

NIO提供的一种连接到TCP套接字的通道，就是`SocketChannel`，对应IO中`Socket`

*  打开一个SocketChannel

1.  `SocketChannel socketChannel = SocketChannel.open();`
2.  `socketChannel.connect(new InetSocketAddress("localhost", 8080));`

Channel读写

*  读

1.  `ByteBuffer buf = ByteBuffer.allocate(48);`
2.  `int bytesRead = socketChannel.read(buf);`

*  写

1.  `ByteBuffer writeBuffer = ByteBuffer.allocate(48);`
2.  `String msg = "hello";`
3.  `writeBuffer.put(msg.getBytes());`
4.  `writeBuffer.flip();`
5.  `channel.write(writeBuffer);`

*  读完写

1.  `ByteBuffer buffer = ByteBuffer.allocate(1024);`
2.  `int byteRead = channel.read(buffer);`
3.  `if(byteRead<=0){`
4.  `channel.close();`
5.  `break;`
6.  `}`
7.  `buffer.flip();`
8.  `channel.write(buffer);`
9.  `read += byteRead;`
10. `buffer.clear();`

> 每次写完buffer，如果buffer数据不需要再使用，建议clear清空buffer，准备下一次写操作

### Selector—多路复用器(选择器) ###

**多路复用器，这个名字很形象，使用一个线程去处理多个channel，从而管理多个channel**

> 为什么要使用一个线程管理多个channel？
> 
> 线程上下文切换开销很大，线程越少处理channel更高效

![9f61d36355db8b3b417db53f8eb27170.png][]

创建Selector—创建比赛

Selector selector = Selector.open();

注册channel—购买入场卷

channel通过注册到selector上来把channel的事件交给Selector管理，并且返回一个SelectionKey

SelectionKey key = channel.register(selector, SelectionKey.OP_READ);

*  与 Selector 一起使用时，Channel 必须处于`非阻塞模式`下。这意味着不能将 FileChannel 与 Selector 一起使用，因为 FileChannel 不能切换到非阻塞模式

channel.configureBlocking(false);

*  通过SelectionKey获取channel和selector以及准备好的事件

1.  `Channel channel = selectionKey.channel();`
2.  `Selector selector = selectionKey.selector();`

Selector执行选择—拿着入场卷入场

把channel注册到Selector后，我们可以使用`Selector.select();`方法获取准备就绪的通道，返回一个int型整数，表示准备好的channel数

通过`selector.selectedKeys();`方法获取准备就绪的SelectionKey，再通过SelectionKey获取channel和selector，一般使用迭代器遍历这些准备好的channel

> 在每一次处理完一个SelectionKey，必须把它从迭代器中删除，处理完，这个SelectionKey就没有用了，就像一个入场卷，你可以通过它进入赛场并且它上面有入场人和座位对应信息，比赛结束后你无法再通过它执行任何有效的操作。

*  看完比赛，举办者不会回收所有的票据，需要你们自己处理，不能乱丢在场地中，需要自己丢到垃圾桶中或者带回家

iterator.remove()；

*  wakeUp()方法

某个线程调用 select() 方法后阻塞了，即使没有通道已经就绪，也无法返回，wakeUp方法使得立马返回。

### Scatter、Gather ###

> scatter / gather 经常用于需要将传输的数据分开处理的场合，例如传输一个由消息头和消息体组成的消息，你可能会将消息体和消息头分散到不同的 buffer 中，这样你可以方便的处理消息头和消息体。

Scatter

分散（scatter）从 Channel 中读取是指在读操作时将读取的数据写入多个 buffer 中。因此，Channel 将从 Channel 中读取的数据 “分散（scatter）” 到多个 Buffer 中。

Gather

聚集（gather）写入 Channel 是指在写操作时将多个 buffer 的数据写入同一个 Channel，因此，Channel 将多个 Buffer 中的数据 “聚集（gather）” 后发送到 Channel。

例子：用三个长度分别为3,4,5的buffer存储输入的字符串，前3个字符存储在第一个buffer，4-7字符存储在第二个buffer，长度为4,8-12存储在第三个buffer，长度为5

1.  `ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();`
2.  `InetSocketAddress inetSocketAddress = new InetSocketAddress(8899);`
3.  `serverSocketChannel.socket().bind(inetSocketAddress);`
4.  `int messageLength = 3 + 4 + 5;`
5.  `ByteBuffer[] byteBuffer = new ByteBuffer[3];`
6.  `byteBuffer[0] = ByteBuffer.allocate(3);`
7.  `byteBuffer[1] = ByteBuffer.allocate(4);`
8.  `byteBuffer[2] = ByteBuffer.allocate(5);`
9.  `SocketChannel socketChannel = serverSocketChannel.accept();`
10. `while (true){`
11. `int byteRead = 0;`
12. `while (byteRead<messageLength){`
13. `long r = socketChannel.read(byteBuffer);`
14. `byteRead += r;`
15. `System.out.println("byteread:"+byteRead);`
16. `Arrays.stream(byteBuffer).map(o->"position:"+o.position()+",limit:"+o.limit()).forEach(System.out::println);`
17. `}`
18.  
19. `Arrays.stream(byteBuffer).forEach(Buffer::flip);`
20.  
21. `int byteWrite = 0;`
22. `while(byteWrite<messageLength){`
23. `long r = socketChannel.write(byteBuffer);`
24. `byteWrite += r;`
25. `System.out.println("bytewrite:"+byteWrite);`
26. `Arrays.stream(byteBuffer).map(o->"position:"+o.position()+",limit:"+o.limit()).forEach(System.out::println);`
27. `}`
28.  
29. `Arrays.stream(byteBuffer).forEach(Buffer::clear);`
30. `}`
31.  
32. `测试：使用linux nc localhost 8899测试`
33. `输入：helloworld回车`
34. `输出：`
35. `byteread:11`
36. `position:3,limit:3`
37. `position:4,limit:4`
38. `position:4,limit:5`
39. `解释：`
40. `回车算一个字符一共11个字符，前三个存储到第一个buffer了，存满了；中间四个存储到第二个buffer，存满了；剩下多余的存储到第三个buffer，没有存满`

### NIO服务端客户端 ###

这个程序演示使用NIO创建一个聊天室，服务端和多个客户端连接，客户端可以互发消息

*  server服务端

1.  `/**`
2.  `* 可以直接使用 linux nc命令当做客户端`
3.  `* nc localhost 端口`
4.  `*/`
5.  `public class Server {`
6.  `private static Map<SocketChannel,String> clientMap = new HashMap<>();`
7.  `public static void main(String[] args) throws IOException {`
8.  `//打开服务器channel`
9.  `ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();`
10. `//设置非阻塞 即将使用selector`
11. `serverSocketChannel.configureBlocking(false);`
12. `//获取服务器的socket`
13. `ServerSocket serverSocket = serverSocketChannel.socket();`
14. `//绑定端口`
15. `serverSocket.bind(new InetSocketAddress(8089));`
16. `//打开一个多路复用器，使用一条线程处理客户端channel`
17. `Selector selector = Selector.open();`
18. `//注册服务器channel到`
19. `serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);`
20. `while (true){`
21. `//阻塞获取channel事件`
22. `//一旦调用了 select() 方法，并且返回值表明有一个或更多个通道就绪了`
23. `int num = selector.select();`
24. `/**`
25. `* 获取到后 拿到多路复用器的SelectionKey 核心方法channel获取注册在起上的channel`
26. `* SelectionKey 每次注册一个channel都会创建一个SelectionKey 其中常量定义channel状态`
27. `**/`
28. `Set<SelectionKey> selectionKeys = selector.selectedKeys();`
29. `//对其中每一个SelectionKey进行操作`
30. `selectionKeys.forEach(selectionKey->{`
31. `try {`
32. `//如果该服务器SelectionKey被接收`
33. `if(selectionKey.isAcceptable()){`
34. `//拿到服务器channel`
35. `ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();`
36. `SocketChannel client = null;`
37. `//拿到本次连接上服务器的客户端`
38. `client = server.accept();`
39. `client.configureBlocking(false);`
40. `//把客户端注册到多路复用器，监听客户端的可读事件`
41. `client.register(selector,SelectionKey.OP_READ);`
42. `//为每个客户端分配id`
43. `String key = "["+ UUID.randomUUID()+"]";`
44. `clientMap.put(client,key);`
45. `//如果SelectionKey读就绪，执行读操作`
46. `}else if(selectionKey.isReadable()){`
47. `//拿到channel`
48. `SocketChannel channel = (SocketChannel) selectionKey.channel();`
49. `//创建读buffer`
50. `ByteBuffer readBuffer = ByteBuffer.allocate(1024);`
51. `//读取channel中数据到读buffer`
52. `int read = channel.read(readBuffer);`
53. `String reMsg = "";`
54. `//如果有数据`
55. `if(read>0){`
56. `//翻转进行写操作`
57. `readBuffer.flip();`
58. `//制定解码集utf-8，对读buffer解码打印`
59. `Charset charset = Charset.forName("utf-8");`
60. `reMsg = String.valueOf(charset.decode(readBuffer).array());`
61. `System.out.println(clientMap.get(channel)+" receive: "+reMsg);`
62. `}else if(read==-1) channel.close();//如果客户端关闭就关闭客户端channel`
63. `//群发：发送数据到其他客户端channel`
64. `for(SocketChannel ch:clientMap.keySet()){`
65. `if(ch!=channel) {`
66. `String key = clientMap.get(ch);`
67. `ByteBuffer writeBuffer = ByteBuffer.allocate(1024);`
68. `writeBuffer.put(("来自"+key + ":" + reMsg).getBytes());`
69. `writeBuffer.flip();`
70. `ch.write(writeBuffer);`
71. `}`
72. `}`
73. `}`
74.  
75. `} catch (IOException e) {`
76. `e.printStackTrace();`
77.  
78. `}`
79. `});`
80. `selectionKeys.clear();//每次处理完一个SelectionKey的事件，把该SelectionKey删除`
81. `}`
82. `}`
83. `}`

*  客户端

1.  `public class Client {`
2.  `public static void main(String[] args) throws IOException {`
3.  `//打开客户端channel`
4.  `SocketChannel socketChannel = SocketChannel.open();`
5.  `//设置为非阻塞模式，可以配合selector使用`
6.  `socketChannel.configureBlocking(false);`
7.  `//打开selector`
8.  `Selector selector = Selector.open();`
9.  `//注册客户端channel到多路复用器，监听连接事件`
10. `socketChannel.register(selector, SelectionKey.OP_CONNECT);`
11. `//连接到指定地址`
12. `socketChannel.connect(new InetSocketAddress("localhost",8089));`
13. `while (true){`
14. `try{`
15. `//执行selector方法，阻塞获取channel事件的触发`
16. `int num = selector.select();`
17. `//获取注册到多路复用器上的SelectionKey`
18. `Set<SelectionKey> selectionKeys = selector.selectedKeys();`
19. `//通过迭代器遍历SelectionKey`
20. `Iterator<SelectionKey> iterator = selectionKeys.iterator();`
21. `while (iterator.hasNext()) {`
22. `SelectionKey selectionKey = iterator.next();`
23. `//如果SelectionKey触发的事件为连接就绪`
24. `if(selectionKey.isConnectable()){`
25. `//拿到SelectionKey的客户端channel`
26. `SocketChannel client = (SocketChannel) selectionKey.channel();`
27. `if(client.isConnectionPending()){`
28. `//完成连接`
29. `client.finishConnect();`
30. `//新建一个写buffer`
31. `ByteBuffer writeBuffer = ByteBuffer.allocate(1024);`
32. `//写入客户端连接成功消息`
33. `writeBuffer.put((client.toString()+":连接成功!").getBytes());`
34. `//翻转读写操作 执行写操作`
35. `writeBuffer.flip();`
36. `//写入buffer数据刅客户端`
37. `client.write(writeBuffer);`
38. `//开辟一个线程写，因为标准输入是阻塞的，当前线程不能阻塞写`
39. `ExecutorService executorService = Executors.newSingleThreadExecutor();`
40. `executorService.submit(()->{`
41. `while (true){`
42. `writeBuffer.clear();`
43. `InputStreamReader reader = new InputStreamReader(System.in);`
44. `BufferedReader br = new BufferedReader(reader);`
45. `String msg = br.readLine();`
46. `//每次读入一行，写入数据到buffer并且写入客户端channel`
47. `writeBuffer.put(msg.getBytes());`
48. `writeBuffer.flip();`
49. `client.write(writeBuffer);`
50. `}`
51. `});`
52. `}`
53. `//注册客户端可读事件到多路复用器`
54. `client.register(selector,SelectionKey.OP_READ);`
55. `//如果多路复用器上的SelectionKey处于读就绪状态`
56. `}else if(selectionKey.isReadable()){`
57. `//拿到SelectionKey触发相应事件对应的客户端channel，执行读操作`
58. `SocketChannel client = (SocketChannel) selectionKey.channel();`
59. `//创建一个新的读buffer，`
60. `ByteBuffer readBuffer = ByteBuffer.allocate(1024);`
61. `//从准备好读操作的channel中读取数据`
62. `int count = client.read(readBuffer);`
63. `if (count>0){`
64. `//转码并数据使用String保存且打印`
65. `String reMsg = new String(readBuffer.array(),0,count);`
66. `System.out.println(reMsg);`
67. `}else if(count==-1) client.close();//关闭客户端`
68. `}`
69. `}`
70. `selectionKeys.clear();//每次处理完一个SelectionKey的事件，把该SelectionKey删除`
71. `}`
72. `catch (Exception e){`
73. `e.printStackTrace();`
74. `}`
75. `}`
76. `}`
77. `}`

*  测试

**1.创建一个服务端和三个客户端**

**2.客户端1,2,3分别发送数据**

服务端拿到连接信息和三个客户端发送信息

![31494b0360f652f475f0f32ba3ff2a7c.png][] 客户端1先创建，拿到2,3连接信息和2,3发送信息

![df097d1b27927a47efa76db7a67753c0.png][]

客户端2先于3创建，拿到3连接信息和1,3发送信息

![850b610ac8f5781ab2efd96120b3947b.png][]

客户端3最后创建，只能拿到1,2发送信息

![611af406190de56e161ad2237016676a.png][]

**3.此时再使用nc命令创建一个客户端**

发送信息，客户端可以收到

![5e735137271bd746b913ad93f8cbbcb5.png][] 客户端2发送信息，该终端客户端也可以收到

![2359ec3500e1a78e116946ada365a346.png][]

### NIO案例—跨端口传输数据—MultiServer ###

**实现目标：服务端监听两个端口，一个8089，一个8090,8089只有唯一的一个主客户端A连接，8090有多个客户端B连接，客户端A接收多个客户端B连接的发送的消息，实现跨端口的消息转发**

![ac6bd361166e29b132c82665193d82a8.png][]

*  **服务端**

我们先看服务端，服务端首先需要监听两个端口，我们创建两个服务端channel；服务端接收到连接后监听客户端B的发送数据事件(也就是客户端writable服务端readable事件)；拿到客户端B的消息后，把它发送到客户端A

> 服务端怎么发送数据到客户端A？
> 
> 保存一个客户端channel集合，为不同端口客户端分配不同的id的结尾部分，客户端A分配为wxq\]，客户端B分配为gzh\]，在他们channel创建的时候保存到HashMap中，channel作为key，id作为值保存

**下面说一下服务端流程：**

1.  创建两个服务端channel，绑定不同端口
2.  创建一个多路复用器selector，把两个服务端注册到selector上，并监听acceptable事件
3.  执行selector.select()方法，拿到SelectionKey集合，对不同事件做不停处理
    
    1.  如果事件为`接收就绪`，通过`SelectionKey.channel()`方法拿到服务端channel，根据端口不同注册不同的监听事件，如果是8090的，说明是客户端B的连接完成，拿到客户端B的channel，监听它的`可读事件`，并且分配id后缀为gzh\]并且保存；如果是8089端口的服务端channel，说明是客户端A的连接完成，客户端客户端A的channel，监听它的`可写事件`，并且分配id后缀为wxq\]，保存到hashmap
    2.  如果事件是`读就绪`，说明客户端B已经完后数据的写操作，可以读取客户端B的数据，执行读取；首先把数据读取并写入到`readBuffer`，使用`new String(readBuffer.array()`创建即将发送的msg，遍历客户端channel的key，如果后缀为wxq\]，说明是客户端A，则把数据写入writeBuffer中，并把数据写入客户端A的channel中
4.  每次SelectionKey的事件执行完毕，把该SelectionKey删除

**代码：**

1.  `public class Server {`
2.  `private static int CAPACITY = 1024;`
3.  `private static ByteBuffer readBuffer = ByteBuffer.allocate(CAPACITY);`
4.  `private static ByteBuffer writeBuffer = ByteBuffer.allocate(CAPACITY);`
5.  `private static Map<SocketChannel,String> clientMap = new HashMap<>();`
6.   
7.  `public static void main(String[] args) throws IOException {`
8.  `ServerSocketChannel serverSocketChannelWxq = ServerSocketChannel.open();`
9.  `ServerSocketChannel serverSocketChannelGzh = ServerSocketChannel.open();`
10. `serverSocketChannelGzh.configureBlocking(false);`
11. `serverSocketChannelWxq.configureBlocking(false);`
12. `ServerSocket serverSocketWxq = serverSocketChannelWxq.socket();`
13. `ServerSocket serverSocketGzh = serverSocketChannelGzh.socket();`
14. `serverSocketWxq.bind(new InetSocketAddress(8089));`
15. `System.out.println("监听8089：微信墙服务端口");`
16. `serverSocketGzh.bind(new InetSocketAddress(8090));`
17. `System.out.println("监听8090：公众号服务端口");`
18. `Selector selector = Selector.open();`
19. `serverSocketChannelWxq.register(selector, SelectionKey.OP_ACCEPT);`
20. `serverSocketChannelGzh.register(selector, SelectionKey.OP_ACCEPT);`
21. `while (true){`
22. `int num = selector.select();`
23. `Set<SelectionKey> selectionKeys = selector.selectedKeys();`
24. `selectionKeys.forEach(selectionKey->{`
25. `try {`
26. `if(selectionKey.isAcceptable()){`
27. `ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();`
28. `SocketChannel client = null;`
29. `client = server.accept();`
30. `client.configureBlocking(false);`
31. `String key = "";`
32. `if(server==serverSocketChannelGzh) {//如果是公众号server，注册其客户端的可读事件`
33. `client.register(selector, SelectionKey.OP_READ);`
34. `key = "["+ UUID.randomUUID()+":gzh]";`
35. `}else if(server==serverSocketChannelWxq){//如果是`
36. `client.register(selector,SelectionKey.OP_WRITE);`
37. `key = "["+ UUID.randomUUID()+":wxq]";`
38. `}`
39. `System.out.println(key+":连接成功!");`
40. `clientMap.put(client,key);`
41. `}else if(selectionKey.isReadable()){`
42. `SocketChannel channel = (SocketChannel) selectionKey.channel();`
43. `readBuffer.clear();`
44. `int read = 0;`
45. `while(true){`
46. `int byteRead = channel.read(readBuffer);`
47. `if(byteRead<=0){`
48. `break;`
49. `}`
50. `readBuffer.flip();`
51. `channel.write(readBuffer);`
52. `read += byteRead;`
53. `readBuffer.clear();`
54. `}`
55. `String reMsg = new String(readBuffer.array(),0,read);`
56. `System.out.println(clientMap.get(channel)+" send to wxq: "+reMsg);`
57. `//写入微信墙服务`
58. `for(SocketChannel ch:clientMap.keySet()){`
59. `if(ch!=channel) {`
60. `String key = clientMap.get(ch);`
61. `if(key.endsWith("wxq]")) {`
62. `writeBuffer.clear();`
63. `writeBuffer.put(("来自" + clientMap.get(channel) + ":" + reMsg).getBytes(StandardCharsets.UTF_8));`
64. `writeBuffer.flip();`
65. `ch.write(writeBuffer);`
66. `}`
67. `}`
68. `}`
69. `}`
70.  
71. `} catch (IOException e) {`
72. `e.printStackTrace();`
73. `}`
74. `});`
75. `selectionKeys.clear();//每次处理完一个SelectionKey的事件，把该SelectionKey删除`
76. `}`
77. `}`
78. `}`

**到此，服务端写完了，你就可以使用linux或者win下的`nc`命令连接到服务端，模拟客户端A和客户端B发送消息**

![8873895e7f62a8b7a17b01eddae31445.png][]

![7a8940979232b6eb3c32b4c458d0d33c.png][]

> 客户端发送消息后会会写一条是因为我在接收到消息后把消息写入客户端B的buffer中了

*  **客户端B—发送消息**

客户端B负责发送消息，主要事件就是负责写数据

流程：

1.  创建一个客户端channel`SocketChannel`，打开一个多留复用器selector，绑定`可连接事件`，连接到服务端监听的8090端口
2.  执行`selector.select()`方法，处理`连接就绪`和`写就绪两个事件`
    
    1.  如果事件为`连接就绪`，只需要拿到channel，执行`finishConnect`方法完成连接，并且注册监听事件为`可写事件`
    2.  如果事件为`写就绪`，执行写操作，使用标准输入从控制台读取输入并且写入writebuffer中，通过`channel.write()`方法写入数据到客户端
3.  清理事件的SelectionKey

**代码：**

1.  `public class GzhClient {`
2.  `public static void main(String[] args) throws IOException {`
3.  `SocketChannel socketChannel = SocketChannel.open();`
4.  `socketChannel.configureBlocking(false);`
5.  `Selector selector = Selector.open();`
6.  `socketChannel.register(selector, SelectionKey.OP_CONNECT);`
7.  `socketChannel.connect(new InetSocketAddress("localhost",8090));`
8.  `while (true){`
9.  `try{`
10. `int num = selector.select();`
11. `Set<SelectionKey> selectionKeys = selector.selectedKeys();`
12. `Iterator<SelectionKey> iterator = selectionKeys.iterator();`
13. `while (iterator.hasNext()) {`
14. `SelectionKey selectionKey = iterator.next();`
15. `if(selectionKey.isConnectable()){`
16. `SocketChannel client = (SocketChannel) selectionKey.channel();`
17. `if(client.isConnectionPending()){`
18. `client.finishConnect();`
19. `}`
20. `client.register(selector,SelectionKey.OP_WRITE);`
21. `}else if(selectionKey.isWritable()){`
22. `SocketChannel client = (SocketChannel) selectionKey.channel();`
23. `ByteBuffer writeBuffer = ByteBuffer.allocate(1024);`
24. `writeBuffer.clear();`
25. `InputStreamReader reader = new InputStreamReader(System.in);`
26. `BufferedReader br = new BufferedReader(reader);`
27. `String msg = br.readLine();`
28. `//每次读入一行，写入数据到buffer并且写入客户端channel`
29. `writeBuffer.put(msg.getBytes());`
30. `writeBuffer.flip();`
31. `client.write(writeBuffer);`
32. `}`
33. `}`
34. `selectionKeys.clear();//每次处理完一个SelectionKey的事件，把该SelectionKey删除`
35. `}`
36. `catch (Exception e){`
37. `e.printStackTrace();`
38. `}`
39. `}`
40. `}`
41. `}`
42.

*  **客户端A—接收服务端转发消息**

客户端A负责发送消息，主要事件就是负责读数据

流程：

1.  创建一个客户端channel`SocketChannel`，打开一个多留复用器selector，绑定`可连接事件`，连接到服务端监听的8089端口
2.  执行`selector.select()`方法，处理`连接就绪`和`读就绪两个事件`
    
    1.  如果事件为`连接就绪`，只需要拿到channel，执行`finishConnect`方法完成连接，并且注册监听事件为`可写事件`
    2.  如果事件为`读就绪`，执行读操作，把channel中数据使用`read()`方法读取到readBuffer中，通过`new String(readBuffer.array()`方法接收String类型数据，并且打印到控制台
3.  清理事件的SelectionKey

**代码：**

1.  `public class WxQClient {`
2.  `public static void main(String[] args) throws IOException {`
3.  `SocketChannel socketChannel = SocketChannel.open();`
4.  `socketChannel.configureBlocking(false);`
5.  `Selector selector = Selector.open();`
6.  `socketChannel.register(selector, SelectionKey.OP_CONNECT);`
7.  `socketChannel.connect(new InetSocketAddress("localhost",8089));`
8.  `while (true){`
9.  `try{`
10. `int num = selector.select();`
11. `Set<SelectionKey> selectionKeys = selector.selectedKeys();`
12. `Iterator<SelectionKey> iterator = selectionKeys.iterator();`
13. `while (iterator.hasNext()) {`
14. `SelectionKey selectionKey = iterator.next();`
15. `if(selectionKey.isConnectable()){`
16. `SocketChannel client = (SocketChannel) selectionKey.channel();`
17. `if(client.isConnectionPending()){`
18. `client.finishConnect();`
19. `}`
20. `client.register(selector,SelectionKey.OP_READ);`
21. `}else if(selectionKey.isReadable()){`
22. `SocketChannel client = (SocketChannel) selectionKey.channel();`
23. `ByteBuffer readBuffer = ByteBuffer.allocate(1024);`
24. `int count = client.read(readBuffer);`
25. `if (count>0){`
26. `String reMsg = new String(readBuffer.array(),0,count);`
27. `System.out.println(reMsg);`
28. `}else if(count==-1) client.close();//关闭客户端`
29. `}`
30. `}`
31. `selectionKeys.clear();//每次处理完一个SelectionKey的事件，把该SelectionKey删除`
32. `}`
33. `catch (Exception e){`
34. `e.printStackTrace();`
35. `}`
36. `}`
37. `}`
38. `}`

**至此，我们服务端和客户端AB都已经完后，现在我们测试一下**

1.  启动服务端，启动一个WxQClient也就是ClientA，启动两个GzhClient，也就是ClientB

服务端显示连接成功

![e931f1290b83562d7de026de60ff8fba.png][]

1.  客户端B发送消息

服务端接收到消息并打印，并转发到客户端A，客户端A打印消息

![7174c79cad67232f1672363a7ae3c3f7.png][]

[b7a5790209d76e451917b765e79bbcd5.png]: /images/20210923/15f7988df75042b1b36217b555cb7e7c.png
[9f61d36355db8b3b417db53f8eb27170.png]: /images/20210923/b661b242839f4f278d95d66bc1af4209.png
[31494b0360f652f475f0f32ba3ff2a7c.png]: /images/20210923/e13cc80fd089403895cd58bc040a3ca5.png
[df097d1b27927a47efa76db7a67753c0.png]: /images/20210923/45e1a1a6080346b5876e364b21db7a5c.png
[850b610ac8f5781ab2efd96120b3947b.png]: /images/20210923/99ff702817294dab9f8b631c8cec8088.png
[611af406190de56e161ad2237016676a.png]: /images/20210923/a9b9a5334bdf4c878209a622e88c0682.png
[5e735137271bd746b913ad93f8cbbcb5.png]: /images/20210923/8030cb9c4dc54fcba39df01f431ffc1c.png
[2359ec3500e1a78e116946ada365a346.png]: /images/20210923/43ffdca3237343478524de04c2de4a6b.png
[ac6bd361166e29b132c82665193d82a8.png]: /images/20210923/a58e8ae21d244cad904f82ea7678dd10.png
[8873895e7f62a8b7a17b01eddae31445.png]: /images/20210923/d5a21476bac34c6b8944420976be8712.png
[7a8940979232b6eb3c32b4c458d0d33c.png]: /images/20210923/c92fba0369ff4beb94c2c41700a7492a.png
[e931f1290b83562d7de026de60ff8fba.png]: /images/20210923/bf6b7931df364875a34b942ebfd17abc.png
[7174c79cad67232f1672363a7ae3c3f7.png]: /images/20210923/d3c014aab8074f98beb7f279de455709.png