Network I/O model concepts
There are several steps involved in a network read/write process, such as Socket communication between the client and the server:
- Client: First the client copies the user buffer’s data to the kernel buffer through a write system call; The kernel writes the data in the buffer to the nic and sends it to the server.
- Server: reads data from the nic and stores it in the kernel buffer. Copy the data from the kernel buffer to the user buffer for processing by calling the read function.
You can use the read and write functions if you don’t understand themman readorman writeLook at it.
Synchronous blocking IO (BIO)
When a read or write system call occurs, the user space is blocked until the corresponding read or write kernel space returns a result. In Java, the IO operations of the Socket and ServerSocket classes are typically blocking IO.
The general interaction of blocking IO is as follows:
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(9000);
while (true) {
Socket socket = serverSocket.accept();
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
PrintWriter writer = new PrintWriter(socket.getOutputStream());
String value = reader.readLine();
System.out.println("Message sent by client:"+ value); writer.write(value); writer.flush(); }}Copy the codeServersocket.accept (); The system call is blocked, and a detailed explanation of accept can be seen on Linux using man Accept (I’m only copying some of the information here) :
[root@izbp1hvx6s6h8yr3sgj333z ~]# man accept
ACCEPT(2) Linux Programmer's Manual ACCEPT(2)
NAME
accept, accept4 - accept a connection on a socket
DESCRIPTION
The accept() system call is used with connection-based socket types (SOCK_STREAM, SOCK_SEQPACKET). It extracts the first connection request on
the queue of pending connections for the listening socket, sockfd, creates a new connected socket, and returns a new file descriptor referring
to that socket. The newly created socket is not in the listening state. The original socket sockfd is unaffected by this call.
Copy the codeThe accept() system call is used with connection-based socket types (SOCK_STREAM, SOCK_SEQPACKET). It extracts the first connection request listening socket sockFD’s pending connection queue to create a new connection socket and returns a new file descriptor reference to that socket. The newly created socket is not listening. The raw socket sockFD is not affected by this call.
For BIO model is the biggest problem is, every time listening, speaking, reading and writing operation will cause obstruction to the user thread, the thread can’t do anything, can only wait, for we pursue high concurrency system has a lot of restrictions, based on this idea, we can use with multithreading and the thread pool technology in an asynchronous thought to the question, But at the same time, it brings new problems: the frequent context switching of multiple threads, limited by the number of operating system threads.
None Blocking IO
We found that the nature of the problem is blocking, and the advantage of synchronous non-blocking is that it can keep our user threads from blocking without using multithreaded asynchrony (note: for kernel space, there is still blocking, but it does not affect the user threads); If the read function has not yet reached the user’s buffer, it returns the read function without blocking. If the read function has not yet reached the user’s buffer, it returns the read function without blocking. If the read function is called and the data is already in the kernel buffer, the user buffer is copied, which blocks.
public static void main(String[] args) throws IOException {
ServerSocketChannel server = ServerSocketChannel.open();
server.bind(new InetSocketAddress(9000)).configureBlocking(false);
List<SocketChannel> list = new ArrayList<>();
while (true) {
SocketChannel socket = server.accept();
if(socket ! =null) {
socket.configureBlocking(false);
System.out.println("Client has been connected...");
list.add(socket);
}
Iterator<SocketChannel> iterator = list.iterator();
while (iterator.hasNext()) {
SocketChannel channel = iterator.next();
ByteBuffer buffer = ByteBuffer.allocate(32);
int read = channel.read(buffer);
if (read > 0) {
System.out.println("Received a message:" + new String(buffer.array()));
} else if(read == -1) {
System.out.println("Disconnect"); iterator.remove(); }}}}Copy the codeConfigureBlocking (false) is set above, so server.accept(); And the channel. The read (buffer); The user thread is constantly making IO system calls, polling to see if the data is ready, but the problem with this approach is that it incurs a lot of CPU overhead in empty polling, and it does not satisfy high concurrency.
Multiplexing IO
In an upgrade based on the non-blocking concept, clients do not need to call the read function in an endless loop, and do not need to determine whether the read data is copied to user space. Instead, they add an event listener for each Socket connection, and then execute the corresponding operation when the event is triggered. Like I’m going to read, but I don’t go directly to read, because I don’t know if there is any data ready, I’ll register a listener, to let the listener to listen to have read the complete, once completed data read, the listener will tell me the data that you can go to read, at this time my client to call the read function to fetch the data. This listener is a multiplexer, which can also bind multiple events simultaneously.
The following is an implementation of the multiplexing model in Java:
public static void main(String[] args) throws Exception {
ServerSocketChannel serverSocket = ServerSocketChannel.open();
serverSocket.socket().bind(new InetSocketAddress(9000));
serverSocket.configureBlocking(false);
// 1. This selector is the listener mentioned in the article, also known as the multiplexer
Selector selector = Selector.open();
Bind ServerSocket to selector and tell it to listen for the Accept event for me
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
while (true) {
// 3. Wait for the event to be triggered. If there is no event, it will block
selector.select();
Set<SelectionKey> selectionKeySet = selector.selectedKeys();
Iterator<SelectionKey> iterator = selectionKeySet.iterator();
while (iterator.hasNext()) {
SelectionKey key = iterator.next();
if (key.isAcceptable()) {
// If the accept event is triggered, a client is connected
ServerSocketChannel serverChannel = (ServerSocketChannel)key.channel();
SocketChannel socketChannel = serverChannel.accept();
socketChannel.configureBlocking(false);
socketChannel.register(selector, SelectionKey.OP_READ);
} else if (key.isReadable()) {
// If the read event is triggered, the kernel buffer contains data that can be read
SocketChannel socketChannel = (SocketChannel)key.channel();
ByteBuffer buffer = ByteBuffer.allocate(128);
int read = socketChannel.read(buffer);
if (read > 0) {
System.out.println("Data sent from client:" + new String(buffer.array()));
} else if (read == -1){
System.out.println("Client disconnects"); } } iterator.remove(); }}}Copy the codeThe multiplexing code is pretty much the same as synchronizing non-blocking code, but with an extra Selector object, which registers a Accept event for ServerSocketChannel and a read event for SocketChannel, This method relies on Selector. Select (). However, the underlying method of this method is to call the OS select/poll/epoll function. Two system calls are made to perform an IO operation: the first is a SELECT call and the second is a read call. That is multiplexing IO not necessarily higher than BIO performance, because itself multiplexing also blocking problems, but the fundamental problems of BIO is unable to support high concurrency, and in multiplexing IO can solve this problem, in other words, if my system is not high concurrent system directly using BIO is better, because it involves a system call. If you want to support high concurrency you can use the multiplexed IO model.
Asynchronous IO (AIO)
After a system call, a new thread will pass the data back through an event callback. Note that unlike multiplexing IO, multiplexing IO is a system callEvent listeners, AIO isEvent callback; Event listening means that when the event I care about is triggered, I handle it myself, and event callback means that when the event I care about is triggered, a new thread will pass the data to me through the callback method, so I don’t need to fetch it myself.
AIO support is also provided in the Java NIO package:
public static void main(String[] args) throws Exception {
final AsynchronousServerSocketChannel serverChannel =
AsynchronousServerSocketChannel.open().bind(new InetSocketAddress(9000));
serverChannel.accept(null.new CompletionHandler<AsynchronousSocketChannel, Object>() {
@Override
public void completed(AsynchronousSocketChannel socketChannel, Object attachment) {
try {
System.out.println("2 --"+Thread.currentThread().getName());
// If you do not write this line of code, the client connection will not connect to the server
serverChannel.accept(attachment, this);
System.out.println(socketChannel.getRemoteAddress());
ByteBuffer buffer = ByteBuffer.allocate(1024);
socketChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
@Override
public void completed(Integer result, ByteBuffer buffer) {
System.out.println("3--"+Thread.currentThread().getName());
buffer.flip();
System.out.println(new String(buffer.array(), 0, result));
socketChannel.write(ByteBuffer.wrap("HelloClient".getBytes()));
}
@Override
public void failed(Throwable exc, ByteBuffer buffer) { exc.printStackTrace(); }}); }catch(IOException e) { e.printStackTrace(); }}@Override
public void failed(Throwable exc, Object attachment) { exc.printStackTrace(); }}); System.out.println("1--"+Thread.currentThread().getName());
Thread.sleep(Integer.MAX_VALUE);
}
Copy the codeAIO programming is truly asynchronous, but in Linux it is done using epoll, so it is rarely used.
The underlying implementation of multiplexing
For more information on how to build openJDK, see this article at blog.csdn.net/qq_35559877…
The following is the openJDK source code analysis of niO is how to achieve multiplexing, which is a key code is the following 3 steps:
1.Selector selector = Selector.open();
2.serverSocket.register(selector, SelectionKey.OP_ACCEPT);
3.selector.select();
Copy the codeSo let’s take a lookSelector.open()Do what things, click into the source code. Found is called DefaultSelectorProvider. The create () method, and this class implements a version in the Windows and Linux, we find its source code to Linux version found in the class
Will eventually be transferred to the EPollSelectorProvider. OpenSeletor () method to create EPollSelectorImpl object
[root@izbp1hvx6s6h8yr3sgj333z ~]# man epoll_create EPOLL_CREATE(2) Linux Programmer's Manual EPOLL_CREATE(2) NAME epoll_create, epoll_create1 - open an epoll file descriptor SYNOPSIS #include <sys/epoll.h> int epoll_create(int size); int epoll_create1(int flags); DESCRIPTION epoll_create() creates an epoll(7) instance. Since Linux 2.6.8, the size argument is ignored, but must be greater than zero; see NOTES below. epoll_create() returns a file descriptor referring to the new epoll instance. This file descriptor is used for all the subsequent calls to the epoll interface. When no longer required, the file descriptor returned by epoll_create() should be closed by using close(2). When all file descriptors referring to an epoll instance have been closed, the kernel destroys the instance and releases the associated resources for reuse. epoll_create1() If flags is 0, then, other than the fact that the obsolete size argument is dropped, Epoll_create1 () is the same as epoll_create(). The follow‐ ing value can be included in flags to obtain different behavior: EPOLL_CLOEXEC Set the close-on-exec (FD_CLOEXEC) flag on the new file descriptor. See the description of the O_CLOEXEC flag in open(2) for reasons why this may be useful. RETURN VALUE On success, these system calls return a nonnegative file descriptor. On error, -1 is returned, and errno is set to indicate the error.Copy the codeSee here that Selector. Open () simply calls epoll_create() to create an epoll object; So this epoll object corresponds to the Selector object in Java. Read on:serverSocket.register(selector, SelectionKey.OP_ACCEPT);Point in, skip the dolls code, to its source code found in the Linux version of the implementation: EPollSelectorImpl. ImplRegister ()
selector.select();The bottom layer calls epollarrayWrapper.poll () and then updateRegistrations() :
In updateRegistrations(), a native method epollCtl() is called.
This system call performs control operations on the epoll(7) instance referred to by the file descriptor epfd. It requests that the operation
op be performed for the target file descriptor, fd.
Copy the codeTranslation:
The system call performs control operations on the epoll(7) instance referenced by the file descriptor EPfd. It requires the operation to perform op on the object file descriptor fd.
That is, the epoll_ctl function actually binds the channel to the op of interest, followed by the core step of executing the epollWait local method after updateRegistrations()
This method must call the OS’s epoll_wait function. However, this function is used to listen for events registered on epoll. The return value corresponds to the Java SelectionKey.
On the above analysis, make a small summary: actually for IO program, the JDK just made a layer of the operating system of encapsulation, and does not have to implement (think implementations are not achieve them, IO involve the hardware interface, Java processes in user mode can only be adjusted operating system), at the time of call system function involves several functions: Epoll_create () : Creates an epoll object; Epoll_ctl () : binds a channel to an op; Epoll_wait () : waits for the event to be triggered;
In addition to epoll, there are select and poll in the multiplexing implementation of Linux, which was also used when the Java NIO package first came out. The underlying implementation of SELECT is to use an array. When an event occurs, all file descriptors in the array will be looped over, with a time complexity of O(n); If I have a total of 1W connections and only 10 IO operations are triggered each time, there will be 9990 invalid loops, and since it is implemented through arrays, the number of connections it supports is limited. Poll is a slight improvement on select, with linked lists and no join caps, but the query is still O(n) based on a loop. Epoll, however, uses hash tables to call back fd by means of horizontal triggering when an event occurs, with time complexity O(1).
He that sows in tears shall reap with a smile.