IO, NIO, AIO in Java:
BIO: prior to Java1.4, we used BIO to establish network connections. It was synchronous blocking IO. By default, when a request comes in, there is a thread dedicated to receiving it. Therefore, when the client makes a request to the server, it will ask if there are idle threads to receive it, and if not, it will wait or reject it. When the amount of concurrency is small, it is acceptable, but when the number of requests is large, many threads will be generated. In Java, multi-threaded context switch will consume limited resources and performance of the computer, resulting in resource waste.
NIO: NIO was developed to solve the problem of high concurrency in rebio. It features the ability to manage all connections in a single thread. As shown below:
NIO is buffer-oriented, meaning that data is written and written through Channel — Buffer. (Two-way flow)
AIO: Unlike the previous two IO models, AIO is an asynchronous non-blocking model. When reading or writing, only the API’s read and write methods are called, both of which are asynchronous. For read methods, when there is a stream to read, the operating system passes the readable stream into the buffer of the read method and notifies the application. For write operations, the operating system actively notifies the application when the stream passed by the write method has been written. In other words, when the API is called, the operating system calls the callback function when it is done.
Summary: General IO is divided into synchronous blocking model (BIO), synchronous non-blocking model (NIO), asynchronous blocking model, asynchronous non-blocking model (AIO)
The synchronous blocking model refers to the fact that when an IO operation is called, it must wait until its IO operation finishes
The synchronous non-blocking model means that when you invoke an IO operation, you don’t have to wait to do something else, but you must constantly ask if the IO operation has completed.
In the asynchronous blocking model, after an application invokes an I/O operation, the OPERATING system completes the I/O operation, but the application must wait or ask the operating system whether the OPERATION is complete.
Asynchronous non-blocking means that after an application invokes AN I/O operation, the operating system completes the I/O operation and invokes the callback function.
NIO small Demo server
First, take a look at the server code in general
Public Class ServerHandle implements Runnable{// Constructor with parameters Public ServerHandle(int port){} // Stop method public voidshop(){} private void write(SocketChannel SocketChannel, String Response)throws IOException{} private void handleInput(SelectionKey key) throws IOException{} // The server runs the principal method @override public voidrun() {}}Copy the codeFirst let’s look at the implementation of the constructor on the server side:
Public ServerHandle(int port){try {// Create selector = selector. Open (); ServerSocketChannel = serverSocketChannel.open (); / / set to non-blocking mode serverSocketChannel. ConfigureBlocking (false); / / incoming port, and set the connection queue 1024 serverSocketChannel maximum. The socket (), bind (new InetSocketAddress (port), 1024); / / to monitor the client request serverSocketChannel. Register (selector, SelectionKey OP_ACCEPT); // Mark the start flag started =true;
System.out.println("Server started with port number:"+ port); } catch (IOException e){ e.printStackTrace(); System.exit(1); }}Copy the codeHere you create a selector and a listening channel, register the listening channel with the selector and select the event it is interested in (Accept). All subsequent connections will come in through this listening channel.
Then there is the implementation of the write method:
private void doWrite(SocketChannel channel, String response) throws IOException { byte[] bytes = response.getBytes(); ByteBuffer wirteBuffer = ByteBuffer.allocate(bytes.length); wirteBuffer.put(bytes); // Change the write mode to read mode wirtebuffer.flip (); // Write channel. Write (wirteBuffer); }Copy the codeThe second is what happens to links when they are passed in by events
Private void handleInput(SelectionKey key) throws IOException{// When the key is availableif (key.isValid()){
if(key.isacceptable ()){// Returns the channel created by this key. ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel(); Through the channel for a link in a SocketChannel SocketChannel = serverSocketChannel. The accept (); socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
}
if(key.isreadable ()){// Returns the channel created by this key. SocketChannel socketChannel = (SocketChannel) key.channel(); ByteBuffer byteBuffer = ByteBuffer.allocate(1024); intreadBytes = socketChannel.read(byteBuffer);
if (readBytes > 0){
byteBuffer.flip();
byte[] bytes = new byte[byteBuffer.remaining()];
byteBuffer.get(bytes);
String expression = new String(bytes, "UTF-8");
System.out.println("Message received by server:"+ expression); // To distinguish whether the data printed on the workbench is generated by the client side or the server sidedoWrite(socketChannel, "+ + + + +" + expression + "+ + + + +");
} else if(readBytes == 0){// No data, ignore}else if (readBytes < 0){// The resource closes key.cancel(); socketChannel.close(); }}}}Copy the codeThe point here is that whenever a ServerSocketChannel and SocketChannel register certain events with a Selector, the Selector monitors whether those events occur. If there is a serverSocketChannel in the constructor that registers the Accept event. When it is ready, it can access the ready channel in the selected keyset by calling the selector selectorKeys() method.
Final method:
@Override
public void run() {// loop throughwhile(started) {try {// Blocks selector. Select () when there is no ready event; Set<SelectionKey> keys = selection.selectedkeys (); Iterator<SelectionKey> iterator = keys.iterator(); SelectionKey key;while(iterator.hasNext()){ key = iterator.next(); Iterator.remove (); iterator.remove(); // Handle the ready channel, as described above handleInput(key); } catch (Exception e){if(key ! = null){ key.cancel();if(key.channel() ! = null) { key.channel().close(); } } } } }catch (Throwable throwable){ throwable.printStackTrace(); }}}Copy the codeThis method is the main method on the server side. The general process is as follows:
- Open ServerSocketChannel and listen for client connections
- Bind the listening port and set the connection to non-blocking mode (blocking mode cannot register with the selector)
- Create a Reactor thread, create a selector, and start the thread
- Registers the ServerSocketChannel with the Selector in the Reactor thread and listens for ACCEPT events
- Selector polls for ready keys
- Selector listens for new client access, processes new access requests, completes TCP three-way handshake, and resumes physical links
- Example Set the client link to non-blocking mode
- The new client connection is registered to the Reactor thread’s Selector, the read operation is monitored, and the network messages sent by the client are asynchronously read to the buffer
- The call to write sends the message asynchronously to the client
NIO small Demo client
Public class ClientHandle implements Runnable{// implements Runnable Public ClientHandle(String IP, Int port){} // Ready channel Private void handleInput(SelectionKey key) throws IOException{} // Write method (the same as the write method on the server) private voiddoWrite(SocketChannel Channel,String Request) throws IOException{} // Connect to the server private voiddoConnect() throws IOException{} // Send information public void sendMsg(String MSG) throws Exception{}}Copy the codeLet’s look at the constructor implementation first:
public ClientHandle(String ip,int port) { this.host = ip; this.port = port; Try {// create selector = selector. Open (); SocketChannel = socketchannel.open (); / / if it istrue, the channel will be placed in blocking mode; If it isfalseThis channel will be placed in a non-blocking mode socketChannel. ConfigureBlocking (false);
started = true; }catch(IOException e){ e.printStackTrace(); System.exit(1); }}Copy the codeLet’s look at what happens to ready channels:
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
SocketChannel sc = (SocketChannel) key.channel();
if(key.isConnectable()){doConnect method)if(sc.finishConnect()){
System.out.println("Connected event");
}
else{ System.exit(1); }} // Read the messageif(key.isreadable ()){// Create ByteBuffer and allocate a 1K buffer. // Read request code stream, return the number of bytes read intreadBytes = sc.read(buffer); // Read bytes, encode and decode bytesif(readBytes>0){ buffer.flip(); Byte [] bytes = new byte[buffer.remaining()]; byte[] bytes = new byte[buffer.remaining()]; Buffer.get (bytes); // Copy the buffer readable byte array into the new array. String result = new String(bytes,"UTF-8");
System.out.println("Client receives message:" + result);
}lse if(readBytes==0){// Ignore}else if(readBytes<0){// The link is down, release the resource key.cancel(); sc.close(); }}}}Copy the codeBefore using the run method, take a look at its implementation:
private void doConnect() throws IOException{
if(socketChannel.connect(new InetSocketAddress(host,port))){
System.out.println("connect");
}
else {
socketChannel.register(selector, SelectionKey.OP_CONNECT);
System.out.println("register"); }}Copy the codeWhen SocketChannel is operating in non-blocking mode, a call to connect() returns true if the connection was successfully established (such as localhost, which immediately established the connection) and false otherwise.
In non-blocking mode, after returning false, the finishConnect() method must be called somewhere afterwards to complete the connection. When SocketChannel is in blocking mode, connect() is blocked until the connection is established or an I/O error occurs.
So the code returns false (but still useful) after the Connect server and registers the channel with the selector and selects the Connect event in the else statement.
The client’s run method:
@Override
public void run() {
try{
doConnect(); }catch(IOException e){ e.printStackTrace(); System.exit(1); } // loop over selectorwhile(started){
try{
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key ;
while(it.hasNext()){
key = it.next();
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key ! = null){ key.cancel();if(key.channel() ! = null){ key.channel().close(); } } } } }catch(Exception e){ e.printStackTrace(); System.exit(1); } // Selector is automatically released when closedif(selector ! = null){ try{ selector.close(); }catch (Exception e) { e.printStackTrace(); }}}Copy the codeSending messages to the server:
Public void sendMsg(String MSG) throws Exception{// Overwrite the previously interested event (connect), Change it to OP_READ socketchannel. register(selector, selectionkey.op_read);doWrite(socketChannel, msg);
}
Copy the codeComplete code:
Server:
/**
* Created by innoyiya on 2018/8/20.
*/
public class Service {
private static int DEFAULT_POST = 12345;
private static ServerHandle serverHandle;
public static void start(){
start(DEFAULT_POST);
}
public static synchronized void start(int post) {
if(serverHandle ! = null){ serverHandle.shop(); } serverHandle = new ServerHandle(post); new Thread(serverHandle,"server").start(); }}Copy the codeServer principal:
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.Iterator; import java.util.Set; /** * Created by innoyiya on 2018/8/20. */ public class ServerHandle implements Runnable{ private Selector selector; private ServerSocketChannel serverSocketChannel; private volatile boolean started; Public ServerHandle(int port){try {// Create selector = selector. Open (); ServerSocketChannel = serverSocketChannel.open (); / / set to non-blocking mode serverSocketChannel. ConfigureBlocking (false); , according to port / / set the connection queue 1024 serverSocketChannel maximum. The socket (), bind (new InetSocketAddress (port), 1024); / / to monitor the client request serverSocketChannel. Register (selector, SelectionKey OP_ACCEPT); // Mark the start flag started =true;
System.out.println("Server started with port number:" + port);
} catch (IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void shop(){
started = false;
}
private void doWrite(SocketChannel channel, String response) throws IOException {
byte[] bytes = response.getBytes();
ByteBuffer wirteBuffer = ByteBuffer.allocate(bytes.length);
wirteBuffer.put(bytes);
wirteBuffer.flip();
channel.write(wirteBuffer);
}
private void handleInput(SelectionKey key) throws IOException{
if (key.isValid()){
if (key.isAcceptable()){
ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
SocketChannel socketChannel = serverSocketChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
}
if (key.isReadable()){
SocketChannel socketChannel = (SocketChannel) key.channel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
int readBytes = socketChannel.read(byteBuffer);
if (readBytes > 0){
byteBuffer.flip();
byte[] bytes = new byte[byteBuffer.remaining()];
byteBuffer.get(bytes);
String expression = new String(bytes, "UTF-8");
System.out.println("Message received by server:" + expression);
doWrite(socketChannel, "+ + + + +" + expression + "+ + + + +");
} else if (readBytes < 0){
key.cancel();
socketChannel.close();
}
}
}
}
@Override
public void run() {// loop throughwhile (started) {
try {
selector.select();
//System.out.println(selector.select());
Set<SelectionKey> keys = selector.selectedKeys();
//System.out.println(keys.size());
Iterator<SelectionKey> iterator = keys.iterator();
SelectionKey key;
while (iterator.hasNext()){
key = iterator.next();
iterator.remove();
try {
handleInput(key);
} catch (Exception e){
if(key ! = null){ key.cancel();if(key.channel() ! = null) { key.channel().close(); } } } } }catch (Throwable throwable){ throwable.printStackTrace(); }}}}Copy the codeClient:
/**
* Created by innoyiya on 2018/8/20.
*/
public class Client {
private static String DEFAULT_HOST = "localhost";
private static int DEFAULT_PORT = 12345;
private static ClientHandle clientHandle;
private static final String EXIT = "exit";
public static void start() {
start(DEFAULT_HOST, DEFAULT_PORT);
}
public static synchronized void start(String ip, int port) {
if(clientHandle ! = null){ clientHandle.stop(); } clientHandle = new ClientHandle(ip, port); new Thread(clientHandle,"Server").start(); } // Send messages to the server. Public static Boolean sendMsg(String MSG) throws Exception {if (msg.equals(EXIT)){
return false;
}
clientHandle.sendMsg(msg);
return true; }}Copy the codeClient body code:
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.SocketChannel; import java.util.Iterator; import java.util.Set; /** * Created by innoyiya on 2018/8/20. */ public class ClientHandle implements Runnable{ private String host; private int port; private Selector selector; private SocketChannel socketChannel; private volatile boolean started; public ClientHandle(String ip,int port) { this.host = ip; this.port = port; Try {// create selector = selector. Open (); SocketChannel = socketchannel.open (); / / if it istrue, the channel will be placed in blocking mode; If it isfalseThis channel will be placed in a non-blocking mode socketChannel. ConfigureBlocking (false);
started = true;
}catch(IOException e){
e.printStackTrace();
System.exit(1);
}
}
public void stop(){
started = false;
}
private void handleInput(SelectionKey key) throws IOException{
if(key.isValid()){
SocketChannel sc = (SocketChannel) key.channel();
if(key.isConnectable()){
if(sc.finishConnect()){
System.out.println("Connected event");
}
else{ System.exit(1); }} // Read the messageif(key.isreadable ()){// Create ByteBuffer, and create a buffer of 1M. // Read request code stream, return the number of bytes read intreadBytes = sc.read(buffer); // Read bytes, encode and decode bytesif(readBytes>0){// will buffer the currentlimitSet to position=0 for subsequent buffer reads buffer.flip(); Byte [] bytes = new byte[buffer.remaining()]; byte[] bytes = new byte[buffer.remaining()]; Buffer.get (bytes); // Copy the buffer readable byte array into the new array. String result = new String(bytes,"UTF-8");
System.out.println("Client receives message:" + result);
} else if(readBytes<0){ key.cancel(); sc.close(); }}}} // Send messages asynchronously private voiddoWrite(SocketChannel channel,String request) throws IOException{ byte[] bytes = request.getBytes(); ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length); writeBuffer.put(bytes); / / the flip operation writeBuffer. Flip (); // Send the buffer's byte array channel.write(writeBuffer); } private voiddoConnect() throws IOException{
if(socketChannel.connect(new InetSocketAddress(host,port))){
System.out.println("connect");
}
else {
socketChannel.register(selector, SelectionKey.OP_CONNECT);
System.out.println("register"); }} public void sendMsg(String MSG) throws Exception{// Overwrite the previously interested event, Change it to OP_READ socketchannel. register(selector, selectionkey.op_read);doWrite(socketChannel, msg);
}
@Override
public void run() {
try{
doConnect(); }catch(IOException e){ e.printStackTrace(); System.exit(1); } // loop over selectorwhile(started){
try{
selector.select();
Set<SelectionKey> keys = selector.selectedKeys();
Iterator<SelectionKey> it = keys.iterator();
SelectionKey key ;
while(it.hasNext()){
key = it.next();
it.remove();
try{
handleInput(key);
}catch(Exception e){
if(key ! = null){ key.cancel();if(key.channel() ! = null){ key.channel().close(); } } } } }catch(Exception e){ e.printStackTrace(); System.exit(1); } // Selector is automatically released when closedif(selector ! = null){ try{ selector.close(); }catch (Exception e) { e.printStackTrace(); }}}}Copy the codeThe test class:
import java.util.Scanner; /** * Created by innoyiya on 2018/8/20. */ public class Test { public static void main(String[] args) throws Exception { Service.start(); Thread.sleep(1000); Client.start();while(Client.sendMsg(new Scanner(System.in).nextLine())); }}Copy the codeConsole printing:
1234 Message received by the server: 1234 Message received by the client: +++++1234+++++ 5678 Message received by the server: 5678 Message received by the client: +++++5678+++++Copy the codeIf anything goes wrong, please let me know.