Netty简单使用

Dear 丶 2022-10-13 05:26 277阅读 0赞

### 1.客户端示例代码: ###

public class NettyClient {
    
        private static Bootstrap b;
        private static ChannelFuture f;
        private static final EventLoopGroup workerGroup = new NioEventLoopGroup();
    
        private static void init () {
            try {
                log.info("init...");
    
                //创建客户端引导类,初始化类
                b = new Bootstrap();
    
                //绑定工作线程租
                b.group(workerGroup);
    
                //设置异步的客户端TCP Socket连接
                b.channel(NioSocketChannel.class);
    
                //是否启用心跳保活机制。在双方TCP套接字建立连接后并且在两个小时左右上层没有任何数据传输的情况下,这套机制才会被激活
                b.option(ChannelOption.SO_KEEPALIVE, true);
    
                //初始化ChannelPipeline实例链
                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        // 解码编码
                        socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
                        socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
                        //执行业务逻辑的客户端handler
                        socketChannel.pipeline().addLast(new ClientHandler());
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public static Object startAndWrite(InetSocketAddress address, Object send) throws InterruptedException {
            init();
            //获取到连接
            f = b.connect(address).sync();
            // 传数据给服务端
            f.channel().writeAndFlush(send);
            //一直阻塞到消息返回demo实例,生产不能用
            f.channel().closeFuture().sync();
            //获取服务端返回的信息
            return f.channel().attr(AttributeKey.valueOf("Attribute_key")).get();
        }
    
        public static void main(String[] args) {
            InetSocketAddress address = new InetSocketAddress("192.168.0.181", 8007);
            String message ="你好,服务端";
            try {
                Object result = NettyClient.startAndWrite(address, message);
                log.info("....返回信息:" + result);
            } catch (Exception e) {
                log.error(e.getMessage());
            } finally {
                f.channel().close();
                workerGroup.shutdownGracefully();
                log.info("Closed client!");
            }
        }
    }

public class ClientHandler extends SimpleChannelInboundHandler<Object> {
        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object msg) {
            channelHandlerContext.channel().attr(AttributeKey.valueOf("Attribute_key")).set(msg);
            channelHandlerContext.close();
        }
    }

### 2.服务端示例代码: ###

@Component
    @Slf4j
    public class NettyServer {
        //专门用于网络事件处理（接受客户端的连接) nio线程组
        private final EventLoopGroup bossGroup = new NioEventLoopGroup();
    
        //进行网络通信的读写nio线程组
        private final EventLoopGroup workerGroup = new NioEventLoopGroup();
    
        private Channel channel;
    
        /**
         * 启动服务
         */
        public ChannelFuture run (int port) {
    
            ChannelFuture f = null;
            try {
    
                //创建服务端引导类
                ServerBootstrap b = new ServerBootstrap();
    
                //绑定boss线程组和工作线程组
                b.group(bossGroup, workerGroup)
                        
                 //异步的服务器端TCP Socket连接
                 .channel(NioServerSocketChannel.class)
    
                 //服务端自定义的ChannelPipeline实例链,里面有多个ChannelHandler处理channel连接
                 .childHandler(new ServerChannelInitializer())
    
                  //配置boss线程组参数,标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
                  .option(ChannelOption.SO_BACKLOG, 500)
    
                  //是否启用心跳保活机制
                  .childOption(ChannelOption.SO_KEEPALIVE, true);
                 
                //绑定监听端口,并且服务端设置不会被中断的同步方法阻塞直到绑定成功
                f = b.bind(port).syncUninterruptibly();
    
                channel = f.channel();
            } catch (Exception e) {
                log.error("Netty start error:", e);
            } finally {
                if (f != null && f.isSuccess()) {
                    log.info("监听成功!");
                } else {
                    log.error("Netty server start up Error!");
                }
            }
    
            return f;
        }
    
        public void destroy() {
            log.info("Shutdown Netty Server...");
            
            //关闭channel连接
            if(channel != null) { channel.close();}
    
            //服务端设置优雅的关闭netty server和通道
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
            log.info("Shutdown Netty Server Success!");
        }
    }

public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel socketChannel) {
            // 解码编码
            socketChannel.pipeline().addLast(new StringDecoder(CharsetUtil.UTF_8));
            socketChannel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
            
            //自定义的消息处理业务逻辑
            socketChannel.pipeline().addLast(new ServerHandler());
        }
    }

@Slf4j
    public class ServerHandler extends SimpleChannelInboundHandler<Object> {
    
        @Override
        public void channelRead0(ChannelHandlerContext context, Object msg) {
            if (msg == null) {
                log.error("没有接受到任何信息======");
                context.close();
            }
            log.info("服务接受到信息为[" + context.channel().remoteAddress() + "]: " + msg.toString());
            String content = "你好,服务端已经收到消息";
            context.channel().writeAndFlush(content);
            context.close();
        }
    
        @Override
        public void channelActive(ChannelHandlerContext context) {
            System.out.println("channelActive>>>>>>>>");
        }
    }