Netty EventLoopGroup ServerBootstrap 简单实现服务端与客户端的通信

ゞ浴缸里的玫瑰 2023-10-01 16:31 34阅读 0赞

#### Netty 简单实现服务端与客户端的通信 ####

*  一、服务端与客户端设计
 *   *  1.服务启动类
     *  2.服务端管道处理器
     *  3.客户端启动类
     *  4.客户端管道处理器
     *  5.启动验证
     *   *  ·服务端
         *  ·客户端
 *  二、示例分析
 *   *  1.ServerBootstrap的Group方法
     *  2.ServerBootstrap的Channel方法
     *   *  1.反射实现类ReflectiveChannelFactory
         *  2.非空校验并赋值channelFactory
     *  3.ServerBootstrap的Option、ChildOption方法
     *   *  1.ChannelConfig实现类DefaultServerSocketChannelConfig
     *  4.ServerBootstrap的ChildHandler方法，实现类

## 一、服务端与客户端设计 ##

### 1.服务启动类 ###

package com.example.netty.server;
    
    import com.example.netty.server.handler.NettyServerHandler;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * Description: Netty服务端
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-11-14 23:31
     * @version: V1.0.0
     */
    public class NettyServer {
        
    
        public static void main(String[] args) throws InterruptedException {
        
    
            //创建BossGroup 和 WorkerGroup
            /**
             * 1.创建两个线程组
             * 2.BossGroup 处理连接请求
             *   WorkerGroup 处理业务逻辑
             * 3.两个都是无限循环
             */
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
    
            try {
        
                //创建服务端的启动对象 配置参数
                ServerBootstrap bootstrap = new ServerBootstrap();
    
                //使用链式编程进行设置
                /**
                 * 1.设置两个线程组
                 * 2.使用 NioSocketChannel 作为服务器的通道实现
                 * 3.设置线程队列得到的连接数
                 * 4.设置保持活动连接状态
                 * 5.给我们的 WorkerGroup 的 EventLoop 对应的管道设置处理器
                 */
                bootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,128)
                        .childOption(ChannelOption.SO_KEEPALIVE,true)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
        
                            //给 PipLine 设置处理器
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                socketChannel.pipeline().addLast(new NettyServerHandler());
                            }
                        });
    
                System.out.println("Server is Ready");
    
                //绑定一个端口并且同步 生成了一个 ChannelFuture 对象 启动服务
                ChannelFuture cf = bootstrap.bind(6668).sync();
    
                //对关闭通道进行侦听
                cf.channel().closeFuture().sync();
            }finally {
        
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

### 2.服务端管道处理器 ###

package com.example.netty.server.handler;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    import java.nio.ByteBuffer;
    import java.nio.charset.Charset;
    
    /**
     * Description: Netty服务处理器
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-11-14 23:53
     * @version: V1.0.0
     */
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
        
    
        /**
         * 读取数据
         * @param ctx 上下文对象 含有 管道 pipLine 通道 Channel 地址
         * @param message 客户端发送的数据
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object message) throws Exception {
        
    
            System.out.println("Server Read Thread :" + Thread.currentThread().getName());
            System.out.println("Server ctx=" + ctx);
    
            //将message转成一个ByteBuf
            ByteBuf buf = (ByteBuf) message;
    
            System.out.println("Client Send Message is:" + buf.toString(CharsetUtil.UTF_8));
            System.out.println("Client Ip Address is:" + ctx.channel().remoteAddress());
        }
    
        /**
         * 读取数据完毕
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        
    
            //将数据写入到缓冲区并刷新
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Client",CharsetUtil.UTF_8));
    
        }
    
        /**
         * 处理异常 一般为关闭通道
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            ctx.close();
        }
    }

### 3.客户端启动类 ###

package com.example.netty.client;
    
    import com.example.netty.client.handler.NettyClientHandler;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /**
     * Description: Netty客户端
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-11-15 0:10
     * @version: V1.0.0
     */
    public class NettyClient {
        
    
        public static void main(String[] args) throws InterruptedException {
        
    
            //客户端需要一个事件循环组
            EventLoopGroup group = new NioEventLoopGroup();
    
            try {
        
                //创建客户端启动对象
                Bootstrap bootstrap = new Bootstrap();
    
                /**
                 * 设置相关参数
                 * 设置线程组
                 * 设置客户端通道实现类（反射）
                 */
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
        
    
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
        
                                socketChannel.pipeline().addLast(new NettyClientHandler());
                            }
                        });
    
                System.out.println("Client OK ...");
    
                //启动客户端 连接服务端
                ChannelFuture cf = bootstrap.connect("127.0.0.1",6668).sync();
    
                //关闭通道增加监听
                cf.channel().closeFuture().sync();
            }finally {
        
                group.shutdownGracefully();
            }
        }
    }

### 4.客户端管道处理器 ###

package com.example.netty.client.handler;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /**
     * Description: Netty客户端处理器
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-11-15 0:29
     * @version: V1.0.0
     */
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
        
    
        /**
         * 当通道就绪就会触发
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            System.out.println("Client :" + ctx);
            ctx.writeAndFlush(Unpooled.copiedBuffer("Hello Server ", CharsetUtil.UTF_8));
        }
    
        /**
         * 当通道有读取事件时会触发
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            ByteBuf buf = (ByteBuf) msg;
            System.out.println("Server SendBack Message:" + buf.toString(CharsetUtil.UTF_8));
            System.out.println("Server Ip Address :" + ctx.channel().remoteAddress());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        
            ctx.close();
        }
    }

### 5.启动验证 ###

#### ·服务端 ####

Connected to the target VM, address: '127.0.0.1:60934', transport: 'socket'
    Java HotSpot(TM) 64-Bit Server VM warning: Sharing is only supported for boot loader classes because bootstrap classpath has been appended
    Server is Ready
    Server Read Thread :nioEventLoopGroup-3-1
    Server ctx=ChannelHandlerContext(NettyServerHandler#0, [id: 0x7748b291, L:/127.0.0.1:6668 - R:/127.0.0.1:60940])
    Client Send Message is:Hello Server 
    Client Ip Address is:/127.0.0.1:60940

#### ·客户端 ####

Connected to the target VM, address: '127.0.0.1:60937', transport: 'socket'
    Java HotSpot(TM) 64-Bit Server VM warning: Sharing is only supported for boot loader classes because bootstrap classpath has been appended
    Client OK ...
    Client :ChannelHandlerContext(NettyClientHandler#0, [id: 0xb9a0d3d1, L:/127.0.0.1:60940 - R:/127.0.0.1:6668])
    Server SendBack Message:Hello Client
    Server Ip Address :/127.0.0.1:6668

## 二、示例分析 ##

### 1.ServerBootstrap的Group方法 ###

/**
     * 查看源码：将定义的bossGroup、workerGroup用于初始化ServerBootstrap的parentGroup、childGroup
     * 结合Netty模型：初始化的两个EventLoopGroup即一个用于接收请求、一个用于处理业务
     */
    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        
        super.group(parentGroup);
        if (this.childGroup != null) {
        
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = ObjectUtil.checkNotNull(childGroup, "childGroup");
        return this;
    }

### 2.ServerBootstrap的Channel方法 ###

/**
     * 泛型 + 工厂方法 ChannelFactory<T extends Channel> extends io.netty.bootstrap.ChannelFactory<T>
     * ChannelFactory 泛型必须基于 Channel 
     * 反射 ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T>
     * 
     * 实现把NioServerSocketChannel.class类型赋值给抽象类AbstractBootstrap的属性channelFactory
     */
    public B channel(Class<? extends C> channelClass) {
        
        return channelFactory(new ReflectiveChannelFactory<C>(
                ObjectUtil.checkNotNull(channelClass, "channelClass")
        ));
    }

#### 1.反射实现类ReflectiveChannelFactory ####

package io.netty.channel;
    
    import io.netty.util.internal.ObjectUtil;
    import io.netty.util.internal.StringUtil;
    
    import java.lang.reflect.Constructor;
    
    /**
     * A {@link ChannelFactory} that instantiates a new {@link Channel} by invoking its default constructor reflectively.
     */
    public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {
        
    
        private final Constructor<? extends T> constructor;
    
        public ReflectiveChannelFactory(Class<? extends T> clazz) {
        
            ObjectUtil.checkNotNull(clazz, "clazz");
            try {
        
                this.constructor = clazz.getConstructor();
            } catch (NoSuchMethodException e) {
        
                throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
                        " does not have a public non-arg constructor", e);
            }
        }
    
        @Override
        public T newChannel() {
        
            try {
        
                return constructor.newInstance();
            } catch (Throwable t) {
        
                throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
            }
        }
    
        @Override
        public String toString() {
        
            return StringUtil.simpleClassName(ReflectiveChannelFactory.class) +
                    '(' + StringUtil.simpleClassName(constructor.getDeclaringClass()) + ".class)";
        }
    }

#### 2.非空校验并赋值channelFactory ####

/**
     * 非空校验并赋值
     */
    public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
        
        return channelFactory((ChannelFactory<C>) channelFactory);
    }
    
    /**
     * 此处废弃指ChannelFactory位置变更：io.netty.channel.ChannelFactory
     */
    @Deprecated
    public B channelFactory(ChannelFactory<? extends C> channelFactory) {
        
        ObjectUtil.checkNotNull(channelFactory, "channelFactory");
        if (this.channelFactory != null) {
        
            throw new IllegalStateException("channelFactory set already");
        }
    
        this.channelFactory = channelFactory;
        return self();
    }

### 3.ServerBootstrap的Option、ChildOption方法 ###

多个客户端如果请求同时到达，则需要一个队列来进行放置，以便服务端依次处理
    Option      ---> bossGroup
    ChildOption ---> workGroup

#### 1.ChannelConfig实现类DefaultServerSocketChannelConfig ####

@Override
    public <T> boolean setOption(ChannelOption<T> option, T value) {
        
        validate(option, value);
    
        if (option == SO_RCVBUF) {
        
            setReceiveBufferSize((Integer) value);
        } else if (option == SO_REUSEADDR) {
        
            setReuseAddress((Boolean) value);
        } else if (option == SO_BACKLOG) {
        
            setBacklog((Integer) value);
        } else {
        
            return super.setOption(option, value);
        }
    
        return true;
    }

### 4.ServerBootstrap的ChildHandler方法，实现类 ###

/**
     * 类比较长，不贴出来了，感兴趣可下载源码查看
     */
    DefaultChannelPipeline implements ChannelPipeline