14.5 基于TCP协议的网络编程2——非阻塞的网络编程
Posted weststar
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了14.5 基于TCP协议的网络编程2——非阻塞的网络编程相关的知识，希望对你有一定的参考价值。
一、使用NIO实现非阻塞Socket通信
可以参考另一篇博客：https://blog.csdn.net/weixin_42762133/article/details/100040141
讲得很细致
1.1 前言
从JDK1.4以来，Java提供了NIO API来开发高性能的网络服务，但是在JDK1.4之前，网络通信程序是基于阻塞式API的——即当程序执行输入，输出操作后，在这些操作返回之前会一直阻塞该线程，所以服务器必须为每个客户端都提供一个独立线程进行处理，当服务器端需要同时处理大量客户端时，这种做法会导致性能下降。使用NIO API则可以让服务器端使用一个或有限几个线程来同时处理链接到服务器端的所有程序
1.2 非阻塞式Socket通信的几个特殊类
1.2.1 Selector类
1.Selector:它是SelectableChannel对象的多路复用器，所有希望采用非阻塞方式进行通信的Channel都应该注册到Selector对象。可以通过调用此类的open()静态方法来创建Selector实例，该方法将使用系统默认的Selector来返回新的Selector.
Selector可以同时监控多个SelectableChannel的IO状态，是非阻塞IO的核心。一个Selector实例有三个SelectionKey集合。
1).所有的SelectionKey集合：代表了注册在该Selector上的Channel,这个集合可以通过keys()方法返回。
2).被选择的SelectionKey集合：代表了所有可通过select()方法获取的，需要进行IO处理的Channel，这个集合可以通过selectedKeys()返回。
3).被取消的SelectionKey集合：代表了所有被取消注册关系的Channel,在下一次执行select()方法时，这些Channel对应的SelectionKey会被彻底解除，程序通常无需直接访问该集合。
2、Selector还提供了一系列和select()相关的方法，如下所示。
1).int select():监控所有注册的Channel,当他们中间有需要处理的IO操作时，该方法返回，并将对应的SelectionKey加入到被选择的SelectionKey集合中，该方法返回这些Channel的数量。
2).int select(long timeout):可以设置超时时长的select()操作。
3).int selectNow():执行一个立即返回的select()操作，相对于无参数的select()方法而言，该方法不会阻塞线程。
4).Selector wakeup():使一个还未返回的select()方法立即返回。
1.2.2 SelectableChannel类
SelectableChannel:它代表可以支持非阻塞IO操作的Channel对象，它可被注册到Selector上，这种注册关系由SelectionKey实例表示。Selector对象提供了一个select()方法，该方法允许应用程序同时监控多个IO Channel。
应用程序可调用SelectableChannel的register()方法将其注册到指定Selector上，当该Selector上的某些SelectableChannel上有需要处理的IO操作时，程序可以调用Selector实例的select()方法获取它们的数量，并可以通过selectredKeys()方法返回它们对应的SelectionKey集合——通过该集合就可以获取所有需要进行IO处理的SelectableChannel集。
SelectableChannel对象支持阻塞和非阻塞两种模式(所有的Channel默认都是阻塞模式)，必须使用非阻塞模式才可以利用非阻塞IO操作
SelectableChannel提供了如下两个方法来设置和返回该Channel的模式状态。
1). SelectableChannel configureBloking(boolean blok):设置是否采用阻塞模式。
2). boolean isBlocking():返回该Channel是否是阻塞模式。
不同的SelectableChannel所支持的操作不一样。例如ServerSocketChannel代表一个ServerSocket，它就只支持OP_ACCEPT操作。SelectableChannel提供了如下方法返回它支持的所有操作。
3). int validOps():返回一个整数值，表示这个Channel所支持的IO操作。
提示：在SelectionKey中，用静态常量定义了4中IO操作：OP_READ(1)、OP_WRITE(4)、OP_CONNECT(8)、OP_ACCEPT(16),这里任意2个、3个、4个进行按位或的结果和相加的结果相等，而且任意2个、3个、4个相加的结果总是互不相同，所以系统可以根据validOps()方法返回值确定该SelectableChannel支持的操作，例如返回5，即可知道它支持（1）和（4）.
除此之外，SelectableChannel还提供了如下几个方法来获取它的注册状态。
1).boolean isRegistered():返回该Channel是否已注册在一个或多个Selector上。
2).SelectionKey keyFor(Selector sel):返回该Channel和sel Selector之间的注册关系，如果不存在注册关系，则返回null.
1.2.3 SelectorKey、ServerSocketChannel和SocketChannel
1、SelectorKey:该对象代表SelectableChannel和Selector之间的注册关系
2、ServerSocketChannel:支持非阻塞操作，对应于java.net.ServerSocket这个类，只支持OP_ACCEPT操作。该类也提供了accept()方法，功能相当于ServerSocket提供的accept()方法。
3、SocketChannel：支持非阻塞操作，对应于java.net.Socket这个类，支持OP_CONNECT,OP_READ和OP_WRITE操作。这个类还实现了ByteChannel接口，ScatteringByteChannel接口和GatheringByteChannel接口，所以可以直接通过SocketChannel来读写ByteBuffer对象。
1.3 非阻塞式服务器示意图
从图中可以看出，服务器上的所有Channel(包括ServerSocketChannel和SocketChannel)都需要向Selector注册，而该Selector则负责监视这些Socket的IO状态，当其中任意一个或多个Channel具有可用的IO操作时，该Selector的select()方法将会返回大于0的整数，该整数值就表示该Selector上有多少个Channel具有可用的IO操作，并提供了selectedKeys()方法来返回这些Channel对应的SelectionKey集合。正是通过Selector,使得服务器端只需要不断地调用Selector实例的select()方法，即可知道当前的所有Channel是否有需要处理的IO操作。
提示：当Selector上注册的所有Channel都没有需要处理的IO操作时，select()方法将会阻塞，调用该方法的线程被阻塞。
本示例程序使用NIO实现了多人聊天室功能，服务器使用循环不断地获取Selector的select()方法的返回值，当该返回值大于0时就处理该Selector上被选择的SelectionKey所对应的Channel。
服务器需要使用ServerSocket Channel来监听客户端的连接请求，Java对该类的设计比较难用：它不像ServerSocket可以直接指定监听某个端口；而且不能使用已有的ServerSocket的getChannel()来获取ServerSocket Channel实例。程序必须先调用它的open()静态方法返回一个返回一个ServerSocketChannel实例，再使用它的bind()方法指定它在某个端口监听。创建一个可用的ServerSocketChannel需要采用如下代码片段：
//通过一个open()方法打开一个未绑定的ServerSocketChannel实例
ServerSocketChannel server=ServerSocketChannel.open();
var isa=new InetSocketAddress("127.0.0.1",30000);
//将该ServerSocketChannel绑定到指定IP地址
server.bind(isa);
如果需要使用非阻塞方式来处理该ServerSocketChannel,还应该设置它的非阻塞模式，并将其注册到指定的Selector。代码片段如下所示：
//设置ServerSocket以非阻塞方式工作
server.configureBlocking(false);
//将server注册到指定的Selector对象
server.register(selector,SelectionKey.OP_ACCEPT);
经过上面的步骤，该ServerSocketChannel可以接受客户端的连接请求，还需要调用Selector的select()方法来监听所有Channel上的IO操作。
package NIO_Socket;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.nio.charset.Charset;
public class NServer
{
//用于检测所有channel状态的Selector——SelectableChannel对象的多路复用器
private Selector selector=null;
static final int PORT=30000;
//定义编码、解码的字符集对象
private Charset charset=Charset.forName("utf-8");
public void init() throws IOException
{
//通过Selector类的open()静态方法来创建Selector实例
selector=Selector.open();
//通过open方法打开一个未绑定的ServerSocketChannel实例
ServerSocketChannel server=ServerSocketChannel.open();
var isa=new InetSocketAddress("127.0.0.1",PORT);
//将ServerSocketChannel绑定到指定IP地址
server.bind(isa);
//设置ServerSocketChannel以非阻塞式工作
server.configureBlocking(false);
//将server注册到指定的Selector对象
server.register(selector, SelectionKey.OP_ACCEPT);
while(selector.select()>0)
{
//依次处理selector上每个已选择的selectionKey
for(SelectionKey sk:selector.selectedKeys())
{
//从selector上的已选择的SelectionKey集合中删除正在处理的SelectionKey
selector.selectedKeys().remove(sk);//代码1
//如果sk对应的Channel包含客户端的连接请求
if(sk.isAcceptable())//代码2
{
//调用accept()方法接受连接，产生服务器端的SocketChannel
SocketChannel sc=server.accept();
//设置非阻塞模式
sc.configureBlocking(false);
//也将该SocketChannel注册到selector
sc.register(selector,SelectionKey.OP_READ);
//将sk对应的channel设置成准备接受其它请求
sk.interestOps(SelectionKey.OP_ACCEPT);
}
//如果sk对应的对应的Channel有数据需要读取
if(sk.isReadable())//代码3
{
//获取该SelectionKey对应的Channel，该Channel中有可读的数据
var sc=(SocketChannel)sk.channel();
//定义准备执行读取数据的ByteBuffer
ByteBuffer buff=ByteBuffer.allocate(1024);
var content="";
//开始读取数据
try
{
while(sc.read(buff)>0)
{
buff.flip();
content+=charset.decode(buff);
}
//打印从该sk对应的Channel里读取的数据
System.out.println("读取的数据："+content);
//将sk对应的的Channel设置成准备下一次读取
sk.interestOps(SelectionKey.OP_READ);
}
//如果捕获到sk对应的Channel出现了异常，即表明该Channel
//对应的Client出现了问题，所以Selector中取消sk的注册
catch (IOException ex)
{
//从Selector中指定删除SelectionKey
sk.cancel();
if(sk.channel()!=null)
{
sk.channel().close();
}
}
//如果content的长度大于0，即聊天信息不为空
if(content.length()>0)
{
//遍历该Selector里注册的所有SelectionKey
for(SelectionKey key:selector.selectedKeys())
{
//获取该key对应的Channel
Channel targetChannel=key.channel();
//如果该Channel是SocketChannel对象
if(targetChannel instanceof SocketChannel)
{
//将读到的内容写入该Channel中
var dest=(SocketChannel)targetChannel;
dest.write(charset.encode(content));
}
}
}
}
}
}
}
public static void main(String[] args) throws IOException
{
new NServer().init();
}
}
上面程序启动了时创建了一个可监听请求的ServerSocketChannel,并将该Channel注册到指定的selector，接着程序直接采用循环方式不断地监控Selector对象的select()方法返回值，当该返回值大于0时，处理该Selector上所有被选择的SelectionKey。
开始处理指定的SelectionKey之后，立即从该Selector上被选择的SelectionKey集合中删除该SelectionKey,如程序中代码1处所示。
服务器端的Selector仅需要监听两种操作：连接和读数据，所以程序分别处理这两种操作，如程序代码2、代码3所示——处理连接操作时，系统只需要将连接完成后产生的SocketChannel注册到指定的Selector对象即可；处理读数据操作时，系统先从该Socket中读取数据，再将数据写入Selector上注册的所有Channel中。
提示：使用NIO来实现服务器端时，无须使用List来保存服务器端所有SocketChannel,因为所有SocketChannel都已注册到指定的Selector对象。除此之外，当客户端关闭时会导致服务器端对应的Channel也抛出异常，而且本程序只有一个线程，因为该异常得不到处理将会导致整个服务器端退出，所以程序捕获了这种异常，并在处理异常时从从Selector删除Channel的注册。
本示例程序的客户端需要两个线程，一个线程负责读取用户键盘的输入，并将输入内容写入SocketChannel;另一个线程则不断查询Selector对象的select()方法，如果方法的返回值大于0，那就说明程序需要对相应的Channel执行IO处理。
package NIO_Socket;
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.nio.charset.Charset;
import java.util.Scanner;
public class NClient
{
//定义检测SocketChannel的Selector对象
private Selector selector=null;
static final int PORT=30000;
//定义处理编码和解码的字符集
private Charset charset= Charset.forName("utf-8");
//客户端SocketChannel
private SocketChannel sc=null;
public void init() throws IOException
{
selector=Selector.open();
var isa=new InetSocketAddress("127.0.0.1",PORT);
//调用open()静态方法创建连接到指定主机SocketChannel
sc=SocketChannel.open(isa);
//设置该sc以非阻塞式工作
sc.configureBlocking(false);
//将SocketChannel对象注册到指定的Selector
sc.register(selector, SelectionKey.OP_READ);
//启动读取服务器端数据的线程
new ClientThread().start();
//创建键盘输入流
var scan=new Scanner(System.in);
while(scan.hasNextLine())
{
//读取键盘输入
String line=scan.nextLine();
//将键盘输入的内容输出到SocketChannel中
sc.write(charset.encode(line));
}
}
//定义读取服务器端数据的线程
private class ClientThread extends Thread
{
public void run()
{
try
{
while(selector.select()>0)//代码1
{
//遍历每个可用IO操作的channel对应的SelectionKey
for(SelectionKey sk:selector.selectedKeys())
{
//删除正在处理的SelectionKey
selector.selectedKeys().remove(sk);
//如果SelectionKey对应的Channel中有可读的数据
if(sk.isReadable())
{
//使用NIO读取Channel中的数据
SocketChannel sc=(SocketChannel)sk.channel();
ByteBuffer buffer=ByteBuffer.allocate(1024);
String content="";
while(sc.read(buffer)>0)
{
buffer.flip();
content+=charset.decode(buffer);
}
//打印读取的内容
System.out.println("聊天信息："+content);
//为下一次读取做好准备
sk.interestOps(SelectionKey.OP_READ);
}
}
}
}
catch (IOException e)
{
e.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException
{
new NClient().init();
}
}
运行结果：
相比之下客户端的程序要简单多了，客户端只有一个SocketChannel，将SocketChannel注册到指定的Selector后，程序启动另外一个线程来监听该Selector即可。如果程序监听到Selector的select()方法返回值大于0，表明该Selecotr上有需要进行IO处理的Channel，接着程序取出该Channel，并使用NIO读取该Channel中的数据。
以上是关于14.5 基于TCP协议的网络编程2——非阻塞的网络编程的主要内容，如果未能解决你的问题，请参考以下文章