Cet article présente en détail les grandes catégories qui composent la communication non bloquante : Buffer, Channel, Selector, SelectionKey
Le processus de communication non bloquante
ServerSocketChannel obtient ServerSocketChannel via la méthode ouverte, le définit en mode non bloquant via ServerSocketChannel, puis obtient le socket via ServerSocketChannel, liaison Déterminez le port d'écoute du processus de service. Le service a démarré avec succès.
Ensuite, il y a l'essence de la communication non bloquante. Selector obtient Selector via la méthode statique open(), puis ServerSocketChannel enregistre l'événement Selection.OP_ACCEPT dans Selector.
Selector surveillera l'occurrence des événements. Selector surveille le nombre d'objets SelectionKey qui se sont produits via select() et renvoie la collection correspondante d'objets selectionKey via la méthode selectKeys(). Parcourez la collection pour obtenir l'objet selectionKey correspondant, obtenez l'objet ServerSocketChannel associé via la méthode channel() de l'objet et obtenez l'objet Selector associé via la méthode selector().
Exécutez la méthode accept() via le ServerSocketChannel obtenu ci-dessus pour obtenir le SocketChannel, puis définissez le SocketChannel en mode non bloquant, enregistrez le SocketChannel dans le sélecteur créé ci-dessus et enregistrez le SelectionKey.OP_READ |SelectionKey.OP_WRITE événement .
Selector surveillera les événements correspondants liés ci-dessus et effectuera des opérations de lecture et d'écriture si les événements correspondants sont surveillés.
Exemple de code :
Ce qui précède décrit un processus de communication non bloquante côté serveur. Ce qui suit est implémenté via un code spécifique :
/**
* 非阻塞模式
*
*/public class EchoServer2 {
private Selector selector = null;
private ServerSocketChannel serverSocketChannel = null;
private int port = 8001;
private Charset charset = Charset.forName("UTF-8");
public EchoServer2() throws IOException {
selector = Selector.open();
serverSocketChannel = ServerSocketChannel.open(); //服务器重启的时候，重用端口
serverSocketChannel.socket().setReuseAddress(true); //设置非阻塞模式
serverSocketChannel.configureBlocking(false);
serverSocketChannel.socket().bind(new InetSocketAddress(port));
System.out.println("服务器启动成功");
} /**
* 服务方法
*/
public void service() throws IOException {
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
while (selector.select() > 0) {
Set readyKes = selector.selectedKeys();
Iterator it = readyKes.iterator();
while (it.hasNext()) {
SelectionKey key = null;
try {
key = (SelectionKey) it.next();
it.remove();
if (key.isAcceptable()) {
System.out.println("连接事件");
//连接事件
ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = ssc.accept();
System.out.println("接收到客户连接，来自：" + socketChannel.socket().getInetAddress() + " : " + socketChannel.socket().getPort());
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(1024);
socketChannel.register(selector, SelectionKey.OP_READ |
SelectionKey.OP_WRITE, buffer);
} else if (key.isReadable()) {
//接收数据
receive(key);
} else if (key.isWritable()) {
//发送数据
send(key);
}
} catch (IOException e) {
e.printStackTrace();
try {
if (key != null) {
key.cancel();
key.channel().close();
}
}catch (IOException ex){
ex.printStackTrace();
}
}
}
}
} private void send(SelectionKey key) throws IOException {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel channel = (SocketChannel) key.channel();
buffer.flip(); //把极限设置为位置，把位置设置为0
String data = decode(buffer);
if (data.indexOf("\r\n") == -1) {
return;
}
String outputData = data.substring(0, data.indexOf("\n") + 1);
System.out.println("请求数据：" + outputData);
ByteBuffer outputBuffer = encode("echo:" + outputData);
while (outputBuffer.hasRemaining()) {
channel.write(outputBuffer);
}
ByteBuffer temp = encode(outputData);
buffer.position(temp.limit());
buffer.compact(); if (outputData.equals("bye\r\n")) {
key.cancel();
channel.close();
System.out.println("关闭与客户的连接");
}
} private String decode(ByteBuffer buffer) {
CharBuffer charBuffer = charset.decode(buffer); return charBuffer.toString();
} private ByteBuffer encode(String s) { return charset.encode(s);
} private void receive(SelectionKey key) throws IOException {
ByteBuffer buffer = (ByteBuffer) key.attachment();
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer readBuff = ByteBuffer.allocate(32);
socketChannel.read(readBuff);
readBuff.flip();
buffer.limit(buffer.capacity());
buffer.put(readBuff);
} public static void main(String[] args) throws IOException { new EchoServer2().service();
}
}/**
* 创建非阻塞客户端
*
*/public class EchoClient2 {
private SocketChannel socketChannel; private int port = 8001; private Selector selector; private ByteBuffer sendBuffer = ByteBuffer.allocate(1024); private ByteBuffer receiveBuffer = ByteBuffer.allocate(1024); private Charset charset = Charset.forName("UTF-8"); public EchoClient2() throws IOException {
socketChannel = SocketChannel.open();
InetSocketAddress inetSocketAddress = new InetSocketAddress(InetAddress.getLocalHost(), port);
socketChannel.connect(inetSocketAddress);//
socketChannel.configureBlocking(false);//设置为非阻塞模式
System.out.println("与服务器连接成功");
selector = Selector.open();
} public static void main(String[] args) throws IOException { final EchoClient2 client = new EchoClient2();
Thread receiver = new Thread(new Runnable() { @Override
public void run() {
client.receiveFromUser();
}
});
receiver.start();
client.talk();
} private void receiveFromUser() { try {
System.out.println("请输入数据：");
BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));
String msg = null; while ((msg = localReader.readLine()) != null) {
System.out.println("用户输入的数据：" + msg); synchronized (sendBuffer) {
sendBuffer.put(encode(msg + "\r\n"));
} if (msg.equalsIgnoreCase("bye")) { break;
}
}
} catch (IOException e) {
e.printStackTrace();
}
} private ByteBuffer encode(String s) { return charset.encode(s);
} private void talk() throws IOException {
socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE); while (selector.select() > 0) {
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator(); while (it.hasNext()) {
SelectionKey key = null; try {
key = it.next();
it.remove(); if (key.isReadable()) { //System.out.println("读事件");
//读事件
receive(key);
} if (key.isWritable()) { // System.out.println("写事件");
//写事件
send(key);
}
} catch (IOException e) {
e.printStackTrace(); if (key != null) {
key.cancel();
key.channel().close();
}
}
}
}
} private void send(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel(); synchronized (sendBuffer) {
sendBuffer.flip();//把极限设为位置，把位置设为零
channel.write(sendBuffer);
sendBuffer.compact();//删除已经发送的数据。
}
} private void receive(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
channel.read(receiveBuffer);
receiveBuffer.flip();//将limit的值设置为position的值，将position的值设置为0
String receiveData = decode(receiveBuffer); if (receiveData.indexOf("\n") == -1) { return;
}
String outputData = receiveData.substring(0, receiveData.indexOf("\n") + 1);
System.out.println("响应数据：" + outputData); if (outputData.equalsIgnoreCase("echo:bye\r\n")) {
key.cancel();
socketChannel.close();
;
System.out.println("关闭与服务器的连接");
selector.close();
System.exit(0);
}
ByteBuffer temp = encode(outputData);
receiveBuffer.position(temp.limit());
receiveBuffer.compact();//删除已经打印的数据
} private String decode(ByteBuffer receiveBuffer) {
CharBuffer buffer = charset.decode(receiveBuffer); return buffer.toString();
}
}Copier après la connexion
Moyens d'obtenir une communication non bloquante
Buffer
Canal
Sélecteur
Buffer
Fonction : Réduire le nombre de lectures et d'écritures physiques, et réduire le nombre de créations et de destructions de mémoire. Attributs du tampon : capacité (capacité maximale), limite (capacité réelle), position (position actuelle). PS : Les autres lieux sont traduits en capacité (capacité), limite (limite), position (position). Je pense personnellement qu'il est préférable de le comprendre lorsqu'il est traduit par ce qui précède. utilisez la méthode suivante. Vous pouvez la comprendre grâce à des analyses et des diagrammes. Bien entendu, il est préférable de l’exprimer plus clairement en anglais.
La relation entre les trois attributs est : capacité≥limite≥position≥0
La relation graphique est la suivante :
Structure de la classe tampon :
java.nio Classe .ByteBuffer C'est une classe abstraite et ne peut pas être instanciée. Cependant, 8 classes d'implémentation spécifiques sont fournies, dont la mémoire tampon la plus basique est ByteBuffer, qui stocke les unités de données en octets.
Méthode courante :
clear() : définissez la limite sur la capacité, puis définissez la position sur 0
flip() : définissez la limite sur la position, puis définissez la position sur 0.
rewind() : Ne modifiez pas la limite et définissez la position sur 0.
allocate() : Crée un tampon, le paramètre de méthode spécifie la taille du tampon
compact() : copie les octets (le cas échéant) entre la position actuelle du tampon et la limite jusqu'au début du tampon.
Testez la méthode ci-dessus :
Testez la méthode clear()
@Test
public void testClear() { //创建一个10chars大小的缓冲区，默认情况下limit和capacity是相等的
CharBuffer buffer = CharBuffer.allocate(10);
System.out.println("创建默认情况");
printBufferInfo(buffer);
buffer.limit(8);//修改limit的值
System.out.println("修改limit后");
printBufferInfo(buffer); // clear()：把limit设置为capacity，再把位置设为0
buffer.clear();
System.out.println("执行clear()方法后");
printBufferInfo(buffer);
}Copier après la connexion
Les résultats de l'exécution sont les suivants :
Tester la méthode flip() :
@Test
public void testFlip() {
CharBuffer buffer = CharBuffer.allocate(10);
System.out.println("创建默认情况");
printBufferInfo(buffer); //put的方法会修改position的值
buffer.put('H');
buffer.put('E');
buffer.put('L');
buffer.put('L');
buffer.put('O');
System.out.println("调用put方法后：");
printBufferInfo(buffer); //flip()：把limit设置为position，再把位置设置为0。
buffer.flip();
System.out.println("调用flip方法后：");
printBufferInfo(buffer);
}Copier après la connexion
Les résultats d'exécution sont les suivants :
Tester la méthode rewind()
@Test
public void testRewind() {
CharBuffer buffer = CharBuffer.allocate(10);
System.out.println("创建默认情况");
printBufferInfo(buffer); //put的方法会修改position的值
buffer.put('H');
buffer.put('E');
buffer.put('L');
buffer.put('L');
buffer.put('O');
buffer.limit(8);
System.out.println("调用put、limit方法后：");
printBufferInfo(buffer); //rewind()：不改变limit，把位置设为0。
buffer.rewind();
System.out.println("调用rewind方法后：");
printBufferInfo(buffer);
}Copier après la connexion
Les résultats de l'exécution sont les suivants :
Test de la méthode compact()
@Test
public void testCompact(){
CharBuffer buffer = CharBuffer.allocate(10);
System.out.println("创建默认情况");
printBufferInfo(buffer); //put的方法会修改position的值
buffer.put('H');
buffer.put('E');
buffer.put('L');
buffer.put('L');
buffer.put('O');
buffer.limit(8);//修改limit的值
System.out.println("调用put和limit方法后：");
printBufferInfo(buffer);
System.out.println("调用compact方法后："); //将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。
buffer.compact();
printBufferInfo(buffer);
}Copier après la connexion
Voici ce que fait cette méthode telle qu'elle est introduite dans le JDK :
Copiez les octets (le cas échéant) entre les la position actuelle du tampon et la limite jusqu'au début du tampon. Autrement dit, copiez l'octet à l'index p = position() vers l'index 0, copiez l'octet à l'index p 1 vers l'index 1, et ainsi de suite jusqu'à ce que l'octet à l'index limit() - 1 soit copié dans l'index n = limit() -1-p. Définissez ensuite la position du tampon sur n 1 et ses limites sur sa capacité. Si la balise est déjà définie, elle est ignorée.
Le communiqué officiel est trop difficile à comprendre :
将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。并将limit(实际容量)设置为 capacity(最大容量)。执行compact()方法前，limit的值是：8，position的值是：5。按照上面描述的执行完compact()后，position的值计算方式是：n+1;n=limit-1-p;所有n=8-1-5=2,最后position的值为：2+1=3。和程序运行的结果一致。
可以在这种情况：从缓冲区写入数据之后调用此方法，以防写入不完整。
buf.clear(); // Prepare buffer for use
while (in.read(buf) >= 0 || buf.position != 0) {
buf.flip();
out.write(buf);
buf.compact(); // In case of partial write
}Copier après la connexion
如果out.write()方法没有将缓存中的数据读取完，这个时候的position位置指向的是剩余数据的位置。达到防止写入不完整。
通道
作用： 连接缓冲区与数据源或数据目的地。
常用类：
Channel
接口有下面两个子接口ReadableByteChannel和WritableByteChannel和一个抽象实现类SelectableChannel。
在ReadableByteChannel接口中申明了read(ByteBuffer
dst)方法。在WritableByteChannel接口中申明了write(ByteBuffer[]
srcs)：方法。SelectableChannel抽象类中主要方法，configureBlocking(boolean
block)、register();方法。 ByteChannel
接口继承了ReadableChannel和WritableChannel。所以ByteChannel具有读和写的功能。
ServerSocketChannel继承了SelectableChannel类抽象类，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件功能。
SocketChannel也继承了SelectableChannel类还实现ByteChannel接口，所以SocketChannel具有设置是否是阻塞模式、向selector注册事件、从缓冲区读写数据的功能。
通过类图展现：
Selector类：
作用：只要ServerSocketChannel及SocketChannel向Selector注册了特定的事件，Selector就会监听这些事件的发生。
流程：
Selector通过静态的open()方法创建一个Selector对象，SelectableChannel类向Selector注册了特定的事件。Selector就会监控这些事件发生，Selector通过select()监控已发生的SelectionKey对象的数目，通过selectKeys()方法返回对应的selectionKey对象集合。遍历该集合得到相应的selectionKey对象，通过该对象的channel()方法获取关联的SelectableChannel对象，
通过selector()方法就可以获取关联的Selector对象。
Note：
当Selector的select()方法还有一个重载方式:select(long timeout)。并且该方法采用阻塞的工作方式，如果相关事件的selectionKey对象的数目一个也没有，就进入阻塞状态。知道出现以下情况之一，才从select()方法中返回。
至少有一个SelectionKey的相关事件已经发生。
其他线程调用了Selector的wakeup()方法，导致执行select()方法的线程立即返回。
当前执行的select()方法的线程被中断。
超出了等待时间。仅限调用select(long timeout)方法时出现。如果没有设置超时时间，则永远不会超时。
Selector类有两个非常重要的方法： 静态方法open()，这是Selector的静态工厂方法，创建一个Selector对象。
selectedKeys()方法返回被Selector捕获的SelectionKey的集合。
SelectionKey类
作用：
ServerSocketChannel或SocketChannel通过register()方法向Selector注册事件时，register()方法会创建一个SelectionKey对象，该对象是用来跟踪注册事件的句柄。在SelectionKey对象的有效期间，Selector会一直监控与SelectionKey对象相关的事件，如果事件发生，就会把SelectionKey对象添加到Selected-keys集合中。
SelectionKey中定义的事件： 定义了4种事件：
1、SelectionKey.OP_ACCEPT：接收连接就绪事件，表示服务器监听到了客户连接，服务器可以接收这个连接了。常量值为16.
2、SelectionKey.OP_CONNECT：连接就绪事件，表示客户与服务器的连接已经建立成功。常量值为8.
3、SelectionKey.OP_READ：读就绪事件，表示通道中已经有了可读数据可以执行读操作。常量值为1.
4、SelectionKey.OP_WRITE：写就绪事件，表示已经可以向通道写数据了。常量值为4.
Méthodes courantes :
Méthode Channel() : renvoie le SelectedChannel qui lui est associé (y compris ServerSocketChannel et SocketChannel).
Méthode selector() : renvoie l'objet Selector qui lui est associé.
La relation entre eux est la suivante :
Ce qui précède est l'explication détaillée des exemples de code graphique et textuel de communication non bloquante dans la programmation réseau Java, du superficiel au profond. Plus de contenu connexe Veuillez faire attention au site Web PHP chinois (www.php.cn) !
![[Web front-end] Démarrage rapide de Node.js](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
