java - 非阻塞套接字

标签 java sockets network-programming blocking nonblocking

在 Java 中实现非阻塞套接字的最佳方法是什么?

或者有这样的事情吗?我有一个通过套接字与服务器通信的程序,但如果数据/连接出现问题,我不希望套接字调用阻塞/导致延迟。

最佳答案

Java 非阻塞套接字 ,在 Java 2 Standard Edition 1.4 中引入,允许应用程序之间进行网络通信,而不会阻塞使用套接字的进程。但是什么是非阻塞套接字,它在哪些上下文中有用,以及它是如何工作的?

什么是非阻塞套接字?

非阻塞套接字允许在 channel 上进行 I/O 操作而不阻塞使用它的进程。这意味着,我们可以使用单个线程来处理多个并发连接并获得“异步高性能”读/写操作(有些人可能不同意)

好的,它在哪些情况下有用?

假设您想实现一个接受不同客户端连接的服务器。还假设您希望服务器能够同时处理多个请求。使用传统方式,您有两种选择来开发这样的服务器:

  • 实现一个多线程服务器,为每个连接手动处理一个线程。
  • 使用外部第三方模块。

  • 两种解决方案都有效,但采用第一种方法您必须开发整个线程管理解决方案,并存在相关的并发和冲突问题。第二种解决方案使应用程序依赖于非 JDK 外部模块,并且您可能必须使库适应您的需要。通过非阻塞套接字,您可以实现非阻塞服务器,而无需直接管理线程或求助于外部模块。

    这个怎么运作?

    在详细介绍之前,您需要了解几个术语:
  • 在基于 NIO 的实现中,我们不是将数据写入输出流和从输入流读取数据,而是从缓冲区读取和写入数据。缓冲区可以定义为临时存储。
  • channel 将大量数据传入和传出缓冲区。此外,它可以被视为通信的端点。
  • Readiness Selection 是一个概念,指的是“选择一个在读取或写入数据时不会阻塞的套接字的能力”。

  • Java NIO 有一个名为 Selector 的类它允许单个线程检查多个 channel 上的 I/O 事件。这怎么可能?那么,selector可以检查 channel 的“就绪”事件,例如客户端尝试连接或读/写操作。这是 Selector 的每个实例可以监控更多 socket channel 从而有更多的联系。现在,当 channel 上发生某些事情(事件发生)时,selector通知申请处理请求 . selector通过创建 来做到这一点事件键 (或选择键),它们是 SelectionKey 的实例类(class)。每个key持有有关的信息谁在提出请求 请求的类型是什么 ,如图 1 所示。

    Figure 1: Structure diagram图1:结构图

    一个基本的实现

    服务器实现由一个无限循环组成,其中 selector等待事件并创建事件键。 key 有四种可能的类型:
  • 可接受:关联的客户端请求连接。
  • 可连接:服务器接受连接。
  • 可读:服务器可以读取。
  • 可写:服务器可以写。

  • 通常 acceptable key 是在服务器端创建的。实际上,这种键只是简单地通知服务器客户端需要连接,然后服务器将套接字 channel 个性化并将其关联到选择器以进行读/写操作。此后,当接受的客户端读取或写入某些内容时,选择器将创建 readablewriteable该客户的 key ..

    现在您已准备好按照建议的算法用 Java 编写服务器。套接字 channel 的创建,selector ,并且可以通过这种方式进行套接字选择器注册:
    final String HOSTNAME = "127.0.0.1";
    final int PORT = 8511;
    
    // This is how you open a ServerSocketChannel
    serverChannel = ServerSocketChannel.open();
    // You MUST configure as non-blocking or else you cannot register the serverChannel to the Selector.
    serverChannel.configureBlocking(false);
    // bind to the address that you will use to Serve.
    serverChannel.socket().bind(new InetSocketAddress(HOSTNAME, PORT));
    
    // This is how you open a Selector
    selector = Selector.open();
    /*
     * Here you are registering the serverSocketChannel to accept connection, thus the OP_ACCEPT.
     * This means that you just told your selector that this channel will be used to accept connections.
     * We can change this operation later to read/write, more on this later.
     */
    serverChannel.register(selector, SelectionKey.OP_ACCEPT);
    

    首先我们创建一个 SocketChannel 的实例与 ServerSocketChannel.open()方法。接下来,configureBlocking(false)调用设置此 channel 作为非阻塞 .到服务器的连接是由 serverChannel.socket().bind() 建立的方法。 HOSTNAME代表服务器的IP地址,PORT是通讯端口。最后,调用 Selector.open()创建 selector 的方法实例并将其注册到 channel和注册类型。本例中,注册类型为OP_ACCEPT ,这意味着选择器仅报告客户端尝试连接到服务器。其他可能的选项是:OP_CONNECT , 将由客户端使用; OP_READ ;和 OP_WRITE .

    现在我们需要使用无限循环来处理这个请求。一个简单的方法如下:
    // Run the server as long as the thread is not interrupted.
    while (!Thread.currentThread().isInterrupted()) {
        /*
         * selector.select(TIMEOUT) is waiting for an OPERATION to be ready and is a blocking call.
         * For example, if a client connects right this second, then it will break from the select()
         * call and run the code below it. The TIMEOUT is not needed, but its just so it doesn't
         * block undefinable.
         */
        selector.select(TIMEOUT);
    
        /*
         * If we are here, it is because an operation happened (or the TIMEOUT expired).
         * We need to get the SelectionKeys from the selector to see what operations are available.
         * We use an iterator for this.
         */
        Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
    
        while (keys.hasNext()) {
            SelectionKey key = keys.next();
            // remove the key so that we don't process this OPERATION again.
            keys.remove();
    
            // key could be invalid if for example, the client closed the connection.
            if (!key.isValid()) {
                continue;
            }
            /*
             * In the server, we start by listening to the OP_ACCEPT when we register with the Selector.
             * If the key from the keyset is Acceptable, then we must get ready to accept the client
             * connection and do something with it. Go read the comments in the accept method.
             */
            if (key.isAcceptable()) {
                System.out.println("Accepting connection");
                accept(key);
            }
            /*
             * If you already read the comments in the accept() method, then you know we changed
             * the OPERATION to OP_WRITE. This means that one of these keys in the iterator will return
             * a channel that is writable (key.isWritable()). The write() method will explain further.
             */
            if (key.isWritable()) {
                System.out.println("Writing...");
                write(key);
            }
            /*
             * If you already read the comments in the write method then you understand that we registered
             * the OPERATION OP_READ. That means that on the next Selector.select(), there is probably a key
             * that is ready to read (key.isReadable()). The read() method will explain further.
             */
            if (key.isReadable()) {
                System.out.println("Reading connection");
                read(key);
            }
        }
    }
    

    您可以找到 the implementation source here

    注意:异步服务器

    我们可以部署一个异步服务器来替代非阻塞实现。例如,您可以使用 AsynchronousServerSocketChannel类,它为面向流的监听套接字提供异步 channel 。

    要使用它,首先执行它的静态 open()方法,然后 bind()它到一个特定的 端口 .接下来,您将执行其 accept()方法,传递给它一个实现 CompletionHandler 的类界面。大多数情况下,您会发现该处理程序被创建为匿名内部类。

    从这里 AsynchronousServerSocketChannel对象,你调用 accept()告诉它开始监听连接,传递给它一个自定义 CompletionHandler实例。当我们调用 accept() ,它立即返回。请注意,这与传统的阻塞方法不同;而accept()方法 阻塞,直到客户端连接到它 , AsynchronousServerSocketChannel accept()方法为您处理。

    这里有一个例子:
    public class NioSocketServer
    {
        public NioSocketServer()
        {
            try {
                // Create an AsynchronousServerSocketChannel that will listen on port 5000
                final AsynchronousServerSocketChannel listener = AsynchronousServerSocketChannel
                        .open()
                        .bind(new InetSocketAddress(5000));
    
                // Listen for a new request
                listener.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>()
                {
                    @Override
                    public void completed(AsynchronousSocketChannel ch, Void att)
                    {
                        // Accept the next connection
                        listener.accept(null, this);
    
                        // Greet the client
                        ch.write(ByteBuffer.wrap("Hello, I am Echo Server 2020, let's have an engaging conversation!\n".getBytes()));
    
                        // Allocate a byte buffer (4K) to read from the client
                        ByteBuffer byteBuffer = ByteBuffer.allocate(4096);
                        try {
                            // Read the first line
                            int bytesRead = ch.read(byteBuffer).get(20, TimeUnit.SECONDS);
    
                            boolean running = true;
                            while (bytesRead != -1 && running) {
                                System.out.println("bytes read: " + bytesRead);
    
                                // Make sure that we have data to read
                                if (byteBuffer.position() > 2) {
                                    // Make the buffer ready to read
                                    byteBuffer.flip();
    
                                    // Convert the buffer into a line
                                    byte[] lineBytes = new byte[bytesRead];
                                    byteBuffer.get(lineBytes, 0, bytesRead);
                                    String line = new String(lineBytes);
    
                                    // Debug
                                    System.out.println("Message: " + line);
    
                                    // Echo back to the caller
                                    ch.write(ByteBuffer.wrap(line.getBytes()));
    
                                    // Make the buffer ready to write
                                    byteBuffer.clear();
    
                                    // Read the next line
                                    bytesRead = ch.read(byteBuffer).get(20, TimeUnit.SECONDS);
                                } else {
                                    // An empty line signifies the end of the conversation in our protocol
                                    running = false;
                                }
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        } catch (ExecutionException e) {
                            e.printStackTrace();
                        } catch (TimeoutException e) {
                            // The user exceeded the 20 second timeout, so close the connection
                            ch.write(ByteBuffer.wrap("Good Bye\n".getBytes()));
                            System.out.println("Connection timed out, closing connection");
                        }
    
                        System.out.println("End of conversation");
                        try {
                            // Close the connection if we need to
                            if (ch.isOpen()) {
                                ch.close();
                            }
                        } catch (I/OException e1)
                        {
                            e1.printStackTrace();
                        }
                    }
    
                    @Override
                    public void failed(Throwable exc, Void att)
                    {
                        ///...
                    }
                });
            } catch (I/OException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args)
        {
            NioSocketServer server = new NioSocketServer();
            try {
                Thread.sleep(60000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    

    您可以找到 the full code here

    关于java - 非阻塞套接字,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/3895461/

    相关文章:

    java - 没有 String 方法的自定义 indexOf

    java - 如何忽略 Eclipse 中的特定异常

    swing - java网络编程的最佳框架?

    linux - 获取 Linux 内核中的网络设备列表

    c - 我应该在套接字生命周期内什么时候设置 TCP_QUICKACK 选项?

    security - 如何唯一标识一个网络?

    java - 尽管使用 RxJava 在另一个线程上订阅,但使用 Google 的 People API 时仍然收到 IllegalStateException

    java - HMAC 一个 php 字节数组

    sockets - 如何设置 std::net::UdpSocket 超时?

    java - 通过套接字发送对象