zoukankan      html  css  js  c++  java
  • Java Socket NIO

    服务端:

    public class NIOServer {
        private static final String HOST = "localhost";
        private static final int PORT = 10086;
    
        public static void main(String[] args) {
            ServerSocketChannel serverSocketChannel = null;
            ServerSocket serverSocket = null;
            Selector selector = null;
            try {
                serverSocketChannel = ServerSocketChannel.open();//工厂方法创建ServerSocketChannel
                serverSocket = serverSocketChannel.socket(); //获取channel对应的ServerSocket
                serverSocket.bind(new InetSocketAddress(HOST, PORT)); //绑定地址
                serverSocketChannel.configureBlocking(false); //设置ServerSocketChannel非阻塞模式
                selector = Selector.open();//工厂方法创建Selector
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//通道注册选择器,接受连接就绪状态。
                while (true) {//循环检查
                    if (selector.select() == 0) {//阻塞检查,当有就绪状态发生,返回键集合
                        continue;
                    }
                    // if (selector.select(2000) == 0) {
                    //     //在等待信道准备的同时,也可以异步地执行其他任务,  这里打印*
                    //     System.out.print("*");
                    //     continue;
                    // }
                    Iterator<SelectionKey> it = selector.selectedKeys().iterator(); //获取就绪键遍历对象。
                    while (it.hasNext()) {
                        SelectionKey selectionKey = it.next();
                        //处理就绪状态
                        if (selectionKey.isAcceptable()) {
                            ServerSocketChannel schannel = (ServerSocketChannel) selectionKey.channel();//只负责监听,阻塞,管理,不发送、接收数据
                            SocketChannel socketChannel = schannel.accept();//就绪后的操作,刚到达的socket句柄
                            if (null == socketChannel) {
                                continue;
                            }
                            socketChannel.configureBlocking(false);
                            socketChannel.register(selector, SelectionKey.OP_READ); //告知选择器关心的通道,准备好读数据
                        } else if (selectionKey.isReadable()) {
                            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                            ByteBuffer byteBuffer = ByteBuffer.allocate(4 * 1024);
                            System.out.println(socketChannel.getRemoteAddress());
                            int read = socketChannel.read(byteBuffer);
                            if (read == -1) {//如果不关闭会一直产生isReadable这个消息
                                selectionKey.cancel();
                                socketChannel.close();
    
                            } else {
                                StringBuilder result = new StringBuilder();
                                while (read > 0) {//确保读完
                                    byteBuffer.flip();
                                    result.append(new String(byteBuffer.array()));
                                    byteBuffer.clear();//每次清空 对应上面flip()
                                    read = socketChannel.read(byteBuffer);
                                }
    
                                System.out.println("server receive: " + result.toString());
                                socketChannel.register(selector, SelectionKey.OP_WRITE);
                            }
                        } else if (selectionKey.isWritable()) {
                            SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                            String sendStr = "server send data: " + Math.random();
                            ByteBuffer send = ByteBuffer.wrap(sendStr.getBytes());
                            while (send.hasRemaining()) {
                                socketChannel.write(send);
                            }
                            socketChannel.register(selector, SelectionKey.OP_READ);
                            System.out.println(sendStr);
                        }
                        it.remove();
                    }
                }
    
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    客户端:

    public class NIOClient {
        public static void main(String[] args) throws Exception {
            SocketChannel clntChan = SocketChannel.open();
            clntChan.configureBlocking(false);
            if (!clntChan.connect(new InetSocketAddress("localhost", 10086))) {
                //不断地轮询连接状态,直到完成连接
                while (!clntChan.finishConnect()) {
                    //在等待连接的时间里,可以执行其他任务,以充分发挥非阻塞IO的异步特性
                    //这里为了演示该方法的使用,只是一直打印"."
                    System.out.print(".");
                }
            }
            //为了与后面打印的"."区别开来,这里输出换行符
            System.out.print("
    ");
            //分别实例化用来读写的缓冲区
            ByteBuffer writeBuf = ByteBuffer.wrap("send send send".getBytes());
            ByteBuffer readBuf = ByteBuffer.allocate("send".getBytes().length - 1);
            while (writeBuf.hasRemaining()) {
                //如果用来向通道中写数据的缓冲区中还有剩余的字节,则继续将数据写入信道
                clntChan.write(writeBuf);
            }
            Thread.sleep(10000);
            StringBuffer stringBuffer = new StringBuffer();
            //如果read()接收到-1,表明服务端关闭,抛出异常
            while ((clntChan.read(readBuf)) > 0) {
                readBuf.flip();
                stringBuffer.append(new String(readBuf.array(), 0, readBuf.limit()));
                readBuf.clear();
            }
            //打印出接收到的数据
            System.out.println("Client Received: " + stringBuffer.toString());
            //关闭信道
            clntChan.close();
        }
    }

    注意当客户端断开socket的时候需要处理,不然服务端对应的channel会一直处于readable的状态,会造成死循环。

    当客户端调用:clntChan.close();

    这一句是正常关闭代码,它会传给服务端关闭的指令(也就是数据)。

    而服务端的socketChannel.read(byteBuffer)=-1则代表收到这个关闭指令(数据),可以依据-1来关闭服务端的channel,不过仍有可能有问题,可能是客户端调用socketChannel.close()的时候,服务端这边的bytebuffer已经满了,那么num只能返回0而不能返回-1。具体该怎么写你应该根据你自己的业务来处理。

    当socketChannel为阻塞方式时(默认就是阻塞方式)read函数,不会返回0,阻塞方式的socketChannel,若没有数据可读,或者缓冲区满了,就会阻塞,直到满足读的条件,所以一般阻塞方式的read是比较简单的,不过阻塞方式的socketChannel的问题也是显而易见的。这里我结合基于NIO 写ftp服务器调试过程中碰到的问题,总结一下非阻塞场景下的read碰到的问题。注意:这里的场景都是基于客户端以阻塞socket的方式发送数据。

    1、read什么时候返回-1

    read返回-1说明客户端的数据发送完毕,并且主动的close socket。所以在这种场景下,(服务器程序)你需要关闭socketChannel并且取消key,最好是退出当前函数。注意,这个时候服务端要是继续使用该socketChannel进行读操作的话,就会抛出“远程主机强迫关闭一个现有的连接”的IO异常。

    2、read什么时候返回0

    其实read返回0有3种情况,一是某一时刻socketChannel中当前(注意是当前)没有数据可以读,这时会返回0,其次是bytebuffer的position等于limit了,即bytebuffer的remaining等于0,这个时候也会返回0,最后一种情况就是客户端的数据发送完毕了(注意看后面的程序里有这样子的代码),这个时候客户端想获取服务端的反馈调用了recv函数,若服务端继续read,这个时候就会返回0。

    -------------------------------------------------------------------------------------------------

    实际写代码过程中观察发现,如果客户端发送数据后不关闭channel,同时服务端收到数据后反倒再次发给客户端,那么此时客户端read方法永远返回0.

    SelectionKey.OP_WRITE

    当时有一个连接进来后,如果注册了SelectionKey.OP_WRITE消息,selectionKey.isWritable()会一直返回true知道写缓冲区写满。所以这点就值注册了SelectionKey.OP_READ.

    OP_WRITE事件的就绪条件并不是发生在调用channel的write方法之后,而是在当底层缓冲区有空闲空间的情况下。因为写缓冲区在绝大部分时候都是有空闲空间的,所以如果你注册了写事件,这会使得写事件一直处于就就绪,选择处理现场就会一直占用着CPU资源。所以,只有当你确实有数据要写时再注册写操作,并在写完以后马上取消注册。
    其实,在大部分情况下,我们直接调用channel的write方法写数据就好了,没必要都用OP_WRITE事件。那么OP_WRITE事件主要是在什么情况下使用的了?
    其实OP_WRITE事件主要是在发送缓冲区空间满的情况下使用的。如:
    while (buffer.hasRemaining()) {
         int len = socketChannel.write(buffer);   
         if (len == 0) {
              selectionKey.interestOps(selectionKey.interestOps() | SelectionKey.OP_WRITE);
              selector.wakeup();
              break;
         }
    }
    当buffer还有数据,但缓冲区已经满的情况下,socketChannel.write(buffer)会返回已经写出去的字节数,此时为0。那么这个时候我们就需要注册OP_WRITE事件,这样当缓冲区又有空闲空间的时候就会触发OP_WRITE事件,这是我们就可以继续将没写完的数据继续写出了。
    而且在写完后,一定要记得将OP_WRITE事件注销:
    selectionKey.interestOps(sk.interestOps() & ~SelectionKey.OP_WRITE);
    注意,这里在修改了interest之后调用了wakeup();方法是为了唤醒被堵塞的selector方法,这样当while中判断selector返回的是0时,会再次调用selector.select()。而selectionKey的interest是在每次selector.select()操作的时候注册到系统进行监听的,所以在selector.select()调用之后修改的interest需要在下一次selector.select()调用才会生效。

    nio的select()的时候,只要数据通道允许写,每次select()返回的OP_WRITE都是true。所以在nio的写数据里面,我们在每次需要写数据之前把数据放到缓冲区,并且注册OP_WRITE,对selector进行wakeup(),这样这一轮select()发现有OP_WRITE之后,将缓冲区数据写入channel,清空缓冲区,并且反注册OP_WRITE,写数据完成。这里面需要注意的是,每个SocketChannel只对应一个SelectionKey,也就是说,在上述的注册和反注册OP_WRITE的时候,不是通过channel.register()和key.cancel()做到的,而是通过key.interestOps()做到的。

    public void write(MessageSession session, ByteBuffer buffer) throws ClosedChannelException {
       SelectionKey key = session.key();
       if((key.interestOps() & SelectionKey.OP_WRITE) == 0) {
        key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
       }
       try {
       writebuf.put(buffer);
       } catch(Exception e) {
        System.out.println("want put:"+buffer.remaining()+", left:"+writebuf.remaining());
        e.printStackTrace();
       }
       selector.wakeup();
    }
    while(true) {
       selector.select();
       .....
            if(key.isWritable()) {
             MessageSession session = (MessageSession)key.attachment();
             //System.out.println("Select a write");
             synchronized(session) {
              writebuf.flip();
              SocketChannel channel = (SocketChannel)key.channel();
              int count = channel.write(writebuf);
              //System.out.println("write "+count+" bytes");
              writebuf.clear();
              key.interestOps(SelectionKey.OP_READ);
             }
            }
            ......
        }

    要点一:不推荐直接写channel,而是通过缓存和attachment传入要写的数据,改变interestOps()来写数据;

    要点二:每个channel只对应一个SelectionKey,所以,只能改变interestOps(),不能register()和cancel()。

    参考:https://www.cnblogs.com/burgeen/p/3618059.html

  • 相关阅读:
    SVN
    jenkins可视化
    Tomcat
    防火墙
    keepalived
    MHA高可用
    http
    inotifywait
    DNS
    nginx
  • 原文地址:https://www.cnblogs.com/grasp/p/10607271.html
Copyright © 2011-2022 走看看