美文网首页bugstac...
Netty源码分析5 - Netty 最简用法

Netty源码分析5 - Netty 最简用法

作者: 原水寒 | 来源:发表于2019-03-02 21:04 被阅读0次

    Netty 是一个事件驱动的异步网络框架。
    重要的四个学习资料:

    本文会提供一个 Netty 最简使用示例,之后的源码分析都会基于该示例及其扩展进行。关于 Netty 的最佳实践见 SOFABolt 源码分析,本系列也会介绍 Netty 在 SOFABolt 中的使用姿势。
    Netty 坐标:

    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.33.Final</version>
    </dependency>
    

    一、服务端

    1.1 服务端

    public final class EchoServer {
        public static void main(String[] args) throws Exception {
            // 1. 配置 bossGroup 和 workerGroup
            final EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            final EventLoopGroup workerGroup = new NioEventLoopGroup();
            // 2. 创建业务逻辑处理器
            final EchoServerHandler serverHandler = new EchoServerHandler();
            // 3. 创建并配置服务端启动辅助类 ServerBootstrap
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.SO_BACKLOG, 100)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(serverHandler);
                    }
                });
            // 4. 阻塞绑定端口
            ChannelFuture f = b.bind(8081).sync();
            // 5. 为服务端关闭的 ChannelFuture 添加监听器,用于实现优雅关闭
            f.channel().closeFuture().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    bossGroup.shutdownGracefully();
                    workerGroup.shutdownGracefully();
                }
            });
        }
    }
    

    建议:

    1. 如果 bossGroup 只用于绑定一个端口,那么 groupSize 设为1,因为只会使用到 bossGroup 中的一个 NioEventLoop,如果 groupSize>1,则其他多余的 NioEventLoop 只会白占内存(后续会进行源码分析)
    2. 建议使用异步的方式进行优雅关闭,不建议 f.channel().closeFuture().sync(); 同步关闭,因为在实际使用中,主线程还需要继续往后执行,不能阻塞。

    1.2 服务端业务逻辑处理器

    @Sharable
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            // 1. 将接收到的信息强转为 ByteBuf
            ByteBuf buf = (ByteBuf)msg;
            // 2. 创建 byte[]
            byte[] req = new byte[buf.readableBytes()];
            // 3. 将 ByteBuf 中的字节写入 byte[]
            buf.readBytes(req);
            // 4. 将 byte[] 转化为 string,并输出
            System.out.println("server received: " + new String(req));
            // 5. 创建响应的 ByteBuf,回写接收到的信息
            ByteBuf resp = Unpooled.wrappedBuffer(req);
            // 6. 将响应消息写入发送缓冲数组
            ctx.write(resp);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) {
            // 将发送缓冲数组中的消息全部冲刷到 SocketChannel 中
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            // Close the connection when an exception is raised.
            cause.printStackTrace();
            ctx.close();
        }
    }
    

    二、客户端

    2.1 客户端

    public final class EchoClient {
        // 1. 配置 EventLoopGroup
        final static EventLoopGroup group = new NioEventLoopGroup();
        public static void main(String[] args) throws Exception {
            // 2. 创建并配置客户端启动辅助类 Bootstrap
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new EchoClientHandler());
                    }
                });
            // 3. 阻塞连接服务端
            ChannelFuture f = b.connect("127.0.0.1", 8081).sync();
            f.channel().closeFuture().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    group.shutdownGracefully();
                }
            });
        }
    }
    

    最佳实践:通常一个服务可能会调用多个其他服务,针对每一个被调服务,会创建一个 NettyClient 与被调服务的 NettyServer 进行通信,为了防止 NioEventLoop 太多,多个 NettyClient 可以共享一个 NioEventLoopGroup。

    2.2 客户端业务逻辑处理器

    public class EchoClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            ByteBuf buf = Unpooled.wrappedBuffer("hello,netty".getBytes());
            ctx.writeAndFlush(buf);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            byte[] resp = new byte[buf.readableBytes()];
            buf.readBytes(resp);
            System.out.println("client received: " + new String(resp));
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            // Close the connection when an exception is raised.
            cause.printStackTrace();
            ctx.close();
        }
    }
    

    整个交互流程

    1. 服务端启动并绑定端口,之后启动一条 NIO 线程监听客户端连接;
    2. 客户端进行阻塞连接,连接完成时,创建 ByteBuf,发送数据给服务端;
    3. 服务端接收到消息,将数据打印到 console 之后,原封数据返回给客户端;
    4. 客户端读取数据并打印到 console,完成一次 ping-pong

    相关文章

      网友评论

        本文标题:Netty源码分析5 - Netty 最简用法

        本文链接:https://www.haomeiwen.com/subject/uwwfuqtx.html