Java网络编程（六）：NIO vs BIO性能对比与场景选择

public class BioServer {
    private static final int PORT = 8080;
    private ServerSocket serverSocket;
    private ExecutorService threadPool;
    
    public void start() throws IOException {
        serverSocket = new ServerSocket(PORT);
        // 创建固定大小线程池
        threadPool = Executors.newFixedThreadPool(200);
        
        System.out.println("物联网平台BIO服务器启动，端口: " + PORT);
        
        while (true) {
            Socket clientSocket = serverSocket.accept();
            threadPool.submit(new ClientHandler(clientSocket));
        }
    }
    
    private static class ClientHandler implements Runnable {
        private final Socket socket;
        
        public ClientHandler(Socket socket) {
            this.socket = socket;
        }
        
        @Override
        public void run() {
            try (BufferedReader reader = new BufferedReader(
                    new InputStreamReader(socket.getInputStream()));
                 PrintWriter writer = new PrintWriter(
                    socket.getOutputStream(), true)) {
                
                String inputLine;
                while ((inputLine = reader.readLine()) != null) {
                    // 模拟业务处理
                    processMessage(inputLine);
                    writer.println("Echo: " + inputLine);
                }
            } catch (IOException e) {
                System.err.println("客户端处理异常: " + e.getMessage());
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    // 忽略关闭异常
                }
            }
        }
        
        private void processMessage(String message) {
            // 模拟CPU密集型操作
            try {
                Thread.sleep(10); // 模拟10ms处理时间
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
}

public class NioServer {
    private static final int PORT = 8080;
    private static final int BUFFER_SIZE = 1024;
    
    private Selector selector;
    private ServerSocketChannel serverChannel;
    
    public void start() throws IOException {
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.bind(new InetSocketAddress(PORT));
        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);
                }
            }
        }
    }
    
    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);
        }
    }
    
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel channel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE);
        
        int bytesRead = channel.read(buffer);
        if (bytesRead == -1) {
            key.cancel();
            channel.close();
            return;
        }
        
        if (bytesRead > 0) {
            buffer.flip();
            String message = new String(buffer.array(), 0, buffer.limit());
            
            // 模拟业务处理
            processMessage(message);
            
            // 回写数据
            String response = "Echo: " + message;
            ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes());
            channel.write(responseBuffer);
        }
    }
    
    private void processMessage(String message) {
        // 模拟轻量级处理，避免阻塞事件循环
        // 在实际应用中，CPU密集型操作应该异步处理
    }
}

// 连接建立速度测试代码
public class ConnectionSpeedTest {
    
    public static void testConnectionSpeed(String serverType, int connectionCount) {
        long startTime = System.currentTimeMillis();
        CountDownLatch latch = new CountDownLatch(connectionCount);
        AtomicInteger successCount = new AtomicInteger(0);
        
        ExecutorService executor = Executors.newFixedThreadPool(50);
        
        for (int i = 0; i < connectionCount; i++) {
            executor.submit(() -> {
                try (Socket socket = new Socket("localhost", 8080)) {
                    successCount.incrementAndGet();
                } catch (IOException e) {
                    // 连接失败
                } finally {
                    latch.countDown();
                }
            });
        }
        
        try {
            latch.await(30, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        long endTime = System.currentTimeMillis();
        long duration = endTime - startTime;
        
        System.out.printf("%s - 连接数: %d, 成功: %d, 耗时: %dms, 速率: %.2f conn/s%n",
            serverType, connectionCount, successCount.get(), duration,
            (double) successCount.get() * 1000 / duration);
        
        executor.shutdown();
    }
}

// BIO内存使用分析
public class BioMemoryAnalysis {
    
    public static void analyzeMemoryUsage() {
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        ThreadMXBean threadBean = ManagementFactory.getThreadMXBean();
        
        // 每个线程的内存开销
        long threadStackSize = 1024 * 1024; // 默认1MB栈空间
        int activeThreads = threadBean.getThreadCount();
        
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
        MemoryUsage nonHeapUsage = memoryBean.getNonHeapMemoryUsage();
        
        System.out.println("=== BIO内存使用分析 ===");
        System.out.printf("活跃线程数: %d%n", activeThreads);
        System.out.printf("线程栈内存: %d MB%n", (activeThreads * threadStackSize) / (1024 * 1024));
        System.out.printf("堆内存使用: %d MB / %d MB%n", 
            heapUsage.getUsed() / (1024 * 1024),
            heapUsage.getMax() / (1024 * 1024));
        System.out.printf("非堆内存使用: %d MB%n", 
            nonHeapUsage.getUsed() / (1024 * 1024));
    }
}

// NIO内存使用分析
public class NioMemoryAnalysis {
    
    public static void analyzeMemoryUsage(int connectionCount) {
        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
        
        // NIO主要内存开销
        int bufferSize = 1024; // 每连接缓冲区大小
        long directMemoryUsage = connectionCount * bufferSize * 2; // 读写缓冲区
        
        MemoryUsage heapUsage = memoryBean.getHeapMemoryUsage();
        MemoryUsage nonHeapUsage = memoryBean.getNonHeapMemoryUsage();
        
        System.out.println("=== NIO内存使用分析 ===");
        System.out.printf("连接数: %d%n", connectionCount);
        System.out.printf("预估直接内存: %d MB%n", directMemoryUsage / (1024 * 1024));
        System.out.printf("堆内存使用: %d MB / %d MB%n", 
            heapUsage.getUsed() / (1024 * 1024),
            heapUsage.getMax() / (1024 * 1024));
        System.out.printf("非堆内存使用: %d MB%n", 
            nonHeapUsage.getUsed() / (1024 * 1024));
    }
}

// CPU使用率监控
public class CpuUsageMonitor {
    
    public static void monitorCpuUsage() {
        OperatingSystemMXBean osBean = ManagementFactory.getOperatingSystemMXBean();
        
        if (osBean instanceof com.sun.management.OperatingSystemMXBean) {
            com.sun.management.OperatingSystemMXBean sunOsBean = 
                (com.sun.management.OperatingSystemMXBean) osBean;
            
            double processCpuLoad = sunOsBean.getProcessCpuLoad() * 100;
            double systemCpuLoad = sunOsBean.getSystemCpuLoad() * 100;
            
            System.out.printf("进程CPU使用率: %.2f%%%n", processCpuLoad);
            System.out.printf("系统CPU使用率: %.2f%%%n", systemCpuLoad);
        }
    }
}

// GC监控代码
public class GcMonitor {
    
    public static void monitorGc() {
        List<GarbageCollectorMXBean> gcBeans = ManagementFactory.getGarbageCollectorMXBeans();
        
        for (GarbageCollectorMXBean gcBean : gcBeans) {
            long collectionCount = gcBean.getCollectionCount();
            long collectionTime = gcBean.getCollectionTime();
            
            System.out.printf("GC名称: %s%n", gcBean.getName());
            System.out.printf("GC次数: %d%n", collectionCount);
            System.out.printf("GC时间: %d ms%n", collectionTime);
            System.out.printf("平均GC时间: %.2f ms%n", 
                collectionCount > 0 ? (double) collectionTime / collectionCount : 0);
        }
    }
}

// 设备数据接入服务
public class DeviceDataService {
    
    // NIO实现 - 适用于大量设备并发接入
    public class NioDeviceServer {
        private final Map<String, DeviceSession> deviceSessions = new ConcurrentHashMap<>();
        
        public void handleDeviceData(SocketChannel channel, ByteBuffer data) {
            try {
                DeviceMessage message = parseDeviceMessage(data);
                String deviceId = message.getDeviceId();
                
                // 更新设备会话
                DeviceSession session = deviceSessions.computeIfAbsent(deviceId, 
                    k -> new DeviceSession(k, channel));
                session.updateLastActiveTime();
                
                // 异步处理数据
                CompletableFuture.runAsync(() -> {
                    processDeviceData(message);
                });
                
                // 发送确认响应
                sendAckResponse(channel, message.getMessageId());
                
            } catch (Exception e) {
                logger.error("处理设备数据异常", e);
            }
        }
        
        private void processDeviceData(DeviceMessage message) {
            // 数据验证
            if (!validateDeviceData(message)) {
                return;
            }
            
            // 数据存储（异步）
            dataStorageService.storeAsync(message);
            
            // 规则引擎处理
            ruleEngineService.processAsync(message);
            
            // 实时监控更新
            monitoringService.updateMetrics(message);
        }
    }
    
    // BIO实现 - 适用于复杂数据处理场景
    public class BioDeviceServer {
        private final ExecutorService processingPool = 
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
        
        public void handleDeviceConnection(Socket socket) {
            processingPool.submit(() -> {
                try (BufferedInputStream input = new BufferedInputStream(socket.getInputStream());
                     BufferedOutputStream output = new BufferedOutputStream(socket.getOutputStream())) {
                    
                    while (true) {
                        DeviceMessage message = readDeviceMessage(input);
                        if (message == null) break;
                        
                        // 同步处理复杂业务逻辑
                        ProcessResult result = processComplexDeviceData(message);
                        
                        // 发送处理结果
                        sendProcessResult(output, result);
                    }
                } catch (IOException e) {
                    logger.warn("设备连接异常", e);
                }
            });
        }
        
        private ProcessResult processComplexDeviceData(DeviceMessage message) {
            // 复杂的同步处理逻辑
            // 1. 数据库查询设备配置
            DeviceConfig config = deviceConfigService.getConfig(message.getDeviceId());
            
            // 2. 数据校验和转换
            ValidationResult validation = dataValidator.validate(message, config);
            if (!validation.isValid()) {
                return ProcessResult.error(validation.getErrorMessage());
            }
            
            // 3. 业务规则处理
            RuleResult ruleResult = ruleEngine.process(message, config);
            
            // 4. 数据持久化
            dataRepository.save(message, ruleResult);
            
            return ProcessResult.success(ruleResult);
        }
    }
}

// 实时监控服务
public class RealTimeMonitoringService {
    
    // NIO实现 - WebSocket长连接
    public class NioMonitoringServer {
        private final Map<String, List<SocketChannel>> subscriberChannels = new ConcurrentHashMap<>();
        
        public void broadcastAlert(Alert alert) {
            String topic = alert.getTopic();
            List<SocketChannel> channels = subscriberChannels.get(topic);
            
            if (channels != null) {
                ByteBuffer alertData = encodeAlert(alert);
                
                channels.parallelStream().forEach(channel -> {
                    try {
                        channel.write(alertData.duplicate());
                    } catch (IOException e) {
                        // 移除失效连接
                        removeChannel(topic, channel);
                    }
                });
            }
        }
        
        public void subscribe(SocketChannel channel, String topic) {
            subscriberChannels.computeIfAbsent(topic, k -> new CopyOnWriteArrayList<>())
                .add(channel);
        }
    }
}

// 文件传输服务
public class FileTransferService {
    
    // NIO实现 - 零拷贝文件传输
    public class NioFileTransfer {
        
        public void transferFile(SocketChannel channel, String filePath) throws IOException {
            try (FileChannel fileChannel = FileChannel.open(Paths.get(filePath), StandardOpenOption.READ)) {
                long fileSize = fileChannel.size();
                long transferred = 0;
                
                // 发送文件头信息
                ByteBuffer header = createFileHeader(filePath, fileSize);
                channel.write(header);
                
                // 零拷贝传输文件内容
                while (transferred < fileSize) {
                    long count = fileChannel.transferTo(transferred, fileSize - transferred, channel);
                    transferred += count;
                    
                    if (count == 0) {
                        // 避免忙等待
                        Thread.sleep(1);
                    }
                }
                
                logger.info("文件传输完成: {}, 大小: {} bytes", filePath, fileSize);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IOException("文件传输被中断", e);
            }
        }
    }
    
    // BIO实现 - 带进度反馈的文件传输
    public class BioFileTransfer {
        
        public void transferFileWithProgress(Socket socket, String filePath, 
                ProgressCallback callback) throws IOException {
            
            try (FileInputStream fileInput = new FileInputStream(filePath);
                 BufferedOutputStream output = new BufferedOutputStream(socket.getOutputStream())) {
                
                File file = new File(filePath);
                long fileSize = file.length();
                long transferred = 0;
                
                byte[] buffer = new byte[8192];
                int bytesRead;
                
                while ((bytesRead = fileInput.read(buffer)) != -1) {
                    output.write(buffer, 0, bytesRead);
                    transferred += bytesRead;
                    
                    // 进度回调
                    if (callback != null) {
                        callback.onProgress(transferred, fileSize);
                    }
                    
                    // 流量控制
                    if (transferred % (1024 * 1024) == 0) { // 每1MB检查一次
                        Thread.sleep(10); // 避免占用过多带宽
                    }
                }
                
                output.flush();
                logger.info("文件传输完成: {}", filePath);
                
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IOException("文件传输被中断", e);
            }
        }
    }
}

// 智能工厂监控系统架构
public class SmartFactoryMonitoringSystem {
    
    @Component
    public class DeviceGatewayServer {
        private final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        private final EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        @PostConstruct
        public void start() {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ChannelPipeline pipeline = ch.pipeline();
                        
                        // 协议解码器
                        pipeline.addLast(new ProtocolDecoder());
                        pipeline.addLast(new ProtocolEncoder());
                        
                        // 业务处理器
                        pipeline.addLast(new DeviceMessageHandler());
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 1024)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true);
            
            try {
                ChannelFuture future = bootstrap.bind(8080).sync();
                logger.info("物联网平台设备网关启动成功，端口: 8080");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    
    @ChannelHandler.Sharable
    public class DeviceMessageHandler extends SimpleChannelInboundHandler<DeviceMessage> {
        
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, DeviceMessage msg) {
            // 异步处理设备消息
            CompletableFuture.runAsync(() -> {
                processDeviceMessage(msg);
            }, businessExecutor);
            
            // 立即发送ACK响应
            ctx.writeAndFlush(createAckMessage(msg.getMessageId()));
        }
        
        private void processDeviceMessage(DeviceMessage message) {
            try {
                // 数据验证
                if (!validateMessage(message)) {
                    return;
                }
                
                // 实时数据处理
                realTimeProcessor.process(message);
                
                // 异常检测
                anomalyDetector.detect(message);
                
                // 数据存储
                dataStorage.storeAsync(message);
                
            } catch (Exception e) {
                logger.error("处理设备消息异常: {}", message.getDeviceId(), e);
            }
        }
    }
}

// 车联网数据采集平台
public class VehicleDataCollectionPlatform {
    
    @Service
    public class VehicleConnectionManager {
        private final Map<String, VehicleSession> vehicleSessions = new ConcurrentHashMap<>();
        private final ScheduledExecutorService heartbeatScheduler = 
            Executors.newScheduledThreadPool(4);
        
        @PostConstruct
        public void init() {
            // 启动心跳检测
            heartbeatScheduler.scheduleAtFixedRate(this::checkHeartbeat, 30, 30, TimeUnit.SECONDS);
        }
        
        public void registerVehicle(String vehicleId, SocketChannel channel) {
            VehicleSession session = new VehicleSession(vehicleId, channel);
            vehicleSessions.put(vehicleId, session);
            
            logger.info("车辆连接注册: {}", vehicleId);
            
            // 发送配置信息
            sendVehicleConfig(session);
        }
        
        public void handleVehicleData(String vehicleId, VehicleDataMessage message) {
            VehicleSession session = vehicleSessions.get(vehicleId);
            if (session != null) {
                session.updateLastActiveTime();
                
                // 异步处理车辆数据
                messageQueue.send("vehicle.data", message);
                
                // 更新车辆状态
                updateVehicleStatus(vehicleId, message);
            }
        }
        
        private void checkHeartbeat() {
            long currentTime = System.currentTimeMillis();
            List<String> timeoutVehicles = new ArrayList<>();
            
            for (Map.Entry<String, VehicleSession> entry : vehicleSessions.entrySet()) {
                VehicleSession session = entry.getValue();
                if (currentTime - session.getLastActiveTime() > 120000) { // 2分钟超时
                    timeoutVehicles.add(entry.getKey());
                }
            }
            
            // 清理超时连接
            timeoutVehicles.forEach(this::removeVehicle);
        }
    }
    
    @Component
    public class VehicleDataProcessor {
        
        @RabbitListener(queues = "vehicle.data")
        public void processVehicleData(VehicleDataMessage message) {
            try {
                // GPS数据处理
                if (message.hasGpsData()) {
                    gpsDataProcessor.process(message.getGpsData());
                }
                
                // 车辆状态数据处理
                if (message.hasStatusData()) {
                    statusDataProcessor.process(message.getStatusData());
                }
                
                // 故障诊断数据处理
                if (message.hasDiagnosticData()) {
                    diagnosticProcessor.process(message.getDiagnosticData());
                }
                
                // 数据持久化
                dataRepository.save(message);
                
            } catch (Exception e) {
                logger.error("处理车辆数据异常: {}", message.getVehicleId(), e);
                // 发送到死信队列进行人工处理
                deadLetterQueue.send(message);
            }
        }
    }
}

// 环境监测系统
public class EnvironmentalMonitoringSystem {
    
    @Service
    public class MonitoringDeviceServer {
        private final ExecutorService connectionPool = 
            new ThreadPoolExecutor(50, 200, 60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new ThreadFactoryBuilder().setNameFormat("device-handler-%d").build());
        
        public void handleDeviceConnection(Socket socket) {
            connectionPool.submit(new DeviceConnectionHandler(socket));
        }
        
        private class DeviceConnectionHandler implements Runnable {
            private final Socket socket;
            
            public DeviceConnectionHandler(Socket socket) {
                this.socket = socket;
            }
            
            @Override
            public void run() {
                try (ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
                     ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream())) {
                    
                    // 设备认证
                    DeviceAuthRequest authRequest = (DeviceAuthRequest) input.readObject();
                    DeviceAuthResponse authResponse = authenticateDevice(authRequest);
                    output.writeObject(authResponse);
                    
                    if (!authResponse.isSuccess()) {
                        return;
                    }
                    
                    String deviceId = authRequest.getDeviceId();
                    logger.info("设备认证成功: {}", deviceId);
                    
                    // 处理设备数据
                    while (true) {
                        try {
                            MonitoringData data = (MonitoringData) input.readObject();
                            
                            // 同步处理监测数据
                            ProcessResult result = processMonitoringData(deviceId, data);
                            
                            // 发送处理结果
                            output.writeObject(result);
                            
                            // 检查是否有配置更新
                            DeviceConfig config = checkConfigUpdate(deviceId);
                            if (config != null) {
                                output.writeObject(config);
                            }
                            
                        } catch (ClassNotFoundException e) {
                            logger.error("数据格式错误: {}", deviceId, e);
                            break;
                        }
                    }
                    
                } catch (IOException e) {
                    logger.warn("设备连接异常", e);
                } finally {
                    try {
                        socket.close();
                    } catch (IOException e) {
                        // 忽略关闭异常
                    }
                }
            }
            
            private ProcessResult processMonitoringData(String deviceId, MonitoringData data) {
                try {
                    // 数据校验
                    ValidationResult validation = dataValidator.validate(data);
                    if (!validation.isValid()) {
                        return ProcessResult.error(validation.getErrorMessage());
                    }
                    
                    // 数据标准化
                    NormalizedData normalizedData = dataNormalizer.normalize(data);
                    
                    // 异常检测
                    AnomalyResult anomaly = anomalyDetector.detect(normalizedData);
                    if (anomaly.hasAnomaly()) {
                        // 发送告警
                        alertService.sendAlert(deviceId, anomaly);
                    }
                    
                    // 数据存储
                    dataRepository.save(normalizedData);
                    
                    // 更新实时统计
                    statisticsService.update(normalizedData);
                    
                    return ProcessResult.success();
                    
                } catch (Exception e) {
                    logger.error("处理监测数据异常: {}", deviceId, e);
                    return ProcessResult.error("数据处理失败: " + e.getMessage());
                }
            }
        }
    }
}