zoukankan      html  css  js  c++  java
  • JAVA I/O(六)多路复用IO

    在前边介绍Socket和ServerSocket连接交互的过程中,读写都是阻塞的。套接字写数据时,数据先写入操作系统的缓存中,形成TCP或UDP的负载,作为套接字传输到目标端,当缓存大小不足时,线程会阻塞。套接字读数据时,如果操作系统缓存没有接收到信息,则读线程阻塞。线程阻塞情况下,就不能处理其他事情。JDK1.4引入了通道和选择器的概念,以支持异步或多路复用的IO。

    Unix系统中的select()方法可以实现异步IO,可以给该Selector注册多个描述符(可读或可写),然后对这些描述符进行监控。在Java中,描述符即为套接字Socket。

    JAVA I/O(二)文件NIO中对选择器的介绍,在非阻塞模式下,用select()方法检测发生变化的通道,每个通道都关联一个Socket,用一个线程实现多个客户端的请求,从而实现多路复用。

     

    1. 简单实例

    服务器端

    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.CharBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.nio.charset.Charset;
    import java.util.Iterator;
    
    public class MultiJabberServer1 {
    
        public static final int PORT = 8080;
        
        public static void main(String[] args) throws IOException{
            
            String encoding = System.getProperty("file.encoding");
            Charset cs = Charset.forName(encoding);
            ByteBuffer buffer = ByteBuffer.allocate(16);
            SocketChannel ch = null;//Socket对应的channel
            //1.创建ServerSocketChannel
            ServerSocketChannel ssc = ServerSocketChannel.open();
            //2.创建选择器Selector
            Selector sel = Selector.open();
            
            try {
                //3.设置ServerSocketChannel通道为非阻塞
                ssc.configureBlocking(false);
                //4.ServerSocketChannel关联Socket,用于监听连接,使用本地ip和port
                //注意:Socket也对通道进行了改造,直接调Socket.getChannel()将返回bull,除非通过下边与通道关联
                //the expression (ssc.socket().getChannel() != null) is true
                ssc.socket().bind(new InetSocketAddress(PORT));
                //5.将通道注册到Selector,感兴趣的事件为  连接  事件
                ssc.register(sel, SelectionKey.OP_ACCEPT);
                System.out.println("Server on port: " + PORT);
                while(true) {
                    //6.没有事件发生时,一直阻塞等待
                    sel.select();
                    //7.有事件发生时,获取Selector中所有SelectorKey(持有选择器与通道的关联关系)。
                    //由于基于操作系统的poll()方法,当有事件发生时,只返回事件个数,无法确定具体通道,故只能对所有注册的通道进行遍历。
                    Iterator<SelectionKey> it = sel.selectedKeys().iterator();
                    //8.遍历所有SelectorKey,处理事件
                    while(it.hasNext()) {
                        SelectionKey sKey = it.next();
                        it.remove();//防止重复处理
                        //9.判断SelectorKey对应的channel发生的事件是否socket连接
                        if(sKey.isAcceptable()) {
                            //10.与ServerSocket.accept()方法相似,接收到该通道套接字的连接,返回SocketChannel,与客户端进行交互
                            ch = ssc.accept();
                            System.out.println(
                                    "Accepted connection from:" + ch.socket());
                            //11.设置该SocketChannel为非阻塞模式
                            ch.configureBlocking(false);
                            //12.将该通道注册到Selector中,感兴趣的事件为OP_READ(读)
                            ch.register(sel, SelectionKey.OP_READ);
                        }else {
                            //13.发生非连接事件,此处为OP_READ事件。SelectorKey获取注册的SocketChannel,用于读写
                            ch = (SocketChannel)sKey.channel();
                            //14.将数据从channel读到ByteBuffer中
                            ch.read(buffer);
                            CharBuffer cb = cs.decode((ByteBuffer)buffer.flip());
                            String response = cb.toString();
                            System.out.print("Echoing : " + response);
                            //15.再将获取到的数据会写给客户端
                            ch.write((ByteBuffer)buffer.rewind());
                            if(response.indexOf("END") != -1)
                                ch.close();
                            buffer.clear();
                        }
                    }
                }
            } finally {
                if(ch != null)
                    ch.close();
                ssc.close();
                sel.close();
            }
        }
    }

     如代码中注释标明,大致步骤包含:

    • 创建ServerSocketChannel和Selector,设置通道非阻塞,并与服务端的Socket绑定
    • 注册 ServerSocketChannel到Selector,感兴趣的事件为OP_CONNECT(获取连接)
    • select()方法阻塞等待,直到有事件发生
    • 遍历Selector中的所有注册事件,通过SelectorKey维护Selector和Channel关联关系
    • 如果是连接事件,则调ServerSocketChannel.accept()方法获取SocketChannel,与客户端交互
    • 如果是读事件,则通过SelectorKey中获取SocketChannel,读写数据

    运行结果:

    Server on port: 8080

    客户端

    import java.io.IOException;
    import java.net.InetAddress;
    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.Iterator;
    
    import com.test.socketio.JabberServer;
    
    /**
     * 采用这种方式,读与写是非阻塞的
     * 普通的读写是阻塞的,直到读完或写完
     *
     */
    public class JabberClient1 {
        
        static final int clPot = 8899;
    
        public static void main(String[] args) throws IOException{
            //1.创建SocketChannel
            SocketChannel sc = SocketChannel.open();
            //2.创建Selector
            Selector sel = Selector.open();
            try {
                sc.configureBlocking(false);
                //3.关联SocketChannel和Socket,socket绑定到本机端口
                sc.socket().bind(new InetSocketAddress(clPot));
                //4.注册到Selector,感兴趣的事件为OP_CONNECT、OP_READ、OP_WRITE
                sc.register(sel, SelectionKey.OP_CONNECT | SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                int i = 0;
                boolean written = false, done = false;
                String encoding = System.getProperty("file.encoding");
                Charset cs = Charset.forName(encoding);
                ByteBuffer buffer = ByteBuffer.allocate(16);
                while(!done) {
                    sel.select();
                    //5.从选择器中获取所有注册的通道信息(SelectionKey作为标识)
                    Iterator<SelectionKey> it = sel.selectedKeys().iterator();
                    while(it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();
                        //6.获取通道,此处即为上边创建的channel
                        sc = (SocketChannel)key.channel();    
                        //7.判断SelectorKey对应的channel发生的事件是否socket连接,并且还没有连接
                        if(key.isConnectable() && !sc.isConnected()) {
                            InetAddress addr = InetAddress.getByName(null);
                            //连接addr和port对应的服务器
                            boolean success = sc.connect(new InetSocketAddress(addr, JabberServer.PORT));
                            if(!success)
                                sc.finishConnect();
                        }
                        //8.读与写是非阻塞的:客户端写一个信息到服务器,服务器发送一个信息到客户端,客户端再读
                        if(key.isReadable() && written) {
                            if(sc.read((ByteBuffer)buffer.clear()) > 0) {
                                written = false;
                                String response = cs.decode((ByteBuffer)buffer.flip()).toString();
                                System.out.println(response);
                                if(response.indexOf("END") != -1)
                                    done = true;
                            }
                        }
                        if(key.isWritable() && !written) {
                            if(i < 10)
                                sc.write(ByteBuffer.wrap(new String("howdy " + i + "
    ").getBytes()));
                            else if(i == 10){
                                sc.write(ByteBuffer.wrap("END".getBytes()));
                            }
                            written = true;
                            i++;
                        }
                    }
                }
            } finally {
                sc.close();
                sel.close();
            }
        }
    }

    客户端与服务端类似,不同之处:

    • 创建SocketChannel通道,注册到选择器,刚兴趣的事件为OP_CONNECT、OP_READ、OP_WRITE
    • 调试发现,客户端sel.select()不会阻塞,对注册通道不断的遍历,并且每次都可写。原因是OP_WRITE事件会持续生效,即只要连接存在就可以写,不管服务端是否有返回
    • 本例中,客户端发送一条数据,服务端接收一条,并返回给客户端;客户端接到服务端的消息后,才会发生下一条数据,主要通过written标识进行控制的。

    运行机制

    运行结果

    服务端

    Server on port: 8080
    Accepted connection from:Socket[addr=/127.0.0.1,port=8899,localport=8080]
    Echoing : howdy 0
    Echoing : howdy 1
    Echoing : howdy 2
    Echoing : howdy 3
    Echoing : howdy 4
    Echoing : howdy 5
    Echoing : howdy 6
    Echoing : howdy 7
    Echoing : howdy 8
    Echoing : howdy 9
    Echoing : END

    客户端

    howdy 0
    howdy 1
    howdy 2
    howdy 3
    howdy 4
    howdy 5
    howdy 6
    howdy 7
    howdy 8
    howdy 9
    END

    2.核心类分析

    (1)通道(SelectableChannel)

     通道Channel继承体系如下,其中ServerSocketChannel和SocketChannel都继承自SelectableChannel。

    • SelectableChannel通道可以通过Selector实现多路复用(multiplexed)。
    • 通道通过register(Selector,int,Object)方法注册到Selector中,并返回SelectorKey(代表注册到Selector上的注册信息)。
    • 在一个Selector中,同一个通道只能注册一份;是否可以注册到多个Selector中,由程序调用isRegistered()方法决定。
    • SelectableChannel通道是线程安全的。
    • SelectableChannel包含阻塞和非阻塞两种模式,只有非阻塞时才可以注册到Selector中。

    ServerSocketChannel(A selectable channel for stream-oriented listening sockets.),用于监听Socket的基于流的可选通道。

    SocketChannel(A selectable channel for stream-oriented connecting sockets.),用于连接Socket的基于流额可选通道。

    (2)选择器(Selector)

    Selector是SelectableChannel的多路复选器,该类包含以下方法。

    • 通过open()方法创建Selector
    • 包含三种SelectorKey Set:所有注册的SelectorKey、被选的SelectorKey(通道发生事件)、被取消的SelectorKey(不可直接访问)
    • 每次select()操作,都会从被选的SelectorKey集合中删除或新增,清楚被取消的SelectorKey中的SelectorKey

    (3)选择建(SelectorKey)

      选择键封装了特定的通道特定的选择器的注册关系。选择键对象被SelectableChannel.register()返回并提供一个表示这种注册关系的标记。选择键包含了两个比特集(以整数的形式进行编码),指示了该注册关系所关心的通道操作,以及通道已经准备好的操作。包括读、写、连接和接收操作,如下:

        public static final int OP_READ = 1 << 0;
    
        public static final int OP_WRITE = 1 << 2;
    
        public static final int OP_CONNECT = 1 << 3;
    
        public static final int OP_ACCEPT = 1 << 4;

    3.Reactor设计模式

    基于Selector的多路复用IO,机制是采用Reactor设计模式,将一个或多个客户的服务请求分离(demultiplex)和事件分发器 (dispatch)给应用程序(I/O模型之三:两种高性能 I/O 设计模式 Reactor 和 Proactor),即通过Selector阻塞等待事件发生,然后再分发给相应的处理器接口。详情可以参考该篇文章或更多的资料。

    摘自链接文章中的一幅图如下:

    • Reactor是调度中心,包含select()阻塞,等待事件发生,并分发不同的业务处理。
    • 客户端请求连接时,select()接收到事件后,会调acceptor,创建连接并与客户端交互。
    • 客户端写数据给服务端时,select()接收到事件后,调read操作,读取客户端数据,可以采用线程池对与客户端交互,对数据进行处理。
    • 服务端可也以发生数据给客户端。

    4.总结

      1. SelectableChannel(ServerSocketChannel和SocketChannel)可以注册到Selector中,并用选择键(SelectorKey)进行分装

      2. SelectorKey中包含选择器感兴趣的事件(读、写、连接和接收)

      3. Selector中select()方法阻塞,直到注册通道有事件发生,可以一个线程监控多个客户端,实现多路复用

      4. 基于Selector的多路复用采用Reactor设计模式,使得选择器与业务处理进行分离。

      5. Netty是异步基于事件的应用框架,其实现是基于Java NIO的,并对其进行了优化,可以进一步学习。

    5. 参考

    《Thinking in Enterprise Java》

    Java NIO系列教程(六) 多路复用器Selector

  • 相关阅读:
    如何对抗硬件断点--- 调试寄存器
    学会破解要十招
    Ollydbg 中断方法浅探
    脱壳经验(二)--如何优化
    jquery 选择器,模糊匹配
    CSS实现内容超过长度后以省略号显示
    微信内置浏览器浏览H5页面弹出的键盘遮盖文本框的解决办法
    Exif.js获取图片的详细信息(苹果手机移动端上传图片旋转90度)
    jquery中的$("#id")与document.getElementById("id")的区别
    CentOS7 安装lua环境(我是在mysql读写分离用的)
  • 原文地址:https://www.cnblogs.com/shuimuzhushui/p/10323011.html
Copyright © 2011-2022 走看看