BIO and NIO are all synchronous IO.
If the application does not keep polling to ask operating system, the operating system will never actively notify the ready of data to application.
AIO means Asynchronous IO.
It adopts the subscription-notification mode.
The application registers the IO listener with the operating system and then continues to do its own thing.
When an IO event occurs in the operating system and data is ready, the application is actively notified.
What You Need
- About 6 minutes
- A favorite text editor or IDE
- Java 8 or later
1. AIO Socket Example
Below is an example of client/server mode by using AsynchronousServerSocketChannel :
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.sql.Timestamp;
import java.time.Instant;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ClientServerModeWithCustomThreadPool {
public static void main(String[] args) throws InterruptedException {
final InetSocketAddress serverAddress = new InetSocketAddress("localhost", 8383);
Runnable server = () -> {
try {
ExecutorService threadPool = Executors.newFixedThreadPool(2, r -> {
Timestamp ts = Timestamp.from(Instant.now());
Thread t = new Thread(r);
t.setName("MyThread-" + ts.getTime());
return t;
});
AsynchronousChannelGroup group = AsynchronousChannelGroup.withThreadPool(threadPool);
try (AsynchronousServerSocketChannel assChannel = AsynchronousServerSocketChannel.open(group)
.bind(serverAddress)) {
assChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Void>() {
@Override
public void completed(AsynchronousSocketChannel asChannel, Void attachment) {
assChannel.accept(null, this);
ByteBuffer buffer = ByteBuffer.allocate(1024);
StringBuilder sb = new StringBuilder();
try {
while (asChannel.read(buffer).get() > 0) {
sb.append(new String(buffer.array(), 0, buffer.position()));
buffer.clear();
}
} catch (Exception e) {
e.printStackTrace();
}
System.out.println(
"Server (" + Thread.currentThread().getName() + ") receives : " + sb.toString());
System.out.println();
}
@Override
public void failed(Throwable exc, Void attachment) {
}
});
Thread.currentThread().join();
}
} catch (Exception e) {
e.printStackTrace();
}
};
new Thread(server).start();
Runnable client = () -> {
try (BufferedReader reader = new BufferedReader(new InputStreamReader(System.in))) {
while (true) {
String line = reader.readLine();
try (AsynchronousSocketChannel asChannel = AsynchronousSocketChannel.open()) {
asChannel.connect(serverAddress).get();
System.out.print("Client (" + Thread.currentThread().getName() + ") sends : " + line);
System.out.println();
ByteBuffer buffer = ByteBuffer.wrap(line.getBytes());
asChannel.write(buffer);
}
}
} catch (Exception e) {
e.printStackTrace();
}
};
new Thread(client).start();
}
}
In above code snippet, there is one thread representing client and another thread representing server.
After executing, it waits for us to input messages.
Below is the output after inputting some messages :
hello
Client (Thread-1) sends : hello
Server (MyThread-1738180824500) receives : hello
world
Client (Thread-1) sends : world
Server (MyThread-1738180825880) receives : world
2. N(A)IO Framework – Netty
Netty is a NIO client server framework which enables quick and easy development of network applications such as protocol servers and clients.
It greatly simplifies and streamlines network programming such as TCP and UDP socket server.
2.1 Netty Features
- Netty is non-blocking : When client sends a message to server, that call immediately returns and client code can move further. Client can provide Callback-style handler which will handle response whenever received from server;
- Netty is event based : Once client server connection is established, communication happens in form of events. Channel handlers can be coded to handle these events and process them.
2.2 Netty Architecture
As shown in the picture below, when a client sends a request :
- BossGroup is mainly responsible for the Accept connection;
- WorkerGroup is responsible for distributing tasks to different handlers;
- Each handler handles specific business logic;
- NioEventLoop is similar to the selector in NIO.
It should be noted that almost all operations in netty are asynchronous.
2.3 Netty Client Server Mode Example
In netty, to create a client server mode application, we need below components :
- ServerBootStrap : to bootstrap server channel and start server. It binds the server to the port on which it will listen for connection;
- BootStrap : to bootstrap client channel;
- EventLoop : to handle all the IO operations for a channel once registered;
- EventLoopGroup : Group of EventLoop, ServerBootStrap needs boss group and worker group, ClientBootStrap only needs worker group;
- Boss EventLoopGroup : to look for and to accept incoming connections;
- Worker EventLoopGroup : to handle all the events during the communication through that connection;
- NioServerSocketChannel and NioSocketChannel : to configure server to use NIO selector based implementation to accept new connections;
- Decoder and Encoder : netty communication occurs over network socket channel which is in byte format. So in case we want to send specific data types, we need to provide encoder and decoder to encode data type into bytes and decode bytes to data type.
Below diagram traces the communication path of a message from client to server :
Below is an example of client sending a message to server and receiving a response message from server :
package com.example;
import java.nio.charset.Charset;
import java.util.List;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.ReplayingDecoder;
public class ClientServerModeWithNetty {
public static void main(String[] args) {
final int port = 8080;
final String host = "localhost";
Runnable server = () -> {
final String tName = Thread.currentThread().getName();
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup);
b.channel(NioServerSocketChannel.class);
b.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new MessageEncoder());
p.addLast(new MessageDecoder());
p.addLast(new ServerHandler(tName));
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
}
};
new Thread(server).start();
Runnable client = () -> {
final String clientName = Thread.currentThread().getName();
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(workerGroup);
b.channel(NioSocketChannel.class);
b.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ChannelPipeline p = ch.pipeline();
p.addLast(new MessageEncoder());
p.addLast(new MessageDecoder());
p.addLast(new ClientHandler(clientName));
}
});
ChannelFuture f = b.connect(host, port).sync();
Channel channel = f.sync().channel();
channel.writeAndFlush(new Message(1, "hello server"));
channel.flush();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
workerGroup.shutdownGracefully();
}
};
new Thread(client).start();
}
private static class Message {
private int id;
private String content;
public Message() {
}
public Message(int id, String content) {
this.id = id;
this.content = content;
}
public int getId() {
return id;
}
public void setId(int id) {
this.id = id;
}
public String getContent() {
return content;
}
public void setContent(String content) {
this.content = content;
}
@Override
public String toString() {
return "Message [id = " + id + ", content = " + content + "]";
}
}
private static class MessageDecoder extends ReplayingDecoder<Message> {
@Override
protected void decode(ChannelHandlerContext ctx,
ByteBuf in, List<Object> out) throws Exception {
Message message = new Message();
message.setId(in.readInt());
int len = in.readInt();
message.setContent(in.readCharSequence(len, Charset.forName("UTF-8")).toString());
out.add(message);
}
}
private static class MessageEncoder extends MessageToByteEncoder<Message> {
@Override
protected void encode(ChannelHandlerContext ctx,
Message message, ByteBuf out) throws Exception {
out.writeInt(message.getId());
out.writeInt(message.getContent().length());
out.writeCharSequence(message.getContent(), Charset.forName("UTF-8"));
}
}
private static class ClientHandler extends ChannelInboundHandlerAdapter {
private String clientName;
ClientHandler(String clientName) {
this.clientName = clientName;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message)
throws Exception {
String tName = Thread.currentThread().getName();
System.out.println(tName + " : Client (" + this.clientName + ") receives response = " + (Message) message);
}
}
private static class ServerHandler extends ChannelInboundHandlerAdapter {
private String serverName;
ServerHandler(String serverName) {
this.serverName = serverName;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object message)
throws Exception {
String tName = Thread.currentThread().getName();
System.out.println(tName + " : Server (" + this.serverName + ") receives request = " + (Message) message);
ctx.channel().writeAndFlush(new Message(2, "hello client"));
}
}
}
The output of above code snippet is below :
nioEventLoopGroup-4-1 : Server (Thread-0) receives request = Message [id = 1, content = hello server]
nioEventLoopGroup-3-1 : Client (Thread-1) receives response = Message [id = 2, content = hello client]