Related posts :
Java networking : UDP vs TCP
1. Intro
From wiki : https://en.wikipedia.org/wiki/Non-blocking_I/O_(Java)
Non-blocking I/O (usually called NIO, and sometimes called "New I/O") is a collection of Java programming language APIs that offer features for intensive I/O operations.
Main elements :
- NIO data transfer is based on buffers .
- Channels are designed to provide for bulk data transfers to and from NIO buffers.
- A selector provides a mechanism for waiting on channels and recognizing when one or more become available for data transfer.
2. NIO server
Main concept of NIO : we can have one server thread which can work with multiple client threads in non-blocking mode. This is implemented by selectors, which have to be used of both sides : client and server. Server is "listening" the channel using selector :
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
After that we can check what exactly clients wants from server :
SelectionKey key = keys.next();
...
if (key.isConnectable()){ // client want to connect to server
if (key.isWritable()){ // client want server to write something to channel
if (key.isReadable()){ // client want to read from server
After understanding what exactly client wants from server (connect, read or write), we can get client channel and proceed with that operation :
SocketChannel channel = (SocketChannel) key.channel();
...
channel.accept();
..
ByteBuffer readBuffer = ByteBuffer.allocate(1000);
channel.read(readBuffer);
....
channel.write(ByteBuffer.wrap(data));
Also, as it was mentioned before - data transfer is based on buffers.
Full code of server :
package com.demien.networking;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;
public class NioServer implements Runnable {
public final static String ADDRESS = "127.0.0.1";
public final static int PORT = 6666;
public final static long TIMEOUT = 10000;
private ServerSocketChannel serverChannel;
private Selector selector;
/** * This hashmap is important. It keeps track of the data that will be written to the clients. * This is needed because we read/write asynchronously and we might be reading while the server * wants to write. In other words, we tell the Selector we are ready to write (SelectionKey.OP_WRITE) * and when we get a key for writting, we then write from the Hashmap. The write() method explains this further. */ private Map<SocketChannel, byte[]> clientMessages = new HashMap<SocketChannel, byte[]>();
public static void main(String args[]) {
Thread serverThread = new Thread(new NioServer());
serverThread.run();
}
public NioServer() {
init();
}
private void init() {
System.out.println("initializing server");
// We do not want to call init() twice and recreate the selector or the serverChannel. if (selector != null) return;
if (serverChannel != null) return;
try {
// This is how you open a Selector selector = Selector.open();
// 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(ADDRESS, PORT));
/** * 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);
} catch (IOException e) {
e.printStackTrace();
}
}
@Override public void run() {
System.out.println("Now accepting connections...");
try {
// A 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 undefinitely. */ 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);
}
}
}
} catch (IOException e) {
e.printStackTrace();
} finally {
closeConnection();
}
}
/** * We registered this channel in the Selector. This means that the SocketChannel we are receiving * back from the key.channel() is the same channel that was used to register the selector in the accept() * method. Again, I am just explaning as if things are synchronous to make things easy to understand. * This means that later, we might register to write from the read() method (for example). */ private void write(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
writeToChannel(channel);
key.interestOps(SelectionKey.OP_READ);
}
private void writeToChannel(SocketChannel channel) throws IOException {
/** * The hashmap contains the object SockenChannel along with the information in it to be written. * In this example, we send the "Hello from server" String and also an echo back to the client. * This is what the hashmap is for, to keep track of the messages to be written and their socketChannels. */ byte[] data = clientMessages.get(channel);
if (data != null) {
clientMessages.remove(channel);
// Something to notice here is that reads and writes in NIO go directly to the channel and in form of // a buffer. channel.write(ByteBuffer.wrap(data));
}
// Since we wrote, then we should register to read next, since that is the most logical thing // to happen next. YOU DO NOT HAVE TO DO THIS. But I am doing it for the purpose of the example // Usually if you register once for a read/write/connect/accept, you never have to register again for that unless you // register for none (0). Like it said, I am doing it here for the purpose of the example. The same goes for all others.
}
// Nothing special, just closing our selector and socket. private void closeConnection() {
System.out.println("Closing server down");
if (selector != null) {
try {
selector.close();
serverChannel.socket().close();
serverChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
/** * Since we are accepting, we must instantiate a serverSocketChannel by calling key.channel(). * We use this in order to get a socketChannel (which is like a socket in I/O) by calling * serverSocketChannel.accept() and we register that channel to the selector to listen * to a WRITE OPERATION. I do this because my server sends a hello message to each * client that connects to it. This doesn't mean that I will write right NOW. It means that I * told the selector that I am ready to write and that next time Selector.select() gets called * it should give me a key with isWritable(). More on this in the write() method. */ private void accept(SelectionKey key) throws IOException {
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_WRITE);
byte[] hello = new String("Hello from server").getBytes();
clientMessages.put(socketChannel, hello);
}
/** * We read data from the channel. In this case, my server works as an echo, so it calls the echo() method. * The echo() method, sets the server in the WRITE OPERATION. When the while loop in run() happens again, * one of those keys from Selector.select() will be key.isWritable() and this is where the actual * write will happen by calling the write() method. */ private void read(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1024);
readBuffer.clear();
int read;
try {
read = channel.read(readBuffer);
} catch (IOException e) {
System.out.println("Reading problem, closing connection");
key.cancel();
channel.close();
return;
}
if (read == -1) {
System.out.println("Nothing was there to be read, closing connection");
channel.close();
key.cancel();
return;
}
// IMPORTANT - don't forget the flip() the buffer. It is like a reset without clearing it. readBuffer.flip();
byte[] data = new byte[1000];
readBuffer.get(data, 0, read);
System.out.println("Received: " + extractString(data));
echo(key, data);
}
private void echo(SelectionKey key, byte[] data) throws IOException {
SocketChannel socketChannel = (SocketChannel) key.channel();
clientMessages.put(socketChannel, data);
writeToChannel(socketChannel);
key.interestOps(SelectionKey.OP_WRITE);
}
public static String extractString(byte[] array) {
List<Byte> wrapper = new ArrayList<>();
for (int i = 0; i < array.length; i++) {
if (array[i] != 0) {
wrapper.add(array[i]);
} else {
break;
}
}
byte[] resultArray = new byte[wrapper.size()];
for (int i = 0; i < wrapper.size(); i++) {
resultArray[i] = wrapper.get(i);
}
return new String(resultArray);
}
}
3. NIO client
Client implementation is based on the same principals : register selector and wait for "events" to understand what to do : connect, read or write.
Full code of client :
package com.demien.networking;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
public class NioClient implements Runnable {
private Selector selector;
private BufferedReader stdIn = new BufferedReader( new InputStreamReader(System.in));
public static void main(String[] args) {
NioClient clientThread = new NioClient();
Thread thread = new Thread(clientThread);
thread.start();
}
@Override public void run() {
SocketChannel channel;
try {
selector = Selector.open();
channel = SocketChannel.open();
channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_CONNECT);
channel.connect(new InetSocketAddress(NioServer.ADDRESS, NioServer.PORT));
System.out.println("Starting....");
while (!Thread.interrupted()){
selector.select(1000);
Iterator<SelectionKey> keys = selector.selectedKeys().iterator();
while (keys.hasNext()){
SelectionKey key = keys.next();
keys.remove();
if (!key.isValid()) continue;
if (key.isConnectable()){
System.out.println("I am connected to the server");
connect(key);
}
if (key.isWritable()){
write(key);
}
if (key.isReadable()){
read(key);
}
}
}
} catch (Exception e1) {
e1.printStackTrace();
} finally {
close();
}
}
private void close(){
try {
selector.close();
} catch (IOException e) {
e.printStackTrace();
}
}
private void read (SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
ByteBuffer readBuffer = ByteBuffer.allocate(1000);
readBuffer.clear();
int length;
try{
length = channel.read(readBuffer);
} catch (IOException e){
System.out.println("Reading problem, closing connection");
key.cancel();
channel.close();
return;
}
if (length == -1){
System.out.println("Nothing was read from server");
channel.close();
key.cancel();
return;
}
readBuffer.flip();
byte[] buff = new byte[1024];
readBuffer.get(buff, 0, length);
System.out.println("Received: " + NioServer.extractString(buff));
key.interestOps(SelectionKey.OP_WRITE);
}
private void write(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
String userInput = stdIn.readLine();
channel.write(ByteBuffer.wrap(userInput.getBytes()));
// lets get ready to read. key.interestOps(SelectionKey.OP_READ);
}
private void connect(SelectionKey key) throws IOException {
SocketChannel channel = (SocketChannel) key.channel();
if (channel.isConnectionPending()){
channel.finishConnect();
}
channel.configureBlocking(false);
channel.register(selector, SelectionKey.OP_READ);
}
}
4. Running
After starting we can send messages to server by input (console) and watch "echo" output :
Starting....
I am connected to the server
Received: Hello from server
hello. i'm a client
Received: hello. i'm a client
how are you ?
Received: how are you ?
