java 非阻塞套接字

Non-blocking sockets

提问by jasonline

What's the best way to implement a non-blocking socket in Java?

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

Or is there such a thing? I have a program that communicates with a server through socket but I don't want the socket call to block/cause delay if there is a problem with the data/connection.

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

回答by Teocci

Java non-blocking socket, introduced in Java 2 Standard Edition 1.4, allow net communication between applications without blocking the processes using the sockets. But what a non-blocking socket is, in which contexts it can be useful, and how it works?

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

What a non-blocking socket is?

什么是非阻塞套接字？

A non-blocking socket allows I/O operation on a channel without blocking the processes using it. This means, we can use a single thread to handle multiple concurrent connections and gain an "asynchronous high-performance"read/write operations (some people may not agreed with that)

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

Ok, in which contexts it can be useful?

好的，它在哪些情况下有用？

Suppose you would like to implement a server accepting diverse client connections. Suppose, as well, that you would like the server to be able to process multiple requests simultaneously. Using the traditional way you have two choices to develop such a server:

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

• Implement a multi-thread server that manually handles a thread for each connection.
• Using an external third-party module.

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

Both solutions work, but adopting the first one you have to develop the whole thread-management solution, with related concurrency and conflict troubles. The second solution makes the application dependent on a non-JDK external module and probably you have to adapt the library to your necessities. By means of the non-blocking socket, you can implement a non-blocking server without directly managing threads or resorting to external modules.

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

How it works?

怎么运行的？

Before going into details, there are few terms that you need to understand:

在详细介绍之前，您需要了解几个术语：

• In NIO based implementations, instead of writing data onto output streams and reading data from input streams, we read and write data from buffers. A buffercan be defined as a temporary storage.
• Channeltransports bulk of data into and out of buffers. Also, it can be viewed as an endpoint for communication.
• Readiness Selectionis a concept that refers to “the ability to choose a socket that will not block when data is read or written.”

• 在基于 NIO 的实现中，我们不是将数据写入输出流和从输入流读取数据，而是从缓冲区读取和写入数据。甲缓冲器可以被定义为一个临时存储。
• 通道将大量数据传入和传出缓冲区。此外，它可以被视为通信的端点。
• Readiness Selection是一个概念，指的是“选择一个在数据读写时不会阻塞的套接字的能力”。

Java NIO has a class called Selectorthat allows a single thread to examine I/O events on multiple channels. How is this possible? Well, the selectorcan check the "readiness"of a channel for events such as a client attempting a connection, or a read/write operation. This is, each instance of Selectorcan monitor more socket channelsand thus more connections. Now, when something happens on the channel (an event occurs), the selectorinforms the application to process the request. The selectordoes it by creating event keys(or selection keys), which are instances of the SelectionKeyclass. Each keyholds information about who is making the requestand what type of the request is, as shown in the Figure 1.

Java NIO 有一个名为的类Selector，它允许单个线程检查多个通道上的 I/O 事件。这怎么可能？好吧，selector可以检查通道的“就绪”事件，例如客户端尝试连接或读/写操作。也就是说，每个实例都Selector可以监控更多的套接字通道，从而监控更多的连接。现在，当通道上发生某些事情（发生事件）时，selector通知应用程序处理请求。的selector实现方法是创建它的事件的键（或选择键），其为实例SelectionKey的类。每个人都key持有有关谁提出请求的信息，以及请求的类型是什么，如图1所示。

Figure 1: Structure diagram

图1：结构图

A basic implementation

一个基本的实现

A server implementation consists of an infinite loop in which the selectorwaits for events and creates the event keys. There are four possible types for a key:

服务器实现由一个无限循环组成，在该循环中selector等待事件并创建事件键。密钥有四种可能的类型：

• Acceptable: the associated client requests a connection.
• Connectable: the server accepted the connection.
• Readable: the server can read.
• Writeable: the server can write.

• 可接受：关联的客户端请求连接。
• 可连接：服务器接受连接。
• 可读：服务器可以读取。
• 可写：服务器可以写。

Usually acceptablekeys are created on the server side. In fact, this kind of key simply informs the server that a client required a connection, then the server individuates the socket channel and associates this to the selector for read/write operations. After this, when the accepted client reads or writes something, the selector will create readableor writeablekeys for that client..

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

Now you are ready to write the server in Java, following the proposed algorithm. The creation of the socket channel, the selector, and the socket-selector registration can be made in this way:

现在您已准备好按照建议的算法用 Java 编写服务器。可以通过以下方式创建套接字通道、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);

First we create an instance of SocketChannelwith ServerSocketChannel.open()method. Next, configureBlocking(false)invocation sets this channelas nonblocking. The connection to the server is made by serverChannel.socket().bind()method. The HOSTNAMErepresents the IP address of the server, and PORTis the communication port. Finally, invoke Selector.open()method to create a selectorinstance and register it to the channeland registration type. In this example, the registration type is OP_ACCEPT, which means the selector merely reports that a client attempts a connection to the server. Other possible options are: OP_CONNECT, which will be used by the client; OP_READ; and OP_WRITE.

首先我们创建一个SocketChannelwithServerSocketChannel.open()方法的实例。接下来，configureBlocking(false)调用将 this 设置channel为 nonblocking。与服务器的连接是通过serverChannel.socket().bind()方法进行的。该HOSTNAME代表服务器的IP地址，并且PORT是通信端口。最后，调用Selector.open()方法创建一个selector实例并将其注册到channel和注册类型。在此示例中，注册类型为OP_ACCEPT，这意味着选择器仅报告客户端尝试连接到服务器。其他可能的选项是：OP_CONNECT，将由客户端使用；OP_READ; 和OP_WRITE。

Now we need to handle this requests using an infinite loop. A simple way is the following:

现在我们需要使用无限循环来处理这个请求。一个简单的方法如下：

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

You can find the implementation source here

你可以在这里找到实现源

NOTE: Asynchronous Server

注意：异步服务器

An alternative to the the Non-blocking implementation we can deploy an Asynchronous Server. For instance, you can use the AsynchronousServerSocketChannelclass, which provides an asynchronous channel for stream-oriented listening sockets.

我们可以部署异步服务器来替代非阻塞实现。例如，您可以使用AsynchronousServerSocketChannel为面向流的侦听套接字提供异步通道的类。

To use it, first execute its static open()method and then bind()it to a specific port. Next, you'll execute its accept()method, passing to it a class that implements the CompletionHandlerinterface. Most often, you'll find that handler created as an anonymous inner class.

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

From this AsynchronousServerSocketChannelobject, you invoke accept()to tell it to start listening for connections, passing to it a custom CompletionHandlerinstance. When we invoke accept(), it returns immediately. Note that this is different from the traditional blocking approach; whereas the accept()method blocked until a client connected to it, the AsynchronousServerSocketChannelaccept()method handles it for you.

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

Here you have an example:

这里有一个例子：

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

You can find the full code here

你可以在这里找到完整的代码

回答by user207421

What's the best way to implement a non-blocking socket in Java?

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

There is only one way. SocketChannel.configureBlocking(false).

只有一种方法。SocketChannel.configureBlocking(false).

Note that several of these answers are incorrect. SocketChannel.configureBlocking(false)puts it into non-blocking mode. You don't need a Selectorto do that. You only need a Selectorto implement timeouts or multiplexedI/O with non-blocking sockets.

请注意，其中一些答案是不正确的。SocketChannel.configureBlocking(false)将其置于非阻塞模式。你不需Selector要这样做。您只需要使用非阻塞套接字Selector来实现超时或多路复用I/O。

回答by Peter Lawrey

Apart from using non blocking IO, you might find it is much simpler to have a writing thread for your connection.

除了使用非阻塞 IO，您可能会发现为您的连接设置一个写入线程要简单得多。

Note: if you only need a few thousand connections, one to two threads per connection is simpler. If you have around ten thousand or more connections per server you need NIO with Selectors.

注意：如果你只需要几千个连接，每个连接一到两个线程更简单。如果每台服务器有大约一万个或更多的连接，则需要带有选择器的 NIO。

回答by secmask

java.nio package provides Selectorworking much like as in C.

java.nio 包提供了Selector 的工作方式，就像在 C 中一样。

回答by rockstar

I just wrote this code . It works well . This is an example of the Java NIO as mentioned in the above answers but here i post the code .

我刚刚写了这段代码。它运作良好。这是上述答案中提到的 Java NIO 的一个示例，但我在这里发布了代码。

ServerSocketChannel ssc = null;
try {
    ssc = ServerSocketChannel.open();
    ssc.socket().bind(new InetSocketAddress(port));
    ssc.configureBlocking(false);
    while (true) {
        SocketChannel sc = ssc.accept();
        if (sc == null) {
            // No connections came .
        } else {
            // You got a connection. Do something
        }
    }
} catch (IOException e) {
    e.printStackTrace();
}