Getting started with the Asynchronous channel API in the Java NIO.2 series

An overview of the NIO. 2

NIO.2, also known as AIO, introduced NIO 2, an improved version of NIO, in Java 7 as an asynchronous, non-blocking IO approach.

The core concept of AIO is to initiate non-blocking I/O operations, respond immediately with no immediate results, and notify when I/O operations are complete.

This article focuses on some of the asynchronous channel APIS of NIO 2, and subsequent articles will examine other features of NIO.2

Asynchronous channel API

Starting with Java 7, the java.nio.channel package has four new asynchronous channels:

• AsynchronousSocketChannel
• AsynchronousServerSocketChannel
• AsynchronousFileChannel
• AsynchronousDatagramChannel

These classes are style-like to the NIO channel API in that they share the same method and parameter structures, and most of the parameters available to the NIO channel classes are still available to the new asynchronous version.

The asynchronous channel API provides two mechanisms for monitoring and controlling initiated asynchronous operations:

• The first by returning a Java. Util. Concurrent. The Future object to represent the results of the asynchronous operation
• The second is by passed to a new class of object operation java.nio.channels.Com pletionHandler to complete, it will define the operation after the completion of the execution of handler method.

Future

Since Java 1.5, the Future interface has been introduced, which allows you to get the results of a task after it has been executed. In NIO 2, said the Future object asynchronous operations as a result, assuming that we will create a server to listen on a client connection, open AsynchronousServerSocketChannel and bind them to address similar to the ServerSocketChannel:

AsynchronousServerSocketChannel server 
  = AsynchronousServerSocketChannel.open().bind(null);
Copy the code

The bind() method takes a socket address as its argument, passing a Null address that automatically binds the socket to the local host address and uses a free temporary port, just like the traditional ServerSocket port 0, which also uses a temporary port randomly assigned by the operating system. Then call the server’s Accept () method:

Future<AsynchronousSocketChannel> future = server.accept();

Copy the code

When we call the Accept () method of ServerSocketChannel in NIO, it blocks until we receive an incoming connection from the client. But AsynchronousServerSocketChannel the accept () method will return immediately the Future object.

The Future object of generic type is the return type of operation, in the example above, it is a AsynchronousSocketChannel, but it can also be an Integer or a String, depending on the final return type of the operation.

We can use the Future object to query the state of the operation

future.isDone();
Copy the code

This API returns true if the underlying operation has completed. Note that completion may mean normal termination, exception, or cancellation in this case.

We can also explicitly check if the operation has been cancelled, and it returns true if the operation has been cancelled before it completes normally. As follows:

future.isCancelled();
Copy the code

The actual cancel operation is as follows:

future.cancel(true)
Copy the code

The cancel() method uses a Boolean flag to indicate whether the thread performing the accept can be interrupted.

To retrieve the result of the operation, we use the get() method, which blocks waiting for the return of the result:

AsynchronousSocketChannel client= future.get();
Copy the code

Alternatively, we can set the blocking time to 10s in this example:

AsynchronousSocketChannel worker = future.get(10, TimeUnit.SECONDS);
Copy the code

CompletionHandler

An alternative to using futures to handle operations is to use the callback mechanism of the CompletionHandler class. Asynchronous channels allow you to specify completion handlers to use the result of the operation:

AsynchronousServerSocketChannel listener
  = AsynchronousServerSocketChannel.open().bind(null);
 
listener.accept(
  attachment, new CompletionHandler<AsynchronousSocketChannel, Object>() {
    public void completed( AsynchronousSocketChannel client, Object attachment) {
          // do whatever with client
      }
    public void failed(Throwable exc, Object attachment) {
          // handle failure}});Copy the code

When the I/O operation completes successfully, the completed callback API is invoked. If the operation fails, the failed API is called.

Asynchronous channel API instance

Server (with Future)

Here’s how to build the server using a Future.

public class AsyncEchoServer {
    private AsynchronousServerSocketChannel server;
    private Future<AsynchronousSocketChannel> future;
    private AsynchronousSocketChannel worker;

    public AsyncEchoServer(a) throws IOException, ExecutionException, InterruptedException {
        System.out.println("Open Server Channel");
        server = AsynchronousServerSocketChannel.open().bind(new InetSocketAddress("127.0.0.1".9090));
        future = server.accept();
    }

    public void runServer(a) throws ExecutionException, InterruptedException, IOException, TimeoutException {
        // Get the operation result
        worker = future.get();
        if(worker ! =null && worker.isOpen()) {
            ByteBuffer buffer = ByteBuffer.allocate(100);
            // Write data in the channel to the buffer
            worker.read(buffer).get(10,TimeUnit.SECONDS);
            System.out.println("received from client: " + new String(buffer.array()));
        }
        server.close();
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException, IOException, TimeoutException {
        AsyncEchoServer server = newAsyncEchoServer(); server.runServer(); }}Copy the code

Server (With CompletionHandler)

Now we’ll see how to implement the same server-side code using the CompletionHandler method instead of the Future method.

public class AsyncEchoServerWithCallBack {
    private AsynchronousServerSocketChannel server;
    private AsynchronousSocketChannel worker;
    private AsynchronousChannelGroup group;
    public AsyncEchoServerWithCallBack(a) throws IOException, ExecutionException, InterruptedException {
        System.out.println("Open Server Channel");
        group = AsynchronousChannelGroup.withFixedThreadPool(10, Executors.defaultThreadFactory());
        server = AsynchronousServerSocketChannel.open(group).bind(new InetSocketAddress("127.0.0.1".9090));
        The CompletionHandler interface is called when a new connection is established to implement the completed() method on the object
        server.accept(null.new CompletionHandler<AsynchronousSocketChannel, Object>() {
            @Override
            public void completed(AsynchronousSocketChannel result, Object attachment) {
                if (server.isOpen()) {
                    server.accept(null.this);
                }
                worker = result;
                if((worker ! =null) && (worker.isOpen())) {
                    ByteBuffer byteBuffer = ByteBuffer.allocate(100);
                    worker.read(byteBuffer);
                    System.out.println("received the client: "+newString(byteBuffer.array())); }}@Override
            public void failed(Throwable exc, Object attachment) {
                //TODO}}); }public static void main(String[] args) throws InterruptedException, ExecutionException, IOException, TimeoutException {
        AsyncEchoServerWithCallBack server = newAsyncEchoServerWithCallBack(); }}Copy the code

The Completed () method in the CompletionHandler interface implementation object is called when a new connection is established, and the failed method is called when an error occurs.

The first argument to the Accept method can be an object of any type, called an “attached object” at the time of the call. The attachment object is passed in when the Accept () method is called and can be retrieved from the completed and Failed methods’ arguments (attachment) in the implementation object of the CompletionHandler interface so that data can be passed. Methods that use the CompletionHandler interface all support passing data using attachment objects.

AsynchronousChannelGroup class

There can be an AsynchronousChannelGroup class to handle I/O requests. Objects of this class represent groups of asynchronous channels, each with a thread pool. Need to use AsynchronousChannelGroup withFixedThreadPool static factory methods of a class, or withCachedThreadPool withThreaPool set the thread pool. The threads in this thread pool are used to handle I/O events. Multiple asynchronous channels can share a group of thread pool resources.

Call AsynchronousSocketChannel and AsynchronousServerSocketChannel open the open method of asynchronous socket channel, can pass in a AsynchronousChannelGroup class object. If there is no object of the class java.asynchronouschannelGroup called, the default is to use the system provided groups. The threads in the thread pool corresponding to the system groups are daemons. If you use the default groups, the program will exit soon after startup. Because the daemons used by system groups do not prevent the virtual machine from exiting.

The client

public class AsyncEchoClient {
    private AsynchronousSocketChannel client;
    private Future<Void> future;

    public AsyncEchoClient(a) throws IOException {
        // Open an asynchronous channel
        System.out.println("Open client channel");
        client = AsynchronousSocketChannel.open();
        // Connect to the local port and address, and return nothing after a successful connection. However, we can still use Future objects to monitor the status of asynchronous operations
        System.out.println("Connect to server");
        future = client.connect(new InetSocketAddress("127.0.0.1".9090));
    }

    /** * Sends a message to the server **@param message
     * @return* /
    public void sendMessage(String message) throws ExecutionException, InterruptedException {
        if(! future.isDone()) { future.cancel(true);
            return;
        }
        // Encapsulate a byte array into a ByteBuffer
        ByteBuffer byteBuffer = ByteBuffer.wrap(message.getBytes());
        System.out.println("Sending message to the server");
        // Write data to the channel
        int numberBytes = client.write(byteBuffer).get();
        byteBuffer.clear();
    }

    public static void main(String[] args) throws IOException, ExecutionException, InterruptedException {
        AsyncEchoClient client = new AsyncEchoClient();
        client.sendMessage("hello world"); }}Copy the code

The test results

Client:

Open client channel
Connect to server
Copy the code

Server:

Open Server Channel
received the client: hello world
Copy the code

Java NIO 2 asynchronous IO

As we all know, NIO 2.0 with JDK 1.7 added asynchronous socket channels, which are truly asynchronous IO, passing variables during asynchronous IO operations and calling back related methods when the operation is complete. So how does the asynchronous non-blocking nature of NIO 2 work? A lot of detail can be gleaned from the previous description:

Asynchronous embodiment

AsynchronousServerSocketChannel, for example, when the object of the class the accept () method, which returns a Future < AsynchronousSocketChannel > object, Call accept () method is like calling ServerSocket in traditional I/O the accept (), are essentially receive client connection requests, just AsynchronousServerSocketChannel objects have no block waiting, The Future object is the result of the asynchronous operation. We can also use the isDone method of the Future to query the status of the completed operation. This is asynchronous.

The completed() method in the CompletionHandler interface implementation object is called back when a new connection is created, and the failed method is called when an error occurs.

Non-blocking embodiment

When calling AsynchronousServerSocketChannel object the accept () method, after the return to the Future object, the thread can go on to other things, it is blocked, in order to get the results, Future isDone method is invoked by the query operation is finished, Use get() to get the result, typically a non-blocking operation. In the traditional I/O model, the Accept method of a socket object blocks until a new connection comes in.

summary

NIO.2, also known as AIO, also helps us understand asynchronous IO operations better by understanding its asynchronous channel API. When we look at the NIO2 API, we can also look at the CHANNEL API in NIO, and there are many similarities.

References & acknowledgements

• Getting started with NIO.2, Part 1 asynchronous channel API
• A Guide to NIO2 Asynchronous Socket Channel
• Understanding Java 7 in Depth: Core Technologies and Best Practices