InputStream (抽象基类)
├── FileInputStream           // 文件字节输入流
├── ByteArrayInputStream     // 字节数组输入流
├── FilterInputStream        // 过滤器输入流
│   ├── BufferedInputStream  // 缓冲输入流
│   ├── DataInputStream      // 数据输入流
│   └── PushbackInputStream  // 回推输入流
├── ObjectInputStream        // 对象输入流
├── PipedInputStream         // 管道输入流
└── SequenceInputStream      // 序列输入流

OutputStream (抽象基类)
├── FileOutputStream         // 文件字节输出流
├── ByteArrayOutputStream    // 字节数组输出流
├── FilterOutputStream       // 过滤器输出流
│   ├── BufferedOutputStream // 缓冲输出流
│   ├── DataOutputStream     // 数据输出流
│   └── PrintStream          // 打印流
├── ObjectOutputStream       // 对象输出流
└── PipedOutputStream        // 管道输出流

public class FileStreamDemo {
    // 基本文件读取
    public void readFile(String filename) throws IOException {
        try (FileInputStream fis = new FileInputStream(filename)) {
            int data;
            while ((data = fis.read()) != -1) {
                System.out.print((char) data);
            }
        }
    }
    
    // 批量读取提高效率
    public void readFileWithBuffer(String filename) throws IOException {
        try (FileInputStream fis = new FileInputStream(filename)) {
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = fis.read(buffer)) != -1) {
                System.out.print(new String(buffer, 0, bytesRead));
            }
        }
    }
    
    // 文件写入
    public void writeFile(String filename, String content) throws IOException {
        try (FileOutputStream fos = new FileOutputStream(filename)) {
            fos.write(content.getBytes());
        }
    }
    
    // 追加写入
    public void appendToFile(String filename, String content) throws IOException {
        try (FileOutputStream fos = new FileOutputStream(filename, true)) {
            fos.write(content.getBytes());
        }
    }
}

public class ByteArrayStreamDemo {
    public void demonstrateByteArrayStream() throws IOException {
        // 将数据写入内存
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        String data = "Hello World";
        baos.write(data.getBytes());
        
        // 获取内存中的数据
        byte[] bytes = baos.toByteArray();
        
        // 从内存中读取数据
        ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
        int b;
        while ((b = bais.read()) != -1) {
            System.out.print((char) b);
        }
    }
    
    // 实际应用：数据转换
    public byte[] convertImageToBytes(String imagePath) throws IOException {
        try (FileInputStream fis = new FileInputStream(imagePath);
             ByteArrayOutputStream baos = new ByteArrayOutputStream()) {
            
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = fis.read(buffer)) != -1) {
                baos.write(buffer, 0, bytesRead);
            }
            return baos.toByteArray();
        }
    }
}

public class BufferedStreamDemo {
    // 缓冲流提升性能
    public void copyFileWithBuffer(String source, String target) throws IOException {
        try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(source));
             BufferedOutputStream bos = new BufferedOutputStream(new FileOutputStream(target))) {
            
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = bis.read(buffer)) != -1) {
                bos.write(buffer, 0, bytesRead);
            }
        }
    }
    
    // 性能对比测试
    public void performanceTest(String filename) throws IOException {
        long startTime, endTime;
        
        // 无缓冲
        startTime = System.currentTimeMillis();
        try (FileInputStream fis = new FileInputStream(filename)) {
            while (fis.read() != -1) {
                // 逐字节读取
            }
        }
        endTime = System.currentTimeMillis();
        System.out.println("无缓冲时间: " + (endTime - startTime) + "ms");
        
        // 有缓冲
        startTime = System.currentTimeMillis();
        try (BufferedInputStream bis = new BufferedInputStream(new FileInputStream(filename))) {
            while (bis.read() != -1) {
                // 逐字节读取
            }
        }
        endTime = System.currentTimeMillis();
        System.out.println("有缓冲时间: " + (endTime - startTime) + "ms");
    }
}

public class DataStreamDemo {
    // 写入基本数据类型
    public void writeDataTypes(String filename) throws IOException {
        try (DataOutputStream dos = new DataOutputStream(new FileOutputStream(filename))) {
            dos.writeInt(100);
            dos.writeDouble(3.14);
            dos.writeBoolean(true);
            dos.writeUTF("Hello World");
            dos.writeLong(System.currentTimeMillis());
        }
    }
    
    // 读取基本数据类型
    public void readDataTypes(String filename) throws IOException {
        try (DataInputStream dis = new DataInputStream(new FileInputStream(filename))) {
            int intValue = dis.readInt();
            double doubleValue = dis.readDouble();
            boolean booleanValue = dis.readBoolean();
            String stringValue = dis.readUTF();
            long longValue = dis.readLong();
            
            System.out.println("Int: " + intValue);
            System.out.println("Double: " + doubleValue);
            System.out.println("Boolean: " + booleanValue);
            System.out.println("String: " + stringValue);
            System.out.println("Long: " + longValue);
        }
    }
}

Reader (抽象基类)
├── InputStreamReader        // 字节流到字符流的桥梁
│   └── FileReader          // 文件字符输入流
├── BufferedReader          // 缓冲字符输入流
├── StringReader            // 字符串输入流
├── CharArrayReader         // 字符数组输入流
├── PipedReader             // 管道字符输入流
└── FilterReader            // 过滤字符输入流
    └── PushbackReader      // 回推字符输入流

Writer (抽象基类)
├── OutputStreamWriter      // 字符流到字节流的桥梁
│   └── FileWriter         // 文件字符输出流
├── BufferedWriter         // 缓冲字符输出流
├── StringWriter           // 字符串输出流
├── CharArrayWriter        // 字符数组输出流
├── PipedWriter            // 管道字符输出流
├── PrintWriter            // 打印字符输出流
└── FilterWriter           // 过滤字符输出流

public class StreamReaderWriterDemo {
    // 指定编码读取文件
    public void readFileWithEncoding(String filename, String encoding) throws IOException {
        try (InputStreamReader isr = new InputStreamReader(
                new FileInputStream(filename), encoding)) {
            
            char[] buffer = new char[1024];
            int charsRead;
            while ((charsRead = isr.read(buffer)) != -1) {
                System.out.print(new String(buffer, 0, charsRead));
            }
        }
    }
    
    // 指定编码写入文件
    public void writeFileWithEncoding(String filename, String content, String encoding) throws IOException {
        try (OutputStreamWriter osw = new OutputStreamWriter(
                new FileOutputStream(filename), encoding)) {
            osw.write(content);
        }
    }
    
    // 编码转换
    public void convertFileEncoding(String inputFile, String outputFile, 
                                   String inputEncoding, String outputEncoding) throws IOException {
        try (InputStreamReader isr = new InputStreamReader(
                new FileInputStream(inputFile), inputEncoding);
             OutputStreamWriter osw = new OutputStreamWriter(
                new FileOutputStream(outputFile), outputEncoding)) {
            
            char[] buffer = new char[1024];
            int charsRead;
            while ((charsRead = isr.read(buffer)) != -1) {
                osw.write(buffer, 0, charsRead);
            }
        }
    }
}

public class BufferedReaderWriterDemo {
    // 按行读取文件
    public void readFileByLines(String filename) throws IOException {
        try (BufferedReader br = new BufferedReader(new FileReader(filename))) {
            String line;
            int lineNumber = 1;
            while ((line = br.readLine()) != null) {
                System.out.println(lineNumber + ": " + line);
                lineNumber++;
            }
        }
    }
    
    // 按行写入文件
    public void writeFileByLines(String filename, String[] lines) throws IOException {
        try (BufferedWriter bw = new BufferedWriter(new FileWriter(filename))) {
            for (String line : lines) {
                bw.write(line);
                bw.newLine(); // 写入系统相关的换行符
            }
        }
    }
    
    // 文本文件处理示例
    public void processTextFile(String inputFile, String outputFile) throws IOException {
        try (BufferedReader br = new BufferedReader(new FileReader(inputFile));
             BufferedWriter bw = new BufferedWriter(new FileWriter(outputFile))) {
            
            String line;
            int lineNumber = 1;
            while ((line = br.readLine()) != null) {
                // 处理每一行，例如添加行号
                bw.write(lineNumber + ": " + line);
                bw.newLine();
                lineNumber++;
            }
        }
    }
}

public class StringReaderWriterDemo {
    public void demonstrateStringStreams() throws IOException {
        // StringWriter 用于构建字符串
        StringWriter sw = new StringWriter();
        sw.write("Hello ");
        sw.write("World");
        sw.write("!");
        String result = sw.toString();
        System.out.println("StringWriter result: " + result);
        
        // StringReader 用于读取字符串
        StringReader sr = new StringReader(result);
        char[] buffer = new char[5];
        int charsRead;
        while ((charsRead = sr.read(buffer)) != -1) {
            System.out.print(new String(buffer, 0, charsRead));
        }
    }
    
    // 实际应用：模板处理
    public String processTemplate(String template, Map<String, String> variables) throws IOException {
        StringWriter result = new StringWriter();
        try (StringReader sr = new StringReader(template)) {
            int ch;
            while ((ch = sr.read()) != -1) {
                if (ch == '$') {
                    // 简单的变量替换逻辑
                    StringBuilder varName = new StringBuilder();
                    while ((ch = sr.read()) != -1 && Character.isLetterOrDigit(ch)) {
                        varName.append((char) ch);
                    }
                    String value = variables.get(varName.toString());
                    result.write(value != null ? value : "${" + varName + "}");
                    if (ch != -1) {
                        result.write(ch);
                    }
                } else {
                    result.write(ch);
                }
            }
        }
        return result.toString();
    }
}

public class StreamSelectionGuide {
    
    // 处理二进制文件 - 使用字节流
    public void copyBinaryFile(String source, String target) throws IOException {
        try (FileInputStream fis = new FileInputStream(source);
             FileOutputStream fos = new FileOutputStream(target)) {
            
            byte[] buffer = new byte[1024];
            int bytesRead;
            while ((bytesRead = fis.read(buffer)) != -1) {
                fos.write(buffer, 0, bytesRead);
            }
        }
    }
    
    // 处理文本文件 - 使用字符流
    public void copyTextFile(String source, String target) throws IOException {
        try (FileReader fr = new FileReader(source);
             FileWriter fw = new FileWriter(target)) {
            
            char[] buffer = new char[1024];
            int charsRead;
            while ((charsRead = fr.read(buffer)) != -1) {
                fw.write(buffer, 0, charsRead);
            }
        }
    }
    
    // 需要指定编码 - 使用 InputStreamReader/OutputStreamWriter
    public void copyTextFileWithEncoding(String source, String target, String encoding) throws IOException {
        try (InputStreamReader isr = new InputStreamReader(new FileInputStream(source), encoding);
             OutputStreamWriter osw = new OutputStreamWriter(new FileOutputStream(target), encoding)) {
            
            char[] buffer = new char[1024];
            int charsRead;
            while ((charsRead = isr.read(buffer)) != -1) {
                osw.write(buffer, 0, charsRead);
            }
        }
    }
}

public class FileTypeDetector {
    public String detectFileType(String filename) throws IOException {
        try (FileInputStream fis = new FileInputStream(filename)) {
            byte[] header = new byte[4];
            fis.read(header);
            
            // 检查文件头
            if (header[0] == (byte) 0xFF && header[1] == (byte) 0xD8) {
                return "JPEG";
            } else if (header[0] == (byte) 0x89 && header[1] == (byte) 0x50 && 
                      header[2] == (byte) 0x4E && header[3] == (byte) 0x47) {
                return "PNG";
            } else if (header[0] == (byte) 0x47 && header[1] == (byte) 0x49 && 
                      header[2] == (byte) 0x46) {
                return "GIF";
            } else {
                return "Unknown";
            }
        }
    }
}

public class LargeFileProcessor {
    // 分块处理大文件
    public void processLargeFile(String filename, int chunkSize) throws IOException {
        try (FileInputStream fis = new FileInputStream(filename)) {
            byte[] buffer = new byte[chunkSize];
            int bytesRead;
            int chunkNumber = 1;
            
            while ((bytesRead = fis.read(buffer)) != -1) {
                System.out.println("Processing chunk " + chunkNumber + ", size: " + bytesRead);
                // 处理当前块
                processChunk(buffer, bytesRead);
                chunkNumber++;
            }
        }
    }
    
    private void processChunk(byte[] chunk, int size) {
        // 处理数据块的逻辑
        System.out.println("Processing " + size + " bytes");
    }
}

public class EncodingConverter {
    // 检测文件编码（简单版本）
    public String detectEncoding(String filename) throws IOException {
        try (FileInputStream fis = new FileInputStream(filename)) {
            byte[] bom = new byte[3];
            fis.read(bom);
            
            if (bom[0] == (byte) 0xEF && bom[1] == (byte) 0xBB && bom[2] == (byte) 0xBF) {
                return "UTF-8";
            } else if (bom[0] == (byte) 0xFF && bom[1] == (byte) 0xFE) {
                return "UTF-16LE";
            } else if (bom[0] == (byte) 0xFE && bom[1] == (byte) 0xFF) {
                return "UTF-16BE";
            } else {
                return "Unknown";
            }
        }
    }
    
    // 转换文件编码
    public void convertEncoding(String inputFile, String outputFile, 
                               String fromEncoding, String toEncoding) throws IOException {
        try (InputStreamReader isr = new InputStreamReader(
                new FileInputStream(inputFile), fromEncoding);
             OutputStreamWriter osw = new OutputStreamWriter(
                new FileOutputStream(outputFile), toEncoding)) {
            
            char[] buffer = new char[1024];
            int charsRead;
            while ((charsRead = isr.read(buffer)) != -1) {
                osw.write(buffer, 0, charsRead);
            }
        }
    }
}

package com.io;

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;

public class BufferOptimization {
    // 测试不同缓冲区大小的性能
    public void testBufferSizes(String filename) throws IOException {
        int[] bufferSizes = {1024, 4096, 8192, 16384, 32768};

        for (int i = 0; i < 5; i++) {
            int curBufferSize = bufferSizes[i];
            new Thread(() -> {
                long startTime = System.nanoTime();
                try {
                    copyFileWithBufferSize(filename, "temp_" + curBufferSize, curBufferSize);
                    long endTime = System.nanoTime();

                    System.out.println(Thread.currentThread().getName() + "\tBuffer size: " + curBufferSize + ", Time: " + (endTime - startTime) / 1000000 + "ms");
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }

            }, "Thread-" + (i + 1)).start();
        }
    }

    private void copyFileWithBufferSize(String source, String target, int bufferSize) throws IOException {
        try (FileInputStream fis = new FileInputStream(source); FileOutputStream fos = new FileOutputStream(target)) {

            byte[] buffer = new byte[bufferSize];
            int bytesRead;
            while ((bytesRead = fis.read(buffer)) != -1) {
                fos.write(buffer, 0, bytesRead);
            }
        }
    }

    public static void main(String[] args) throws IOException {
        BufferOptimization ob = new BufferOptimization();
        ob.testBufferSizes("D:\\random_words.txt");
    }
}

public class MemoryOptimization {
    // 处理大文件时避免内存溢出
    public void processLargeFileEfficiently(String filename) throws IOException {
        try (BufferedReader br = new BufferedReader(new FileReader(filename))) {
            String line;
            while ((line = br.readLine()) != null) {
                // 逐行处理，避免将整个文件加载到内存
                processLine(line);
                // 处理完立即释放引用
                line = null;
            }
        }
    }
    
    private void processLine(String line) {
        // 处理单行数据
        System.out.println("Processing: " + line);
    }
}

// 传统BIO服务器 - 问题代码
public class BIOServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket = new ServerSocket(8080);
        ExecutorService executor = Executors.newFixedThreadPool(1000); // 线程池很快耗尽
        
        while (true) {
            Socket socket = serverSocket.accept(); // 阻塞等待
            executor.submit(() -> {
                try {
                    handleClient(socket); // 每个连接一个线程
                } catch (IOException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

/**
NIO的核心：用一个线程通过Selector监听多个Channel的事件(IO多路复用模型)，当某个Channel有数据可读/可写时才去处理，避免了传统IO中线程阻塞等待的资源浪费。所有数据传输都通过Buffer作为中介，Channel负责传输，Buffer负责存储，Selector负责调度。

工作流程：首先将需要监听的Channel注册到Selector上并指定关注的事件类型（如连接、读、写），然后Selector通过select()方法轮询检查所有注册的Channel，找出有事件发生的Channel。一旦发现有事件的Channel，就遍历对应的SelectionKey集合，根据不同的事件类型分别处理，所有的数据传输都通过Buffer作为中介在Channel之间进行读写操作。

核心优势：用更少的线程处理更多的连接，打破了传统BIO中一个连接对应一个线程的限制。通过非阻塞和事件驱动机制，单个线程可以管理数千个并发连接，线程不会被I/O操作阻塞，只有当数据真正准备好时才进行处理，这大大提高了系统的并发能力和资源利用率。同时Buffer的重用机制减少了内存分配开销，整体上实现了高性能、低资源消耗的I/O处理模型。

**/

public class ZeroCopyFileServer {
    private static final int BUFFER_SIZE = 8192;
    
    public void transferFile(String filePath, SocketChannel socketChannel) throws IOException {
        try (RandomAccessFile file = new RandomAccessFile(filePath, "r");
             FileChannel fileChannel = file.getChannel()) {
            
            long position = 0;
            long fileSize = fileChannel.size();
            
            // 零拷贝传输，性能提升10倍以上
            while (position < fileSize) {
                long transferred = fileChannel.transferTo(position, fileSize - position, socketChannel);
                position += transferred;
            }
        }
    }
}

// 1.创建选择器
Selector selector = Selector.open();

// 2.注册通道
server.register(selector, SelectionKey.OP_ACCEPT);  // 监听连接
client.register(selector, SelectionKey.OP_READ);    // 监听读取

// 3.事件循环 
while (true) {
    selector.select();  // 等待事件
    // 处理事件
}

package com.io.nio;

import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.net.InetSocketAddress;
import java.io.IOException;
import java.util.Iterator;
import java.util.Set;

/**
 * NIO核心编程模型 - 这是理解Netty和RPC框架的基础
 * <p>
 * 核心思想：
 * 1. 一个线程 + 一个Selector + 多个Channel
 * 2. 事件驱动：只处理就绪的Channel
 * 3. 非阻塞：不等待，立即返回
 */
public class NIOCorePattern {

    // =============== 核心模型：Reactor模式 ===============
    public static class NIOReactor {
        private Selector selector;
        private ServerSocketChannel serverChannel;
        private boolean running = false;

        public void start(int port) throws IOException {
            // 1. 创建核心组件
            selector = Selector.open();
            serverChannel = ServerSocketChannel.open();

            // 2. 配置非阻塞模式
            serverChannel.configureBlocking(false);
            serverChannel.bind(new InetSocketAddress(port));

            // 3. 注册到Selector，监听连接事件
            serverChannel.register(selector, SelectionKey.OP_ACCEPT);

            running = true;
            System.out.println("NIO Reactor启动，端口: " + port);

            // 4. 事件循环 - NIO核心
            eventLoop();
        }

        /**
         * 事件循环 - NIO Core
         * 这个模式被Netty、RPC框架广泛使用
         */
        private void eventLoop() throws IOException {
            while (running) {
                // 阻塞等待事件，但可以处理多个连接
                int readyChannels = selector.select(); // 核心调用

                if (readyChannels == 0) continue;

                // 获取就绪的事件
                Set<SelectionKey> selectedKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectedKeys.iterator();

                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove(); // 重要：处理完必须移除

                    // 事件分发 - 根据不同事件类型处理
                    try {
                        handleEvent(key);
                    } catch (Exception e) {
                        handleException(key, e);
                    }
                }
            }
        }

        /**
         * 事件处理器 - 这是业务逻辑的入口
         */
        private void handleEvent(SelectionKey key) throws IOException {
            if (key.isAcceptable()) {
                // 新连接事件
                handleAccept(key);
            } else if (key.isReadable()) {
                // 数据可读事件
                handleRead(key);
            } else if (key.isWritable()) {
                // 数据可写事件
                handleWrite(key);
            }
        }

        /**
         * 处理新连接 - 这是服务器的起点
         */
        private void handleAccept(SelectionKey key) throws IOException {
            ServerSocketChannel server = (ServerSocketChannel) key.channel();
            SocketChannel clientChannel = server.accept();

            if (clientChannel != null) {
                System.out.println("新连接: " + clientChannel.getRemoteAddress());

                // 配置新连接
                clientChannel.configureBlocking(false);

                // 注册到Selector，监听读事件
                SelectionKey clientKey = clientChannel.register(selector, SelectionKey.OP_READ);

                // 为每个连接创建上下文 - 这是状态管理的关键
                clientKey.attach(new ChannelContext());
            }
        }

        /**
         * 处理读事件 - 这是数据接收的核心
         */
        private void handleRead(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            ChannelContext context = (ChannelContext) key.attachment();

            ByteBuffer buffer = ByteBuffer.allocate(1024);
            int bytesRead = channel.read(buffer);

            if (bytesRead > 0) {
                buffer.flip();

                // 数据处理 - 这里是业务逻辑
                String message = processMessage(buffer);
                System.out.println("收到消息: " + message);

                // 准备响应数据
                String response = "Echo: " + message;
                context.setWriteBuffer(ByteBuffer.wrap(response.getBytes()));

                // 注册写事件
                key.interestOps(SelectionKey.OP_WRITE);

            } else if (bytesRead == -1) {
                // 连接关闭
                System.out.println("连接关闭: " + channel.getRemoteAddress());
                closeChannel(key);
            }
        }

        /**
         * 处理写事件 - 这是数据发送的核心
         */
        private void handleWrite(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            ChannelContext context = (ChannelContext) key.attachment();

            ByteBuffer writeBuffer = context.getWriteBuffer();
            if (writeBuffer != null && writeBuffer.hasRemaining()) {
                channel.write(writeBuffer);

                if (!writeBuffer.hasRemaining()) {
                    // 写完了，重新监听读事件
                    key.interestOps(SelectionKey.OP_READ);
                    context.setWriteBuffer(null);
                }
            }
        }

        /**
         * 异常处理 - 健壮性的保证
         */
        private void handleException(SelectionKey key, Exception e) {
            System.err.println("处理连接异常: " + e.getMessage());
            closeChannel(key);
        }

        /**
         * 关闭连接 - 资源清理
         */
        private void closeChannel(SelectionKey key) {
            try {
                key.channel().close();
                key.cancel();
            } catch (IOException e) {
                System.err.println("关闭连接失败: " + e.getMessage());
            }
        }

        /**
         * 消息处理 - 业务逻辑抽象
         */
        private String processMessage(ByteBuffer buffer) {
            byte[] bytes = new byte[buffer.remaining()];
            buffer.get(bytes);
            return new String(bytes).trim();
        }

        public void stop() throws IOException {
            running = false;
            if (selector != null) selector.close();
            if (serverChannel != null) serverChannel.close();
        }
    }

    // =============== 连接上下文 - 状态管理 ===============

    /**
     * 每个连接的上下文信息
     * 这是有状态协议处理的基础（如HTTP、RPC）
     */
    static class ChannelContext {
        private ByteBuffer writeBuffer;
        private long connectTime;
        private String clientId;

        public ChannelContext() {
            this.connectTime = System.currentTimeMillis();
            this.clientId = "client_" + connectTime;
        }

        public ByteBuffer getWriteBuffer() {
            return writeBuffer;
        }

        public void setWriteBuffer(ByteBuffer writeBuffer) {
            this.writeBuffer = writeBuffer;
        }

        public long getConnectTime() {
            return connectTime;
        }

        public String getClientId() {
            return clientId;
        }
    }

    // =============== 客户端模型 - 完整的通信模型 ===============
    public static class NIOClient {
        private Selector selector;
        private SocketChannel clientChannel;

        public void connect(String host, int port) throws IOException {
            selector = Selector.open();
            clientChannel = SocketChannel.open();
            clientChannel.configureBlocking(false);

            // 发起连接
            clientChannel.connect(new InetSocketAddress(host, port));
            clientChannel.register(selector, SelectionKey.OP_CONNECT);

            // 客户端事件循环
            while (true) {
                selector.select();

                Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if (key.isConnectable()) {
                        handleConnect(key);
                    } else if (key.isReadable()) {
                        handleRead(key);
                    } else if (key.isWritable()) {
                        handleWrite(key);
                    }
                }
            }
        }

        private void handleConnect(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            if (channel.finishConnect()) {
                System.out.println("连接成功");

                // 发送数据
                String message = "Hello NIO Server!";
                ByteBuffer buffer = ByteBuffer.wrap(message.getBytes());
                channel.write(buffer);

                // 注册读事件
                key.interestOps(SelectionKey.OP_READ);
            }
        }

        private void handleRead(SelectionKey key) throws IOException {
            SocketChannel channel = (SocketChannel) key.channel();
            ByteBuffer buffer = ByteBuffer.allocate(1024);

            int bytesRead = channel.read(buffer);
            if (bytesRead > 0) {
                buffer.flip();
                byte[] bytes = new byte[buffer.remaining()];
                buffer.get(bytes);
                System.out.println("收到响应: " + new String(bytes));
            }
        }

        private void handleWrite(SelectionKey key) throws IOException {
            // 客户端写事件处理
        }
    }

    // =============== 这就是Netty的核心思想！ ===============
    public static void main(String[] args) {
        System.out.println("=== NIO核心编程模型 - Netty/RPC基础 ===");
        System.out.println();

        System.out.println("这个模型就是：");
        System.out.println("1. Netty的EventLoop原理");
        System.out.println("2. RPC框架的网络层基础");
        System.out.println("3. 高性能服务器的标准模式");
        System.out.println();

        System.out.println("核心组件映射：");
        System.out.println("- NIOReactor.eventLoop() -> Netty的EventLoop");
        System.out.println("- handleEvent() -> Netty的ChannelHandler");
        System.out.println("- ChannelContext -> Netty的Channel");
        System.out.println("- 事件驱动模型 -> RPC的异步处理");
        System.out.println();

        try {
            // 启动服务器演示
            NIOReactor reactor = new NIOReactor();
            System.out.println("启动NIO服务器...");
            System.out.println("可以用telnet localhost 8080测试");

            // 在新线程中启动，避免阻塞
            new Thread(() -> {
                try {
                    reactor.start(8080);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();

            // 等待一段时间后关闭
            Thread.sleep(5000);
            reactor.stop();

        } catch (Exception e) {
            e.printStackTrace();
        }

        System.out.println();
        System.out.println("模型旨在说明：");
        System.out.println("- 1、为什么Netty性能这么高");
        System.out.println("- 2、RPC框架如何实现高并发");
        System.out.println("- 3、现代网络编程的核心思想");
    }
}

// 1. 高并发网络服务器 - 基于NIO的HTTP服务器
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class HighConcurrencyNIOServer {
    private ServerSocketChannel serverChannel;
    private Selector selector;
    private final int port;
    private final ExecutorService workerPool;
    
    // 连接管理器 - 企业级应用中通常需要管理连接状态
    private final ConcurrentHashMap<SocketChannel, ClientSession> sessions = new ConcurrentHashMap<>();
    
    public HighConcurrencyNIOServer(int port) {
        this.port = port;
        this.workerPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
    }
    
    public void start() throws IOException {
        // 1. 创建ServerSocketChannel
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress(port));
        
        // 2. 创建Selector
        selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        System.out.println("NIO服务器启动，监听端口: " + port);
        
        // 3. 事件循环
        while (true) {
            // 阻塞等待事件
            int eventCount = selector.select();
            if (eventCount == 0) continue;
            
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                keyIterator.remove();
                
                try {
                    handleKey(key);
                } catch (Exception e) {
                    handleException(key, e);
                }
            }
        }
    }
    
    private void handleKey(SelectionKey key) throws IOException {
        if (key.isAcceptable()) {
            // 处理连接请求
            handleAccept(key);
        } else if (key.isReadable()) {
            // 处理读事件
            handleRead(key);
        } else if (key.isWritable()) {
            // 处理写事件
            handleWrite(key);
        }
    }
    
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        
        if (clientChannel != null) {
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ);
            
            // 创建会话
            ClientSession session = new ClientSession(clientChannel);
            sessions.put(clientChannel, session);
            
            System.out.println("新客户端连接: " + clientChannel.getRemoteAddress());
        }
    }
    
    private void handleRead(SelectionKey key) {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ClientSession session = sessions.get(clientChannel);
        
        // 将IO操作提交给线程池处理，避免阻塞事件循环
        workerPool.submit(() -> {
            try {
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                int bytesRead = clientChannel.read(buffer);
                
                if (bytesRead > 0) {
                    buffer.flip();
                    String message = new String(buffer.array(), 0, buffer.limit());
                    System.out.println("收到消息: " + message);
                    
                    // 业务处理
                    String response = processBusinessLogic(message, session);
                    
                    // 准备响应
                    session.addResponse(response);
                    key.interestOps(SelectionKey.OP_WRITE);
                    selector.wakeup(); // 唤醒selector
                    
                } else if (bytesRead == -1) {
                    // 客户端断开连接
                    closeClient(clientChannel, key);
                }
            } catch (IOException e) {
                closeClient(clientChannel, key);
            }
        });
    }
    
    private void handleWrite(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ClientSession session = sessions.get(clientChannel);
        
        if (session != null && session.hasResponse()) {
            String response = session.getResponse();
            ByteBuffer buffer = ByteBuffer.wrap(response.getBytes());
            
            while (buffer.hasRemaining()) {
                clientChannel.write(buffer);
            }
            
            // 写完后切换回读模式
            key.interestOps(SelectionKey.OP_READ);
        }
    }
    
    private String processBusinessLogic(String message, ClientSession session) {
        // 模拟业务处理 - 企业级应用中这里会有复杂的业务逻辑
        return "HTTP/1.1 200 OK\r\n" +
               "Content-Type: text/plain\r\n" +
               "Content-Length: " + message.length() + "\r\n" +
               "\r\n" +
               "Echo: " + message;
    }
    
    private void closeClient(SocketChannel clientChannel, SelectionKey key) {
        try {
            sessions.remove(clientChannel);
            key.cancel();
            clientChannel.close();
            System.out.println("客户端断开连接");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    
    private void handleException(SelectionKey key, Exception e) {
        System.err.println("处理连接异常: " + e.getMessage());
        if (key.channel() instanceof SocketChannel) {
            closeClient((SocketChannel) key.channel(), key);
        }
    }
    
    // 客户端会话类
    private static class ClientSession {
        private final SocketChannel channel;
        private String pendingResponse;
        private long lastActiveTime;
        
        public ClientSession(SocketChannel channel) {
            this.channel = channel;
            this.lastActiveTime = System.currentTimeMillis();
        }
        
        public void addResponse(String response) {
            this.pendingResponse = response;
        }
        
        public String getResponse() {
            String response = pendingResponse;
            pendingResponse = null;
            return response;
        }
        
        public boolean hasResponse() {
            return pendingResponse != null;
        }
    }
}

/** 2. 高性能文件处理 - 大文件复制和处理
    零拷贝技术（transferTo）提升传输效率
    内存映射文件处理大文件
    分块处理避免内存溢出
**/

import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;

public class HighPerformanceFileProcessor {
    
    // 使用零拷贝技术进行文件传输
    public void copyFileWithZeroCopy(String sourcePath, String destPath) throws IOException {
        try (FileChannel sourceChannel = FileChannel.open(Paths.get(sourcePath), StandardOpenOption.READ);
             FileChannel destChannel = FileChannel.open(Paths.get(destPath), 
                 StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)) {
            
            long fileSize = sourceChannel.size();
            long transferred = 0;
            
            // 使用transferTo实现零拷贝
            while (transferred < fileSize) {
                long count = sourceChannel.transferTo(transferred, fileSize - transferred, destChannel);
                transferred += count;
            }
            
            System.out.println("文件复制完成，大小: " + fileSize + " 字节");
        }
    }
    
    // 使用内存映射处理大文件
    public void processLargeFileWithMemoryMapping(String filePath) throws IOException {
        try (FileChannel channel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ)) {
            long fileSize = channel.size();
            
            // 内存映射文件
            MappedByteBuffer mappedBuffer = channel.map(FileChannel.MapMode.READ_ONLY, 0, fileSize);
            
            // 统计文件中的行数 - 企业级应用中常见的日志分析需求
            int lineCount = 0;
            while (mappedBuffer.hasRemaining()) {
                if (mappedBuffer.get() == '\n') {
                    lineCount++;
                }
            }
            
            System.out.println("文件行数: " + lineCount);
        }
    }
    
    // 分块处理大文件 - 适用于日志处理系统
    public void processFileInChunks(String filePath, int chunkSize) throws IOException {
        try (FileChannel channel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ)) {
            ByteBuffer buffer = ByteBuffer.allocate(chunkSize);
            long position = 0;
            int chunkNumber = 1;
            
            while (true) {
                buffer.clear();
                int bytesRead = channel.read(buffer, position);
                
                if (bytesRead == -1) break;
                
                buffer.flip();
                
                // 处理当前块
                processChunk(buffer, chunkNumber);
                
                position += bytesRead;
                chunkNumber++;
            }
        }
    }
    
    private void processChunk(ByteBuffer buffer, int chunkNumber) {
        // 模拟数据处理 - 企业级应用中这里会有实际的业务逻辑
        String content = new String(buffer.array(), 0, buffer.limit());
        System.out.println("处理块 " + chunkNumber + ", 大小: " + buffer.limit() + " 字节");
        
        // 例如：日志解析、数据统计等
        long errorCount = content.lines().filter(line -> line.contains("ERROR")).count();
        if (errorCount > 0) {
            System.out.println("发现 " + errorCount + " 个错误日志");
        }
    }
}

/** 3. 消息队列服务端 - 基于NIO
NIO在消息队列中的核心价值是通过非阻塞I/O和事件驱动机制，实现单线程处理数万并发连接，大幅提升系统性能和资源利用率。
Kafka - 基于NIO实现零拷贝传输，单台服务器可处理数十万并发连接，吞吐量达到百万级消息/秒
RocketMQ - 采用Netty框架（NIO封装），通过异步处理和事件驱动架构，支持万级QPS的消息处理能力
ActiveMQ - NIO传输连接器相比传统TCP连接器，并发处理能力提升10倍以上，内存占用减少70%
**/
public class NIOMessageQueueServer {
    private ServerSocketChannel serverChannel;
    private Selector selector;
    private final Map<SocketChannel, ByteBuffer> clientBuffers = new ConcurrentHashMap<>();
    private final BlockingQueue<Message> messageQueue = new LinkedBlockingQueue<>();
    
    public void start(int port) throws IOException {
        // 创建服务端通道
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress(port));
        
        // 创建选择器
        selector = Selector.open();
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        System.out.println("NIO消息队列服务启动，监听端口: " + port);
        
        // 事件循环
        while (true) {
            selector.select(); // 阻塞等待事件
            
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                keyIterator.remove();
                
                if (key.isAcceptable()) {
                    handleAccept(key);
                } else if (key.isReadable()) {
                    handleRead(key);
                } else if (key.isWritable()) {
                    handleWrite(key);
                }
            }
        }
    }
    
    // 处理客户端连接
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        
        if (clientChannel != null) {
            clientChannel.configureBlocking(false);
            clientChannel.register(selector, SelectionKey.OP_READ);
            clientBuffers.put(clientChannel, ByteBuffer.allocate(1024));
            System.out.println("客户端连接: " + clientChannel.getRemoteAddress());
        }
    }
    
    // 处理消息读取
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = clientBuffers.get(clientChannel);
        
        try {
            int bytesRead = clientChannel.read(buffer);
            if (bytesRead > 0) {
                buffer.flip();
                
                // 解析消息
                Message message = parseMessage(buffer);
                if (message != null) {
                    messageQueue.offer(message); // 入队
                    System.out.println("收到消息: " + message.getContent());
                    
                    // 广播给所有客户端
                    broadcastMessage(message);
                }
                
                buffer.clear();
            } else if (bytesRead == -1) {
                // 客户端断开连接
                clientChannel.close();
                clientBuffers.remove(clientChannel);
                key.cancel();
            }
        } catch (IOException e) {
            clientChannel.close();
            clientBuffers.remove(clientChannel);
            key.cancel();
        }
    }
    
    // 处理消息写入
    private void handleWrite(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = (ByteBuffer) key.attachment();
        
        if (buffer != null && buffer.hasRemaining()) {
            clientChannel.write(buffer);
            if (!buffer.hasRemaining()) {
                key.interestOps(SelectionKey.OP_READ);
            }
        }
    }
    
    // 广播消息到所有客户端
    private void broadcastMessage(Message message) {
        ByteBuffer messageBuffer = ByteBuffer.wrap(message.toBytes());
        
        for (SelectionKey key : selector.keys()) {
            if (key.channel() instanceof SocketChannel && key.isValid()) {
                SocketChannel clientChannel = (SocketChannel) key.channel();
                try {
                    ByteBuffer buffer = messageBuffer.duplicate();
                    clientChannel.write(buffer);
                } catch (IOException e) {
                    // 处理写入异常
                    key.cancel();
                    clientBuffers.remove(clientChannel);
                }
            }
        }
    }
    
    // 消息解析
    private Message parseMessage(ByteBuffer buffer) {
        // 简化的消息解析逻辑
        byte[] data = new byte[buffer.remaining()];
        buffer.get(data);
        String content = new String(data, StandardCharsets.UTF_8);
        return new Message(System.currentTimeMillis(), content.trim());
    }
}

// 2. 消息实体类
class Message {
    private long timestamp;
    private String content;
    
    public Message(long timestamp, String content) {
        this.timestamp = timestamp;
        this.content = content;
    }
    
    public String getContent() { return content; }
    public long getTimestamp() { return timestamp; }
    
    public byte[] toBytes() {
        return (timestamp + ":" + content + "\n").getBytes(StandardCharsets.UTF_8);
    }
}

// 3. NIO消息队列客户端
public class NIOMessageQueueClient {
    private SocketChannel socketChannel;
    private Selector selector;
    private ByteBuffer buffer;
    
    public void connect(String host, int port) throws IOException {
        socketChannel = SocketChannel.open();
        socketChannel.configureBlocking(false);
        socketChannel.connect(new InetSocketAddress(host, port));
        
        selector = Selector.open();
        socketChannel.register(selector, SelectionKey.OP_CONNECT);
        buffer = ByteBuffer.allocate(1024);
        
        // 客户端事件循环
        while (true) {
            selector.select();
            
            Iterator<SelectionKey> keyIterator = selector.selectedKeys().iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();
                keyIterator.remove();
                
                if (key.isConnectable()) {
                    handleConnect(key);
                } else if (key.isReadable()) {
                    handleRead(key);
                } else if (key.isWritable()) {
                    handleWrite(key);
                }
            }
        }
    }
    
    private void handleConnect(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        if (channel.finishConnect()) {
            System.out.println("连接到服务器成功");
            key.interestOps(SelectionKey.OP_READ);
            
            // 启动消息发送线程
            new Thread(this::sendMessages).start();
        }
    }
    
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        buffer.clear();
        
        int bytesRead = channel.read(buffer);
        if (bytesRead > 0) {
            buffer.flip();
            byte[] data = new byte[buffer.remaining()];
            buffer.get(data);
            System.out.println("收到广播消息: " + new String(data, StandardCharsets.UTF_8));
        }
    }
    
    private void handleWrite(SelectionKey key) throws IOException {
        // 处理写入逻辑
    }
    
    // 发送消息
    public void sendMessage(String message) throws IOException {
        ByteBuffer writeBuffer = ByteBuffer.wrap(message.getBytes(StandardCharsets.UTF_8));
        socketChannel.write(writeBuffer);
    }
    
    // 模拟消息发送
    private void sendMessages() {
        try {
            Thread.sleep(1000);
            for (int i = 0; i < 5; i++) {
                sendMessage("消息 " + i + " 来自客户端");
                Thread.sleep(2000);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

// 4. 启动示例
public class MessageQueueExample {
    public static void main(String[] args) throws IOException {
        // 启动服务端
        new Thread(() -> {
            try {
                new NIOMessageQueueServer().start(8080);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
        
        // 等待服务端启动
        try { Thread.sleep(1000); } catch (InterruptedException e) {}
        
        // 启动多个客户端
        for (int i = 0; i < 3; i++) {
            final int clientId = i;
            new Thread(() -> {
                try {
                    NIOMessageQueueClient client = new NIOMessageQueueClient();
                    client.connect("localhost", 8080);
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }
    }
}

public class NIOConfiguration {
    // 连接池配置
    public static final int MAX_CONNECTIONS = 10000;
    public static final int WORKER_THREADS = Runtime.getRuntime().availableProcessors() * 2;
    public static final int BUFFER_SIZE = 8192;
    public static final int TIMEOUT_SECONDS = 30;
    
    // 文件处理配置
    public static final int FILE_CHUNK_SIZE = 1024 * 1024; // 1MB
    public static final int MAX_FILE_SIZE = 100 * 1024 * 1024; // 100MB
    
    // 队列配置
    public static final int QUEUE_CAPACITY = 100000;
    public static final int BATCH_SIZE = 1000;
}

import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.ConcurrentHashMap;

public class NIOMonitor {
    private final AtomicLong totalConnections = new AtomicLong(0);
    private final AtomicLong activeConnections = new AtomicLong(0);
    private final AtomicLong totalRequests = new AtomicLong(0);
    private final AtomicLong failedRequests = new AtomicLong(0);
    private final ConcurrentHashMap<String, AtomicLong> requestCounts = new ConcurrentHashMap<>();
    
    public void recordConnection() {
        totalConnections.incrementAndGet();
        activeConnections.incrementAndGet();
    }
    
    public void recordDisconnection() {
        activeConnections.decrementAndGet();
    }
    
    public void recordRequest(String endpoint) {
        totalRequests.incrementAndGet();
        requestCounts.computeIfAbsent(endpoint, k -> new AtomicLong(0)).incrementAndGet();
    }
    
    public void recordFailure() {
        failedRequests.incrementAndGet();
    }
    
    public void printStats() {
        System.out.println("=== NIO服务器统计信息 ===");
        System.out.println("总连接数: " + totalConnections.get());
        System.out.println("活跃连接数: " + activeConnections.get());
        System.out.println("总请求数: " + totalRequests.get());
        System.out.println("失败请求数: " + failedRequests.get());
        System.out.println("成功率: " + String.format("%.2f%%", 
            (totalRequests.get() - failedRequests.get()) * 100.0 / totalRequests.get()));
        System.out.println("各端点请求统计:");
        requestCounts.forEach((endpoint, count) -> 
            System.out.println("  " + endpoint + ": " + count.get()));
    }
}

public class ResourceManager {
    private final ExecutorService ioExecutor;
    private final ExecutorService businessExecutor;
    private final ScheduledExecutorService scheduledExecutor;
    
    public ResourceManager() {
        // IO线程池 - 处理网络IO
        this.ioExecutor = Executors.newFixedThreadPool(
            NIOConfiguration.WORKER_THREADS,
            r -> {
                Thread t = new Thread(r, "NIO-IO-Worker");
                t.setDaemon(true);
                return t;
            }
        );
        
        // 业务线程池 - 处理业务逻辑
        this.businessExecutor = Executors.newFixedThreadPool(
            NIOConfiguration.WORKER_THREADS * 2,
            r -> {
                Thread t = new Thread(r, "NIO-Business-Worker");
                t.setDaemon(true);
                return t;
            }
        );
        
        // 定时任务线程池 - 处理定时任务
        this.scheduledExecutor = Executors.newScheduledThreadPool(2,
            r -> {
                Thread t = new Thread(r, "NIO-Scheduled-Worker");
                t.setDaemon(true);
                return t;
            }
        );
    }
    
    public ExecutorService getIOExecutor() { return ioExecutor; }
    public ExecutorService getBusinessExecutor() { return businessExecutor; }
    public ScheduledExecutorService getScheduledExecutor() { return scheduledExecutor; }
    
    public void shutdown() {
        ioExecutor.shutdown();
        businessExecutor.shutdown();
        scheduledExecutor.shutdown();
    }
}

public class ByteBufferPool {
    private final ConcurrentLinkedQueue<ByteBuffer> directBuffers;
    private final ConcurrentLinkedQueue<ByteBuffer> heapBuffers;
    private final int bufferSize;
    private final int maxPoolSize;
    
    public ByteBufferPool(int bufferSize, int maxPoolSize) {
        this.bufferSize = bufferSize;
        this.maxPoolSize = maxPoolSize;
        this.directBuffers = new ConcurrentLinkedQueue<>();
        this.heapBuffers = new ConcurrentLinkedQueue<>();
        
        // 预分配一些buffer
        for (int i = 0; i < maxPoolSize / 2; i++) {
            directBuffers.offer(ByteBuffer.allocateDirect(bufferSize));
            heapBuffers.offer(ByteBuffer.allocate(bufferSize));
        }
    }
    
    public ByteBuffer getDirectBuffer() {
        ByteBuffer buffer = directBuffers.poll();
        if (buffer == null) {
            buffer = ByteBuffer.allocateDirect(bufferSize);
        }
        buffer.clear();
        return buffer;
    }
    
    public ByteBuffer getHeapBuffer() {
        ByteBuffer buffer = heapBuffers.poll();
        if (buffer == null) {
            buffer = ByteBuffer.allocate(bufferSize);
        }
        buffer.clear();
        return buffer;
    }
    
    public void returnBuffer(ByteBuffer buffer) {
        if (buffer.isDirect() && directBuffers.size() < maxPoolSize) {
            directBuffers.offer(buffer);
        } else if (!buffer.isDirect() && heapBuffers.size() < maxPoolSize) {
            heapBuffers.offer(buffer);
        }
    }
}

/** JVM参数建议
    1. 堆内存                -Xms2g -Xmx2g"
    2. 直接内存              -XX:MaxDirectMemorySize=1g
    3. G1垃圾收集器          -XX:+UseG1GC
    4. 最大GC暂停时间        -XX:MaxGCPauseMillis=20
    5. -XX:+UnlockExperimentalVMOptions
    6. ZGC低延迟垃圾收集器   -XX:+UseZGC 
**/

/** 系统参数调优   
    1. 监听队列大小            net.core.somaxconn = 65535
    2. 网络设备队列            net.core.netdev_max_backlog = 5000
    3. SYN队列大小             net.ipv4.tcp_max_syn_backlog = 65535
    4. FIN超时时间             net.ipv4.tcp_fin_timeout = 30
    5. TCP keepalive时间       net.ipv4.tcp_keepalive_time = 1200 
    6. 文件句柄数              fs.file-max = 1000000
    7. 用户级文件句柄限制      ulimit -n 1000000
**/

/**

CPU空转问题：就像一个焦虑的人不停地查看手机，即使没有新消息也要每秒检查100次。

产生的影响：
    1、CPU利用率可能达到100%，但实际工作量很少
    2、轮询开销：每次系统调用约消耗1-2微秒
    3、1000个连接每秒轮询1000次 = 1-2毫秒的纯开销

**/

/**
基于操作系统的异步IO机制，通过异步通道（AsynchronousChannel）发起IO请求后立即返回，由操作系统内核在后台完成实际的IO操作，操作完成后通过CompletionHandler回调或Future机制通知应用程序，实现真正的非阻塞IO，避免线程在IO等待期间的阻塞，提高系统并发处理能力和资源利用率。
**/

/**
AsynchronousFileChannel（异步文件IO）
AsynchronousServerSocketChannel/AsynchronousSocketChannel（异步网络IO）
CompletionHandler（异步回调处理器）
Future（异步结果获取）
底层依赖操作系统的epoll/kqueue/IOCP等异步IO机制
**/

// 核心模式：异步读写 + 回调处理
public class AIOFileTemplate {
    public void asyncOperation(String filePath) {
        AsynchronousFileChannel.open(Paths.get(filePath), StandardOpenOption.READ)
            .read(ByteBuffer.allocate(1024), 0, null, 
                new CompletionHandler<Integer, Object>() {
                    @Override
                    public void completed(Integer result, Object attachment) {
                        // I/O完成回调
                    }
                    
                    @Override
                    public void failed(Throwable exc, Object attachment) {
                        // I/O失败回调
                    }
                });
    }
}

// 服务端：接受连接 -> 读写数据 -> 处理完成
public class AIONetworkTemplate {
    private AsynchronousServerSocketChannel serverChannel;
    
    public void startServer(int port) {
        serverChannel = AsynchronousServerSocketChannel.open()
            .bind(new InetSocketAddress(port));
        
        acceptLoop();
    }
    
    private void acceptLoop() {
        serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
            @Override
            public void completed(AsynchronousSocketChannel client, Object attachment) {
                acceptLoop(); // 继续接受新连接
                handleClient(client); // 处理当前连接
            }
            
            @Override
            public void failed(Throwable exc, Object attachment) {
                // 连接失败处理
            }
        });
    }
    
    private void handleClient(AsynchronousSocketChannel client) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        client.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                // 数据读取完成，处理业务逻辑
                processData(attachment);
                // 继续读取或关闭连接
            }
            
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                // 读取失败处理
            }
        });
    }
}

// 场景特点：大文件分块处理，避免内存溢出
@Service
public class LargeFileProcessor {
    private static final int CHUNK_SIZE = 1024 * 1024; // 1MB 分块
    
    public void processLargeFile(String inputPath, String outputPath) {
        try {
            AsynchronousFileChannel inputChannel = AsynchronousFileChannel.open(
                Paths.get(inputPath), StandardOpenOption.READ);
            AsynchronousFileChannel outputChannel = AsynchronousFileChannel.open(
                Paths.get(outputPath), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            
            long fileSize = inputChannel.size();
            
            // 核心：分块异步处理，避免大文件内存问题
            processFileChunks(inputChannel, outputChannel, 0, fileSize);
            
        } catch (Exception e) {
            throw new RuntimeException("文件处理失败", e);
        }
    }
    
    private void processFileChunks(AsynchronousFileChannel inputChannel, 
                                  AsynchronousFileChannel outputChannel,
                                  long position, long totalSize) {
        
        if (position >= totalSize) {
            // 文件处理完成
            closeChannels(inputChannel, outputChannel);
            return;
        }
        
        ByteBuffer buffer = ByteBuffer.allocate(CHUNK_SIZE);
        
        inputChannel.read(buffer, position, buffer, 
            new CompletionHandler<Integer, ByteBuffer>() {
                @Override
                public void completed(Integer result, ByteBuffer attachment) {
                    if (result > 0) {
                        attachment.flip();
                        
                        // 处理数据块（例如：加密、压缩、格式转换等）
                        ByteBuffer processedData = processDataChunk(attachment);
                        
                        // 异步写入处理后的数据
                        outputChannel.write(processedData, position, processedData,
                            new CompletionHandler<Integer, ByteBuffer>() {
                                @Override
                                public void completed(Integer result, ByteBuffer attachment) {
                                    // 继续处理下一个数据块
                                    processFileChunks(inputChannel, outputChannel, 
                                        position + result, totalSize);
                                }
                                
                                @Override
                                public void failed(Throwable exc, ByteBuffer attachment) {
                                    log.error("写入数据块失败", exc);
                                }
                            });
                    } else {
                        // 读取完成
                        closeChannels(inputChannel, outputChannel);
                    }
                }
                
                @Override
                public void failed(Throwable exc, ByteBuffer attachment) {
                    log.error("读取数据块失败", exc);
                    closeChannels(inputChannel, outputChannel);
                }
            });
    }
    
    private ByteBuffer processDataChunk(ByteBuffer input) {
        // 实际的数据处理逻辑
        byte[] data = new byte[input.remaining()];
        input.get(data);
        
        // 例如：数据压缩
        byte[] processedData = compress(data);
        
        return ByteBuffer.wrap(processedData);
    }
}

// 场景特点：持续的数据流，实时处理
@Component
public class RealTimeDataProcessor {
    private final BlockingQueue<DataPacket> dataQueue = new LinkedBlockingQueue<>();
    private final AtomicBoolean running = new AtomicBoolean(true);
    
    public void startDataCollection(int port) {
        // 启动数据收集服务
        startDataServer(port);
        
        // 启动数据处理
        startDataProcessing();
    }
    
    private void startDataServer(int port) {
        try {
            AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
            serverChannel.bind(new InetSocketAddress(port));
            
            acceptDataConnections(serverChannel);
            
        } catch (Exception e) {
            throw new RuntimeException("数据服务器启动失败", e);
        }
    }
    
    private void acceptDataConnections(AsynchronousServerSocketChannel serverChannel) {
        serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
            @Override
            public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
                acceptDataConnections(serverChannel); // 继续接受新连接
                
                // 开始接收数据流
                receiveDataStream(clientChannel);
            }
            
            @Override
            public void failed(Throwable exc, Object attachment) {
                log.error("接受数据连接失败", exc);
            }
        });
    }
    
    private void receiveDataStream(AsynchronousSocketChannel clientChannel) {
        ByteBuffer buffer = ByteBuffer.allocate(4096);
        
        clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                if (result > 0) {
                    attachment.flip();
                    
                    // 解析数据包
                    List<DataPacket> packets = parseDataPackets(attachment);
                    
                    // 将数据包放入队列进行异步处理
                    packets.forEach(packet -> {
                        try {
                            dataQueue.put(packet);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                        }
                    });
                    
                    // 继续接收数据
                    attachment.clear();
                    clientChannel.read(attachment, attachment, this);
                } else {
                    // 连接断开
                    closeChannel(clientChannel);
                }
            }
            
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                closeChannel(clientChannel);
            }
        });
    }
    
    private void startDataProcessing() {
        // 启动多个处理线程
        for (int i = 0; i < Runtime.getRuntime().availableProcessors(); i++) {
            CompletableFuture.runAsync(() -> {
                while (running.get()) {
                    try {
                        DataPacket packet = dataQueue.take();
                        processDataPacket(packet);
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            });
        }
    }
    
    private void processDataPacket(DataPacket packet) {
        // 实时数据处理逻辑
        // 例如：实时分析、告警检测、数据聚合等
        
        // 异步写入结果
        saveProcessingResult(packet);
    }
    
    private void saveProcessingResult(DataPacket packet) {
        String resultPath = "results/" + packet.getId() + ".json";
        
        try {
            AsynchronousFileChannel resultChannel = AsynchronousFileChannel.open(
                Paths.get(resultPath), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
            
            ByteBuffer resultBuffer = ByteBuffer.wrap(packet.toJson().getBytes());
            
            resultChannel.write(resultBuffer, 0, null, 
                new CompletionHandler<Integer, Object>() {
                    @Override
                    public void completed(Integer result, Object attachment) {
                        closeChannel(resultChannel);
                    }
                    
                    @Override
                    public void failed(Throwable exc, Object attachment) {
                        log.error("保存处理结果失败", exc);
                    }
                });
                
        } catch (Exception e) {
            log.error("创建结果文件失败", e);
        }
    }
}

// 场景特点：大量并发连接，I/O密集型
@Component
public class HighConcurrencyServer {
    private final AtomicInteger connectionCount = new AtomicInteger(0);
    private final Map<String, AsynchronousSocketChannel> activeConnections = new ConcurrentHashMap<>();
    
    public void startServer(int port) {
        try {
            AsynchronousServerSocketChannel serverChannel = AsynchronousServerSocketChannel.open();
            serverChannel.bind(new InetSocketAddress(port));
            
            // 核心：无限循环接受连接，每个连接独立处理
            acceptConnections(serverChannel);
            
        } catch (Exception e) {
            throw new RuntimeException("服务器启动失败", e);
        }
    }
    
    private void acceptConnections(AsynchronousServerSocketChannel serverChannel) {
        serverChannel.accept(null, new CompletionHandler<AsynchronousSocketChannel, Object>() {
            @Override
            public void completed(AsynchronousSocketChannel clientChannel, Object attachment) {
                // 立即继续接受下一个连接
                acceptConnections(serverChannel);
                
                // 处理当前连接
                String clientId = "client-" + connectionCount.incrementAndGet();
                activeConnections.put(clientId, clientChannel);
                
                handleClientConnection(clientChannel, clientId);
            }
            
            @Override
            public void failed(Throwable exc, Object attachment) {
                log.error("接受连接失败", exc);
            }
        });
    }
    
    private void handleClientConnection(AsynchronousSocketChannel clientChannel, String clientId) {
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        clientChannel.read(buffer, buffer, new CompletionHandler<Integer, ByteBuffer>() {
            @Override
            public void completed(Integer result, ByteBuffer attachment) {
                if (result > 0) {
                    attachment.flip();
                    
                    // 处理接收到的数据
                    String message = StandardCharsets.UTF_8.decode(attachment).toString();
                    String response = processMessage(clientId, message);
                    
                    // 异步发送响应
                    sendResponse(clientChannel, response, clientId);
                    
                    // 继续读取下一条消息
                    attachment.clear();
                    clientChannel.read(attachment, attachment, this);
                } else {
                    // 客户端断开连接
                    activeConnections.remove(clientId);
                    closeChannel(clientChannel);
                }
            }
            
            @Override
            public void failed(Throwable exc, ByteBuffer attachment) {
                activeConnections.remove(clientId);
                closeChannel(clientChannel);
            }
        });
    }
}

@Configuration
public class AIOConfiguration {
    
    @Bean
    public AsynchronousChannelGroup channelGroup() {
        try {
            // 自定义线程池，优化性能
            ThreadPoolExecutor executor = new ThreadPoolExecutor(
                Runtime.getRuntime().availableProcessors(), 
                Runtime.getRuntime().availableProcessors() * 2, 
                60L, TimeUnit.SECONDS, // 空闲超时
                new LinkedBlockingQueue<>(1000), // 队列大小
                new ThreadFactoryBuilder()
                    .setNameFormat("aio-worker-%d")
                    .setDaemon(true)
                    .build()
            );
            
            return AsynchronousChannelGroup.withThreadPool(executor);
        } catch (Exception e) {
            throw new RuntimeException("创建通道组失败", e);
        }
    }
}

public class BufferOptimization {
    // 缓冲区大小根据业务场景调整
    private static final int SMALL_BUFFER = 4 * 1024;     // 4KB - 小数据包
    private static final int MEDIUM_BUFFER = 64 * 1024;   // 64KB - 一般文件
    private static final int LARGE_BUFFER = 1024 * 1024;  // 1MB - 大文件
    
    public ByteBuffer allocateBuffer(BufferSize size) {
        return switch (size) {
            case SMALL -> ByteBuffer.allocateDirect(SMALL_BUFFER);
            case MEDIUM -> ByteBuffer.allocateDirect(MEDIUM_BUFFER);
            case LARGE -> ByteBuffer.allocateDirect(LARGE_BUFFER);
        };
    }
}