Java-Netzwerkprogrammierung von einfach bis fortgeschritten 3. Erfahren Sie mehr über nicht blockierende Kommunikation mit grafischen Codebeispielen und ausführlichen Erklärungen

In diesem Artikel werden die Hauptkategorien der nicht blockierenden Kommunikation ausführlich vorgestellt: Puffer, Kanal, Selektor, Auswahlschlüssel

Der Prozess der nicht blockierenden Kommunikation

1. ServerSocketChannel ruft ServerSocketChannel über die offene Methode ab und versetzt ihn in den nicht blockierenden Modus ServerSocketChannel und ruft dann den Socket über ServerSocketChannel ab und bindet ihn. Bestimmen Sie den Überwachungsport des Dienstprozesses. Der Dienst wurde erfolgreich gestartet.

2. Dann gibt es die Essenz der nicht blockierenden Kommunikation. Selector erhält Selector über die statische open()-Methode, und dann registriert ServerSocketChannel das Selection.OP_ACCEPT-Ereignis bei Selector.

3. Selector überwacht das Auftreten von Ereignissen, die über select() aufgetreten sind, und gibt die entsprechende Sammlung von SelectionKey-Objekten über die Methode selectKeys() zurück. Durchlaufen Sie die Sammlung, um das entsprechende SelectionKey-Objekt abzurufen, das zugehörige ServerSocketChannel-Objekt über die Methode „channel()“ des Objekts abzurufen und das zugehörige Selector-Objekt über die Methode „selector()“ abzurufen.

4. Führen Sie die Methode „accept()“ über den oben erhaltenen ServerSocketChannel aus, um den SocketChannel zu erhalten, setzen Sie dann den SocketChannel in den nicht blockierenden Modus, registrieren Sie den SocketChannel beim oben erstellten Selektor und registrieren Sie den SelectionKey.OP_READ |SelectionKey.OP_WRITEEreignis .

5. Der Selektor überwacht die entsprechenden oben gebundenen Ereignisse und führt Lese- und Schreibvorgänge aus, wenn die entsprechenden Ereignisse überwacht werden.

Beispielcode:

Das Obige beschreibt einen Prozess der serverseitigen nicht blockierenden Kommunikation. Das Folgende wird durch spezifischen Code implementiert:

/**
 * 非阻塞模式
 * 
 */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();
    }
}

Wege zur nicht blockierenden Kommunikation

• Puffer

• Kanal

• Selektor

  Puffer

Funktion: Reduzieren Sie die Anzahl der physischen Lese- und Schreibvorgänge und Reduzieren Sie die Anzahl der Speichererstellung und -zerstörung. Pufferattribute: Kapazität (maximale Kapazität), Limit (tatsächliche Kapazität), Position (aktuelle Position). PS: Andere Orte werden in Kapazität (Kapazität), Grenze (Limit) und Position (Position) übersetzt Ich verwende die folgende Methode? Sie können es durch Analyse und Diagramme verstehen. Natürlich ist es am besten, es auf Englisch am deutlichsten auszudrücken.
Die Beziehung zwischen den drei Attributen ist: Kapazität≥Limit≥Position≥0

Die grafische Beziehung ist wie folgt:

Pufferklassenstruktur:
java.nio .ByteBuffer-Klasse Es ist eine abstrakte Klasse und kann nicht instanziiert werden. Es werden jedoch 8 spezifische Implementierungsklassen bereitgestellt, von denen der grundlegendste Puffer ByteBuffer ist, der Dateneinheiten in Bytes speichert.

Übliche Methode:

clear(): Limit auf Kapazität setzen und dann die Position auf 0 setzen
flip(): Limit auf Position setzen und dann Position auf 0 setzen.
rewind(): Das Limit nicht ändern und die Position auf 0 setzen.
allocate(): Erstellen Sie einen Puffer. Der Methodenparameter gibt die Puffergröße
an compact(): Kopiert die Bytes (falls vorhanden) zwischen der aktuellen Position des Puffers und dem Limit an den Anfang des Puffers.

Testen Sie die obige Methode:
Testen Sie die Methode „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);
    }

Die Ausführungsergebnisse sind wie folgt:

Testen Sie die flip()-Methode:

    @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);
    }

Die Ausführungsergebnisse sind wie folgt:

Testen Sie die rewind()-Methode

        @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);
    }

Die Ausführungsergebnisse sind wie folgt:

Testen Sie die compact()-Methode

    @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);
    }

Das macht diese Methode, wie sie im JDK eingeführt wurde:

Kopieren Sie die Bytes (falls vorhanden) zwischen den Die aktuelle Position des Puffers und die Grenze zum Anfang des Puffers. Das heißt, kopieren Sie das Byte bei Index p = position() auf Index 0, kopieren Sie das Byte bei Index p + 1 auf Index 1 usw., bis das Byte bei Index limit() - 1 auf Index n = limit( kopiert wird. ) - 1 - S. Setzen Sie dann die Position des Puffers auf n+1 und seine Grenzen auf seine Kapazität. Wenn das Tag bereits definiert ist, wird es verworfen.

Die offizielle Aussage ist zu schwer zu verstehen:

将缓冲区的当前位置和界限之间的字节（如果有）复制到缓冲区的开始处。并将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
 }

如果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.

Gemeinsame Methoden:
Methode „channel()“: Gibt den damit verbundenen SelectedChannel zurück (einschließlich ServerSocketChannel und SocketChannel).
Methode selector(): Gibt das damit verbundene Selector-Objekt zurück.
Die Beziehung zwischen ihnen ist wie folgt:

Das Obige ist eine detaillierte Erläuterung der Grafik- und Textcodebeispiele der nicht blockierenden Kommunikation in der Java-Netzwerkprogrammierung von flach bis tief. Weitere verwandte Inhalte Bitte beachten Sie die chinesische PHP-Website (www.php.cn)!