zoukankan      html  css  js  c++  java
  • Java Socket(3): NIO

    NIO采取通道(Channel)和缓冲区(Buffer)来传输和保存数据,它是非阻塞式的I/O,即在等待连接、读写数据(这些都是在一线程以客户端的程序中会阻塞线程的操作)的时候,程序也可以做其他事情,以实现线程的异步操作。

       考虑一个即时消息服务器,可能有上千个客户端同时连接到服务器,但是在任何时刻只有非常少量的消息需要读取和分发(如果采用线程池或者一线程一客户端方式,则会非常浪费资源),这就需要一种方法能阻塞等待,直到有一个信道可以进行I/O操作。NIO的Selector选择器就实现了这样的功能,一个Selector实例可以同时检查一组信道的I/O状态,它就类似一个观察者,只要我们把需要探知的SocketChannel告诉Selector,我们接着做别的事情,当有事件(比如,连接打开、数据到达等)发生时,它会通知我们,传回一组SelectionKey,我们读取这些Key,就会获得我们刚刚注册过的SocketChannel,然后,我们从这个Channel中读取数据,接着我们可以处理这些数据。

        Selector内部原理实际是在做一个对所注册的Channel的轮询访问,不断的轮询(目前就这一个算法),一旦轮询到一个Channel有所注册的事情发生,比如数据来了,它就会读取Channel中的数据,并对其进行处理。

        要使用选择器,需要创建一个Selector实例,并将其注册到想要监控的信道上(通过Channel的方法实现)。最后调用选择器的select()方法,该方法会阻塞等待,直到有一个或多个信道准备好了I/O操作或等待超时,或另一个线程调用了该选择器的wakeup()方法。现在,在一个单独的线程中,通过调用select()方法,就能检查多个信道是否准备好进行I/O操作,由于非阻塞I/O的异步特性,在检查的同时,我们也可以执行其他任务。

       基于NIO的TCP连接的建立步骤

     

      服务端

        1、传建一个Selector实例;

        2、将其注册到各种信道,并指定每个信道上感兴趣的I/O操作;

        3、重复执行:

            1)调用一种select()方法;

            2)获取选取的键列表;

            3)对于已选键集中的每个键:

               a、获取信道,并从键中获取附件(如果为信道及其相关的key添加了附件的话);

               b、确定准备就绪的操纵并执行,如果是accept操作,将接收的信道设置为非阻塞模式,并注册到选择器;

               c、如果需要,修改键的兴趣操作集;

               d、从已选键集中移除键

         客户端

       与基于多线程的TCP客户端大致相同,只是这里是通过信道建立的连接,但在等待连接建立及读写时,我们可以异步地执行其他任务。

    服务器端例子 

    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.ServerSocketChannel;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    
    /**
     * NIO服务端
     * @author 小路
     */
    public class NIOServer {
        //通道管理器
        private Selector selector;
    
        /**
         * 获得一个ServerSocket通道,并对该通道做一些初始化的工作
         * @param port  绑定的端口号
         * @throws IOException
         */
        public void initServer(int port) throws IOException {
            // 获得一个ServerSocket通道
            ServerSocketChannel serverChannel = ServerSocketChannel.open();
            // 设置通道为非阻塞
            serverChannel.configureBlocking(false);
            // 将该通道对应的ServerSocket绑定到port端口
            serverChannel.socket().bind(new InetSocketAddress(port));
            // 获得一个通道管理器
            this.selector = Selector.open();
            //将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_ACCEPT事件,注册该事件后,
            //当该事件到达时,selector.select()会返回,如果该事件没到达selector.select()会一直阻塞。
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        }
    
        /**
         * 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
         * @throws IOException
         */
        @SuppressWarnings("unchecked")
        public void listen() throws IOException {
            System.out.println("服务端启动成功!");
            // 轮询访问selector
            while (true) {
                //当注册的事件到达时,方法返回;否则,该方法会一直阻塞
                selector.select();
                //System.out.println(selector.select());
                // 获得selector中选中的项的迭代器,选中的项为注册的事件
                
                Iterator ite = this.selector.selectedKeys().iterator();
                //因为与注册事件关联的可能不止一个,所以下面用next循环处理
                while (ite.hasNext()) {
                    
                    SelectionKey key = (SelectionKey) ite.next();
                    // 删除已选的key,以防重复处理
                    ite.remove();
                    // 客户端请求连接事件
                    if (key.isAcceptable()) {
                        System.out.println("register");
                        ServerSocketChannel server = (ServerSocketChannel) key
                                .channel();
                        // 获得和客户端连接的通道
                        SocketChannel channel = server.accept();
                        // 设置成非阻塞
                        channel.configureBlocking(false);
    
                        //在这里可以给客户端发送信息哦
                        channel.write(ByteBuffer.wrap(new String("向客户端发送了一条信息").getBytes()));
                        //在和客户端连接成功之后,为了可以接收到客户端的信息,需要给通道设置读的权限。
                        channel.register(this.selector, SelectionKey.OP_READ);
                        
                        // 获得了可读的事件
                    } else if (key.isReadable()) {
                        System.out.println("read");
                            read(key);
                    }
    
                }
    
            }
        }
        /**
         * 处理读取客户端发来的信息 的事件
         * @param key
         * @throws IOException 
         */
        public void read(SelectionKey key) throws IOException{
            // 服务器可读取消息:得到事件发生的Socket通道
            SocketChannel channel = (SocketChannel) key.channel();
            // 创建读取的缓冲区
            ByteBuffer buffer = ByteBuffer.allocate(10);
            channel.read(buffer);
            byte[] data = buffer.array();
            String msg = new String(data).trim();
            System.out.println("服务端收到信息:"+msg);
            ByteBuffer outBuffer = ByteBuffer.wrap(msg.getBytes());
            channel.write(outBuffer);// 将消息回送给客户端
        }
        
        /**
         * 启动服务端测试
         * @throws IOException 
         */
        public static void main(String[] args) throws IOException {
            NIOServer server = new NIOServer();
            server.initServer(8000);
            server.listen();
        }
    
    }
    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.util.Iterator;
    
    /**
     * NIO客户端
     * @author 小路
     */
    public class NIOClient {
        //通道管理器
        private Selector selector;
    
        /**
         * 获得一个Socket通道,并对该通道做一些初始化的工作
         * @param ip 连接的服务器的ip
         * @param port  连接的服务器的端口号         
         * @throws IOException
         */
        public void initClient(String ip,int port) throws IOException {
            // 获得一个Socket通道
            SocketChannel channel = SocketChannel.open();
            // 设置通道为非阻塞
            channel.configureBlocking(false);
            // 获得一个通道管理器
            this.selector = Selector.open();
            
            // 客户端连接服务器,其实方法执行并没有实现连接,需要在listen()方法中调
            //用channel.finishConnect();才能完成连接
            channel.connect(new InetSocketAddress(ip,port));
            //将通道管理器和该通道绑定,并为该通道注册SelectionKey.OP_CONNECT事件。
            channel.register(selector, SelectionKey.OP_CONNECT);
        }
    
        /**
         * 采用轮询的方式监听selector上是否有需要处理的事件,如果有,则进行处理
         * @throws IOException
         */
        @SuppressWarnings("unchecked")
        public void listen() throws IOException {
            // 轮询访问selector
            while (true) {
                selector.select();
                // 获得selector中选中的项的迭代器
                Iterator ite = this.selector.selectedKeys().iterator();
                while (ite.hasNext()) {
                    SelectionKey key = (SelectionKey) ite.next();
                    // 删除已选的key,以防重复处理
                    ite.remove();
                    // 连接事件发生
                    if (key.isConnectable()) {
                        SocketChannel channel = (SocketChannel) key
                                .channel();
                        // 如果正在连接,则完成连接
                        if(channel.isConnectionPending()){
                            channel.finishConnect();
                            
                        }
                        // 设置成非阻塞
                        channel.configureBlocking(false);
    
                        //在这里可以给服务端发送信息哦
                        channel.write(ByteBuffer.wrap(new String("向服务端发送了一条信息").getBytes()));
                        //在和服务端连接成功之后,为了可以接收到服务端的信息,需要给通道设置读的权限。
                        channel.register(this.selector, SelectionKey.OP_READ);
                        
                        // 获得了可读的事件
                    } else if (key.isReadable()) {
                            read(key);
                    }
    
                }
    
            }
        }
        /**
         * 处理读取服务端发来的信息 的事件
         * @param key
         * @throws IOException 
         */
        public void read(SelectionKey key) throws IOException{
            //和服务端的read方法一样
        }
        
        
        /**
         * 启动客户端测试
         * @throws IOException 
         */
        public static void main(String[] args) throws IOException {
            NIOClient client = new NIOClient();
            client.initClient("localhost",8000);
            client.listen();
        }
    
    }
  • 相关阅读:
    day 29-2 socket 文件传输、udp协议
    day 29-1 socket
    day 28-1 元类、异常处理
    day 27-1 反射、内置方法
    day 26-1 property、绑定与非绑定方法
    day 25-1 接口类、抽象类、多态
    day 24-1 继承
    day 23-1 类的命名空间、组合
    day 22
    PyMySQL操作mysql数据库(py3必学)
  • 原文地址:https://www.cnblogs.com/yxzfscg/p/4771365.html
Copyright © 2011-2022 走看看