Video tutorial
An overview,
1. I/O model
- I/O model: which channel or communication mode and architecture is used to transmit and receive data, largely determines the performance of program communication. Java supports three network programming /IO models: BIO, NIO, AIO
2. BIO
- Simply put, a session is handled by a thread. The server has a thread that listens for connections to the client. When a new connection comes in, a new thread is opened to listen and process it.
1. Code emulates multiple client operations on one server
- The following is the server-side code implementation
public class Server {
public static void main(String[] args) throws IOException {
ServerSocket server = new ServerSocket(9999);
// The main thread is used to handle connection requests
for(; ;) {// The same as while(true)
Socket socket = server.accept(); // Accept connection requests, block here when no connection requests come in
InputStream inputStream = socket.getInputStream();
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
// For each connection, a new thread is opened for listening and processing
new Thread(() -> {
String msg;
try {
while((msg = reader.readLine()) ! =null) { // Listen, always listen, output data as it comes in, then continue to listen
System.out.println(Thread.currentThread().getName() + ":"+ msg); }}catch(IOException e) { e.printStackTrace(); } }).start(); }}}Copy the code- For the server, multiple clients can be opened and connected to it
public class Client {
public static void main(String[] args) throws IOException {
Socket socket = new Socket("127.0.0.1".9999); / / the connection
OutputStream outputStream = socket.getOutputStream(); // Get the output stream
PrintStream printStream = new PrintStream(outputStream); // Wrap the output stream
Scanner scanner = new Scanner(System.in); // Get user input
while (true) {
System.out.print("Please say:");
printStream.println(scanner.nextLine()); // Put the user input into the output stream to the server
printStream.flush(); // Flush after output}}}Copy the code1.1 summary
- 1. Each Socket receives a packet and creates a thread. The thread competition and context switching affect performance.
- 2. Each thread takes up stack space and CPU resources;
- 3. Not every socket performs I/O operations, which means meaningless thread processing.
- 4. When the concurrent access from the client increases. The server will have a 1:1 thread overhead. The larger the number of visits is, the system will overflow the thread stack, fail to create threads, and eventually cause the process to break down or die, so that the external services cannot be provided.
2. Pseudo asynchronous mode
- The previous approach required one thread per session, which was a waste of resources. Here you can take advantage of the thread pool mechanism to process a fixed number of threads, and block the queue to keep the rest of the session waiting until a session ends.
public class Server {
public static void main(String[] args) throws IOException {
ServerSocket serverSocket = new ServerSocket(9999);
ExecutorService threadPool = createThreadPool(3.1000); // Initialize a thread pool with a maximum of 3 threads and a queue with a maximum of 1000 connections
for(; ;) { Socket socket = serverSocket.accept(); System.out.println("A new connection has been received");
threadPool.execute(() -> { // Thread pool processing
try {
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
String msg;
while((msg = reader.readLine()) ! =null) {
System.out.println(Thread.currentThread().getName() + "Received message:"+ msg); }}catch(IOException e) { e.printStackTrace(); }}); }}private static ExecutorService createThreadPool(int maxPoolSize, int queueSize) {
return new ThreadPoolExecutor(maxPoolSize, maxPoolSize, 120L, TimeUnit.SECONDS, newLinkedBlockingQueue<>(queueSize)); }}Copy the code- Currently, the thread pool can only handle three connections. When a fourth connection comes in, it can’t handle it. It has to wait in the queue until one connection breaks and then it can handle the fourth connection.
2.1 summary
- Pseudo asynchronous IO is implemented by thread pool, so it avoids the problem of thread resource exhaustion caused by creating an independent thread for each request. However, the synchronous blocking model is still adopted at the bottom, so it cannot fundamentally solve the problem.
- If a single message is processed slowly, or if all threads in the server thread pool are blocked, subsequent SOCKET I/O messages will be queued. New Socket requests are rejected and numerous connection timeouts occur on the client.
3. NIO
- NIO has the same role and purpose as original IO, but in a completely different way. NIO supports buffer-oriented, channel-based IO operations. NIO will read and write files in a more efficient way. NIO can be interpreted as non-blocking IO. The read and write of traditional IO can only block execution. The thread cannot do other things during the reading and writing of IO, such as calling socket.read().
- NIO has three core parts: channels, buffers, and selectors.
- Simply put, NIO does not require a single processing thread per connection. NIO generates a Buffer each time a connection is made, reads and writes to the Buffer through a Channel, and registers the Channel with a selector. You can start a new thread for the selector, and the selector’s job is to poll all channels to see if there is new content in their cache, process it if there is, and skip if there is not.
1. NIO versus BIO
- BIO processes data as a stream, while NIO processes data as a block, and block I/O is much more efficient than stream I/O
- BIO is blocking, NIO is non-blocking
- BIO operates based on byte streams and character streams, while NIO operates based on channels and buffers, where data is always read from a Channel into a Buffer or written from a Buffer into a Channel. A Selector listens for events on multiple channels (such as connection requests, data arrivals, etc.), so multiple client channels can be listened for using a single thread
2. Three cores
- NIO has three core parts: channels, buffers, and selectors.
2.1 the buffer
- It’s just a block of memory, or you can view it as an array. Buffer objects wrapped as NIO
- Different classes can be created for different types of caches. For example, ByteBuffer, CharBuffer, ShortBuffer, IntBuffer, LongBuffer, FloatBuffer, DoubleBuffer
- It is worth mentioning that this chunk of memory can be either heap memory or direct memory (that is, memory outside the program, which is in direct contact with the IO stream).
- Important concepts of cache
- Capacity: AS a block of memory, a Buffer has a fixed size, also known as “capacity”. The Buffer capacity cannot be negative and cannot be changed after creation.
- Limit: Indicates the size of data that can be manipulated in the buffer (after limit, data cannot be read or written). The buffer limit cannot be negative and cannot be larger than its capacity. Write mode, limited to the size of buffer. In read mode, limit is equal to the amount of data written.
- Position: The index of the next data to read or write. The position of the buffer cannot be negative and cannot exceed its limit
- Mark and reset: The mark is an index that specifies a specific position in the Buffer by using the mark() method in the Buffer. This position can then be restored by calling the reset() method. Mark, position, limit, and capacity observe the following invariants: 0 <= mark <= position <= limit <= capacity
- Basic operation
Buffer clear(a) // Clear the buffer and return a reference to the buffer
Buffer flip(a) // Set the buffer limit to the current position and recharge the current position to 0
int capacity(a) // Return the capacity of the Buffer
boolean hasRemaining(a) // Determine if there are any more elements in the buffer
int limit(a) // Returns the position of the Buffer limit
Buffer limit(int n) // Sets the buffer limit to n and returns a buffer object with a new limit
Buffer mark(a) // Set the flag for the buffer
int position(a) // Returns the current position of the buffer, position
Buffer position(int n)// Sets the current position of the Buffer to n and returns the modified Buffer object
int remaining(a)Return the number of elements between position and limit
Buffer reset(a) // Change position to the mark position set previously
Buffer rewind(a) // Set position to 0 and cancel markAll Buffer subclasses provide two methods for manipulating data:get(a)put(a)Method to obtain the data in the Bufferget(a): Reads a single byteget(byte[] dst): Reads multiple bytes to DST in batchesget(int index): Reads the bytes at the specified index position(Does not move position)Put data into Bufferput(byte b): Writes the given single byte to the current position of the bufferput(byte[] src): writes bytes from SRC to the current position of the bufferput(int index, byte b): writes the specified byte to the index position of the buffer(Does not move position)
Copy the code @Test
public void testBuffer(a) {
ByteBuffer byteBuffer = ByteBuffer.allocate(10); // Get a 10-byte cache
// Add a content
byteBuffer.put("curley".getBytes()); // 6 bytes of content
System.out.println("byteBuffer.position(): " + byteBuffer.position()); // The current position of the pointer is 6
System.out.println("byteBuffer.limit(): " + byteBuffer.limit()); // No use for write operations, stays at total 10
System.out.println("byteBuffer.capacity(): " + byteBuffer.capacity()); / / total 10
System.out.println("byteBuffer.hasRemaining(): " + byteBuffer.hasRemaining()); // true, whether there is any inventory
System.out.println("byteBuffer.remaining(): " + byteBuffer.remaining()); // 4, the remaining position
System.out.println("* * * * * * * * * * * * * * * * * * * * * * * * *");
byteBuffer.flip(); // Reverse to read operation
System.out.println("ByteBuffer. The get () :" + (char) byteBuffer.get()); // c
System.out.println("byteBuffer.position(): " + byteBuffer.position()); // The read operation starts at 0 and reaches 1
System.out.println("byteBuffer.limit(): " + byteBuffer.limit()); // Indicates the limit of where the content is actually stored
System.out.println("byteBuffer.capacity(): " + byteBuffer.capacity()); / / total 10
System.out.println("byteBuffer.hasRemaining(): " + byteBuffer.hasRemaining()); // true, whether there is any inventory
System.out.println("byteBuffer.remaining(): " + byteBuffer.remaining()); // 5, indicating how much is left unread. The initial value is the limit() value
System.out.println("* * * * * * * * * * * * * * * * * * * * * * * * *");
byteBuffer.mark(); // Let's label it
byte[] buffer = new byte[3]; // Create a new array to receive the content
byteBuffer.get(buffer);
System.out.println("Read 3 bytes:" + new String(buffer));
System.out.println("* * * * * * * * * * * * * * * * * * * * * * * * *");
byteBuffer.reset(); / / reset
// The same data as before reading 3 bytes
System.out.println("byteBuffer.position(): " + byteBuffer.position()); // The read operation starts at 0 and reaches 1
System.out.println("byteBuffer.limit(): " + byteBuffer.limit()); // Indicates the limit of where the content is actually stored
System.out.println("byteBuffer.capacity(): " + byteBuffer.capacity()); / / total 10
System.out.println("byteBuffer.hasRemaining(): " + byteBuffer.hasRemaining()); // true, whether there is any inventory
System.out.println("byteBuffer.remaining(): " + byteBuffer.remaining()); // 5, indicating how much is left unread. The initial value is the limit() value
System.out.println("* * * * * * * * * * * * * * * * * * * * * * * * *");
}
Copy the code2.1.1 Direct and indirect cache areas
byte byfferThere can be two types, one based on direct memory (i.e., non-heap memory); The other is indirect memory (that is, heap memory). For direct memory, the JVM will have higher performance for IO operations because it directly affects the IO operations of the local system. Non-direct memory, that is, the heap memory, if I/O operations, will be copied from the local process memory to direct memory, and then the local I/O processing.- From the perspective of data flow, indirect memory is the following action chain:
Local IO--> Direct memory --> Indirect memory --> Direct memory --> local IOCopy the code- Direct memory is:
Local IO--> Direct memory --> local IOCopy the code- To operate direct memory, modify the method of creating a cache
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(10);
Copy the code2.2 channel
- Channels: defined by the java.nio.Channels package. Channel indicates the open connection between the I/O source and the destination. A Channel is similar to a traditional stream. However, the Channel itself cannot access the data directly; it can only interact with buffers.
- NIO channels are similar to streams, but with some differences as follows:
- Channels can read and write simultaneously, while streams can only read or write
- Channels can read and write data asynchronously
- A channel can either read data from the buffer or write data to the buffer:
- A STREAM in BIO is unidirectional. For example, a FileInputStream can only read data, while a Channel in NIO is bidirectional and can be read or written.
- A Channel is an interface in NIO
public interface Channel extends Closeable{}
Copy the code- Common implementation classes
- FileChannel: Channel for reading, writing, mapping, and manipulating files.
- DatagramChannel: Reads and writes data channels on the network through UDP.
- SocketChannel: Reads and writes data on the network through TCP.
- ServerSocketChannel: Listens for incoming TCP connections and creates a SocketChannel for each incoming connection. ServerSocketChanne Similar to ServerSocket, SocketChannel similar to Socket
2.2.1 Channel input
@Test
public void testInputChannel(a) throws IOException {
FileInputStream fis = new FileInputStream("src/main/resources/data.txt");
FileChannel channel = fis.getChannel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024); // Define a buffer
channel.read(byteBuffer); // Read data through channels
// Reverse to read
byteBuffer.flip();
System.out.println(new String(byteBuffer.array(), 0, byteBuffer.remaining())); // Convert the read data to a string
}
Copy the code2.2.2 Channel output
@Test
public void testOutputChannel(a) {
try (FileOutputStream fos = new FileOutputStream("src/main/resources/data0.txt")) {
FileChannel channel = fos.getChannel();
ByteBuffer byteBuffer = ByteBuffer.allocate(20);
byteBuffer.put(Hello world..getBytes());
byteBuffer.flip(); // Reverse mode
channel.write(byteBuffer);
} catch(Exception e) { e.printStackTrace(); }}Copy the code2.2.3 Copying files through channel
@Test
public void testCopyLargeFile(a) throws IOException {
FileChannel outChannel = new FileOutputStream("SRC/main/resources/wallpaper (2).jpg").getChannel(); // Get the output channel
FileChannel inputChannel = new FileInputStream("SRC/main/resources/wallpaper. JPG").getChannel(); // Get the input channel
ByteBuffer byteBuffer = ByteBuffer.allocateDirect(1024); // Create a new cache. The interesting thing here is to use a direct memory
while(inputChannel.read(byteBuffer) ! = -1) { / / read the data
byteBuffer.flip(); // Reverse conversion to read operation
outChannel.write(byteBuffer); // Output data to a file
byteBuffer.clear(); // Clear the cache
}
// Remember to close
outChannel.close();
inputChannel.close();
}
Copy the code2.2.4 Dispersion and aggregation
- You can read and write from the cache array at once
@Test
public void testScatterAndGather(a) throws IOException {
RandomAccessFile inputFile = new RandomAccessFile("src/main/resources/data.txt"."rw"); // Get files that can be accessed randomly
FileChannel inChannel = inputFile.getChannel();
RandomAccessFile outputFile = new RandomAccessFile("src/main/resources/data1.txt"."rw"); // Write to the file
FileChannel outputChannel = outputFile.getChannel();
// Scatter reads
ByteBuffer[] buffers = {ByteBuffer.allocate(4), ByteBuffer.allocate(1024)};
inChannel.read(buffers);
/ / reverse
for (ByteBuffer buffer : buffers) {
buffer.flip();
System.out.println("Current buffer read:" + new String(buffer.array(), 0, buffer.remaining()));
}
// Aggregate write
outputChannel.write(buffers);
inChannel.close();
inputFile.close();
}
Copy the code2.2.5 transferFrom and transferTo
@Test
public void testTransform(a) throws IOException {
FileChannel outChannel = new FileOutputStream("SRC/main/resources/wallpaper (3).jpg").getChannel(); // Get the output channel
FileChannel inputChannel = new FileInputStream("SRC/main/resources/wallpaper. JPG").getChannel(); // Get the input channel
//outChannel.transferFrom(inputChannel, 0, inputChannel.size());
inputChannel.transferTo(0, inputChannel.size(), outChannel);
// Remember to close
outChannel.close();
inputChannel.close();
}
Copy the code2.3 the selector
- A Selector is a multiplexer of a SelectableChannle object. A Selector can monitor the I/O status of multiple SelectableChannel at the same time. With Selector, a single thread can manage multiple channels. Selector is the heart of non-blocking IO
- Create a Selector()
Selector selector = Selector.open();
Copy the code- Selector registration channel
//1. Obtain the channel
ServerSocketChannel ssChannel = ServerSocketChannel.open();
//2. Switch to non-blocking mode
ssChannel.configureBlocking(false);
//3. Bind the connection
ssChannel.bind(new InetSocketAddress(9898));
//4. Get the selector
Selector selector = Selector.open();
//5. Register the channel with the selector and specify "listen to receive events"
ssChannel.register(selector, SelectionKey.OP_ACCEPT);
Copy the code- Among them,
register(Selector sel, int ops)The second argument is the event that the selector registers to listen for on the channel.
* read: selectionkey.op_read (1Selectionkey.op_write (selectionkey.op_write (4Selectionkey.op_connect (selectionkey.op_connect (8Selectionkey.op_accept (selectionkey.op_accept (16) * If you are listening for more than one event when registering, you can use the bit-or operator to connect.Copy the code- You can do this if you want to listen for multiple events
int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE
Copy the code4. NIO communication case
- Start by putting the channel that receives the connection into the selector, and then start the selector’s listening task. If a task comes in, see if it’s a new connection or a new message and deal with it accordingly.
public class Server {
public static void main(String[] args) throws IOException {
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open(); // Socket Server dedicated channel, enabled
serverSocketChannel.configureBlocking(false); // Enable non-blocking mode
serverSocketChannel.bind(new InetSocketAddress(8888)); // Bind the port number
Selector selector = Selector.open(); // Start a new selector
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT); // Bind the socket processing connection channel to the selector and select receive mode
System.out.println("Turned on");
// Enter the polling state
while (selector.select() > 0) { // When the channel is greater than 0
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) { // If a task is coming, proceed, otherwise block
System.out.println("Here comes a mission.");
SelectionKey selectionKey = iterator.next();
if (selectionKey.isAcceptable()) { // If the connection task is requested
System.out.println("There's a new connection coming in.");
SocketChannel newChannel = serverSocketChannel.accept(); // It must be received after all, but now it is entering the receive request
newChannel.configureBlocking(false);// Enable non-blocking mode
newChannel.register(selector, SelectionKey.OP_READ); // Place the new channel in the selector
} else if (selectionKey.isReadable()) { // If the task is to write
System.out.println("There's new data coming in.");
SocketChannel channel = (SocketChannel) selectionKey.channel(); // Take out the inner passage
// Start basic read operations
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
while (channel.read(byteBuffer) > 0) {
byteBuffer.flip();
System.out.println(new String(byteBuffer.array(), 0, byteBuffer.remaining())); byteBuffer.clear(); }}// Delete it after processingiterator.remove(); }}// The task is completeserverSocketChannel.close(); selector.close(); }}Copy the code- Multiple clients can be started at the same time
public class Client {
public static void main(String[] args) throws IOException {
SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress("127.0.0.1".8888));
socketChannel.configureBlocking(false);
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
Scanner scan = new Scanner(System.in);
while(scan.hasNextLine()) { String line = scan.nextLine(); byteBuffer.put(line.getBytes()); byteBuffer.flip(); socketChannel.write(byteBuffer); byteBuffer.clear(); } socketChannel.close(); }}Copy the code5. Group chat cases
public class Server {
private final Integer PORT;
private ServerSocketChannel serverSocketChannel;
private Selector selector;
public Server(Integer PORT) {
this.PORT = PORT;
try {
serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.configureBlocking(false);
serverSocketChannel.bind(new InetSocketAddress(this.PORT));
selector = Selector.open();
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
System.out.println("Server started successfully");
} catch(IOException e) { e.printStackTrace(); }}/** * start listening tasks, including processing connections, processing messages */
private void listen(a) {
System.out.println("Start listening.");
try {
// Enter the polling state
while (selector.select() > 0) { // When the channel is greater than 0
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) { // If a task is coming, proceed, otherwise block
System.out.println("Here comes a mission.");
SelectionKey selectionKey = iterator.next();
if (selectionKey.isAcceptable()) { // If the connection task is requested
SocketChannel newChannel = serverSocketChannel.accept(); // It must be received after all, but now it is entering the receive request
System.out.println("New connection coming in :" + newChannel.getRemoteAddress());
newChannel.configureBlocking(false);// Enable non-blocking mode
newChannel.register(selector, SelectionKey.OP_READ); // Place the new channel in the selector
} else if (selectionKey.isReadable()) { // If the task is to write
tackleIssue(selectionKey);
}
// Delete it after processingiterator.remove(); }}}catch (IOException e) {
e.printStackTrace();
} finally{ closeResource(selector); closeResource(serverSocketChannel); }}/** * Used to close the resource *@paramCloseable interface * /
private static void closeResource(Closeable closeable) {
if(closeable ! =null) {
try {
closeable.close();
} catch(IOException e) { e.printStackTrace(); }}}/** * Process message tasks *@param selectionKey
*/
private void tackleIssue(SelectionKey selectionKey) {
SocketChannel channel = (SocketChannel) selectionKey.channel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
try {
while ((channel.read(byteBuffer)) > 0) {
byteBuffer.flip();
String msg = new String(byteBuffer.array(), 0, byteBuffer.remaining());
System.out.println("Server received" + channel.getRemoteAddress() + "Message:"+ msg); sendToOtherClients(byteBuffer, channel); byteBuffer.clear(); }}catch (IOException e) {
try {
System.out.println(channel.getRemoteAddress() + "Offline...");
selectionKey.cancel(); // Unregister in the selector
closeResource(channel); // Close the channel
} catch(IOException ioException) { ioException.printStackTrace(); } e.printStackTrace(); }}/** * forwards the message to other clients *@param byteBuffer
* @param channel
* @throws IOException
*/
private void sendToOtherClients(ByteBuffer byteBuffer, SocketChannel channel) throws IOException {
System.out.println("Send to other clients");
for (SelectionKey key : selector.keys()) { // Iterate over all channels in selector
SelectableChannel dst = key.channel();
if (dst instanceofSocketChannel && dst ! = channel) {// If the channel is not the sending channel or the processing channel of the connection request, send a copy to it
byteBuffer.mark(); // Remember the current location
((SocketChannel) dst).write(byteBuffer); // Crazy output to other clients
byteBuffer.reset(); // Go back}}}public static void main(String[] args) {
Server server = new Server(8080); server.listen(); }}Copy the codepublic class Client {
private final String HOST;
private final Integer PORT;
private SocketChannel socketChannel;
private Selector selector;
private String username;
public Client(String HOST, Integer PORT, String username) {
this.HOST = HOST;
this.PORT = PORT;
this.username = username;
try {
socketChannel = SocketChannel.open(new InetSocketAddress(this.HOST, this.PORT)); // Open a socket client channel
socketChannel.configureBlocking(false);
selector = Selector.open();
socketChannel.register(this.selector, SelectionKey.OP_READ); / / register
} catch(IOException e) { e.printStackTrace(); }}/** * After login, open the receiving thread and sending thread */
public void login(a) {
System.out.println(username + "Login successful");
receiveMessage();
sendMessage();
}
/** * used to send messages */
private void sendMessage(a) {
new Thread(() -> {
Scanner scan = new Scanner(System.in);
while (scan.hasNextLine()) {
String msg = scan.nextLine();
msg = username + ":" + msg;
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
byteBuffer.put(msg.getBytes());
byteBuffer.flip();
try {
socketChannel.write(byteBuffer);
System.out.println("Message sent successfully:" + msg);
} catch (IOException e) {
e.printStackTrace();
}
}
}).start();
}
/** * Is used to receive messages */
public void receiveMessage(a) {
// Enable reading information
new Thread(() -> {
System.out.println("Start reading message");
try {
while (selector.select() > 0) {
Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
while (iterator.hasNext()) {
System.out.println("We have news.");
SelectionKey selectionKey = iterator.next();
if (selectionKey.isReadable()) {
SocketChannel channel = (SocketChannel) selectionKey.channel();
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
channel.read(byteBuffer);
byteBuffer.flip();
String msg = new String(byteBuffer.array(), 0, byteBuffer.remaining()); System.out.println(msg); } iterator.remove(); }}}catch (IOException e) {
e.printStackTrace();
}
}).start();
}
public static void main(String[] args) {
Client client = new Client("127.0.0.1".8080."Zhang"); client.login(); }}Copy the code