NIO学习（十三）：网络编程IO、NIO和Netty的写法

爱被打了一巴掌 2023-05-31 14:00 6阅读 0赞

##### 背景 #####

1.  整理网络编程的基本写法（IO 、NIO、Netty）。

--------------------

**IO服务端**

public static void main(String[] args) throws IOException {
            ExecutorService pool = Executors.newCachedThreadPool();
            ServerSocket serverSocket = new ServerSocket(8899);
    
            while(true) {
                final Socket socket = serverSocket.accept();
                pool.execute(new Runnable() {
                    @Override
                    public void run() {
                        byte[] bytes = new byte[1024];
                        try {
                            InputStream inputStream = socket.getInputStream();
                            int read = inputStream.read(bytes);
                            // todo 业务逻辑
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
    
                    }
                });
            }
        }

--------------------

**NIO服务端**

private static Map<String, SocketChannel> clientMap = new HashMap();
    
        public static void main(String[] args) throws Exception{
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);//非阻塞
            ServerSocket serverSocket = serverSocketChannel.socket();
            serverSocket.bind(new InetSocketAddress(8899));
    
            Selector selector = Selector.open();
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    
            while (true) {
                try {
                    selector.select();
    
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
    
                    selectionKeys.forEach(selectionKey -> {
                        final SocketChannel client;
    
                        try{
                            if (selectionKey.isAcceptable()) {
                                ServerSocketChannel server = (ServerSocketChannel) selectionKey.channel();
                                client = server.accept();
                                client.configureBlocking(false);
                                client.register(selector, SelectionKey.OP_READ);
    
                                String key = "【" + UUID.randomUUID().toString() + "】";
    
                                clientMap.put(key, client);
                            } else if (selectionKey.isReadable()) {
                               client = (SocketChannel)selectionKey.channel();
                               ByteBuffer readBuffer = ByteBuffer.allocate(1024);
    
                               int count = client.read(readBuffer);
    
                               if (count > 0) {
                                   readBuffer.flip();
    
                                   Charset charset = Charset.forName("utf-8");
                                   String receivedMessage = String.valueOf(charset.decode(readBuffer).array());
    
                                   System.out.println(client + ": " + receivedMessage);
    
    
                                   String senderKey = null;
                                   for(Map.Entry<String, SocketChannel> entry : clientMap.entrySet()) {
                                       if (client == entry.getValue()) {
                                           senderKey = entry.getKey();
                                           break;
                                       }
                                   }
                                   for(Map.Entry<String, SocketChannel> entry: clientMap.entrySet()) {
                                       SocketChannel value = entry.getValue();
                                       ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                                       writeBuffer.put((senderKey + ": " + receivedMessage).getBytes());
                                       writeBuffer.flip();
                                       value.write(writeBuffer);
                                   }
                               }
                            }
                        } catch (Exception ex) {
                            ex.printStackTrace();
                        }
                    });
                    selectionKeys.clear();
    
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }

**NIO客户端**

public static void main(String[] args) throws Exception {
            try{
                SocketChannel socketChannel = SocketChannel.open();
                socketChannel.configureBlocking(false);
    
                Selector selector = Selector.open();
    
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
                socketChannel.connect(new InetSocketAddress("127.0.0.1",8899));
    
                while (true) {
                    selector.select();
    
                    Set<SelectionKey> keySet = selector.selectedKeys();
    
                    for (SelectionKey selectionKey: keySet) {
                        if (selectionKey.isConnectable()) {
                            SocketChannel client = (SocketChannel)selectionKey.channel();
    
                            if (client.isConnectionPending()) {
                                client.finishConnect();
                                ByteBuffer writeBuffer = ByteBuffer.allocate(1024);
                                writeBuffer.put((LocalDateTime.now() + "  Connected!").getBytes());
                                writeBuffer.flip();
                                client.write(writeBuffer);
    
                                ExecutorService executorService = Executors.newSingleThreadExecutor(Executors.defaultThreadFactory());
                                executorService.submit(() -> {
                                    try {
                                        writeBuffer.clear();
                                        InputStreamReader input = new InputStreamReader(System.in);
                                        BufferedReader br =  new BufferedReader(input);
                                        String sendMessage = br.readLine();
                                        writeBuffer.put(sendMessage.getBytes());
                                        writeBuffer.flip();
                                        client.write(writeBuffer);
                                    } catch (Exception ex) {
                                        ex.printStackTrace();
                                    }
                                });
                            }
    
                            client.register(selector, SelectionKey.OP_READ);
                        } else if (selectionKey.isReadable()) {
                            SocketChannel client = (SocketChannel)selectionKey.channel();
                            ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                            int count = client.read(readBuffer);
                            if (count > 0) {
                                String receivedMessage = new String(readBuffer.array(), 0, count);
                                System.out.println(receivedMessage);
                            }
                        }
                    }
                    keySet.clear();
                }
    
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        }

--------------------

Netty服务端

1.  MyServer

public class MyServer {
    
        public static void main(String[] args) throws InterruptedException {
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workGroup = new NioEventLoopGroup();
    
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class)
                        .handler(new LoggingHandler(LogLevel.WARN))
                        .childHandler(new MyServerInitalizer());
                ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
                channelFuture.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
    }

1.  MyServerInitalizer

public class MyServerInitalizer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
    
            //Netty自己提供的基础组件
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
    
            //开发者自定义的handler
            pipeline.addLast("my server handler: ", new MyServerHandler());
        }
    }

1.  MyHandler

public class MyServerHandler extends SimpleChannelInboundHandler<String> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
    
            System.out.println(ctx.channel().remoteAddress() + ", " + msg);
            ctx.channel().writeAndFlush("from server: " + UUID.randomUUID());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

--------------------

Netty客户端

1.  MyClient

public class MyClient {
    
        public static void main(String[] args) throws InterruptedException {
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try{
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new MyClientInitalizer());
    
                ChannelFuture channelFuture = bootstrap.connect("localhost", 8899).sync();
    
                channelFuture.channel().closeFuture().sync();
            } finally {
                eventLoopGroup.shutdownGracefully();
            }
        }
    }

1.  MyClientInitalizer

public class MyClientInitalizer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
    
            //Netty自己提供的基础组件
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
    
            pipeline.addLast(new MyClientHandler());
        }
    }

1.  MyClientHandler

public class MyClientHandler extends SimpleChannelInboundHandler<String> {
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println(ctx.channel().remoteAddress());
            System.out.println("client output: " + msg);
            ctx.writeAndFlush("from client: " + LocalDateTime.now());
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush("Hello, How's it going!");
            super.channelActive(ctx);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

--------------------

##### 小结 #####

1.  整理网络编程的演讲过程。
2.  Netty与NIO类比  
    Netty的NioEventLoopGroup相当于 NIO的 Selector  
    Netty的ServerSocketChannel相当于 NIO的channle  
    Netty的ByteBuf相当于NIO的ByteBuffer