Netty基本使用

矫情吗；* 2023-10-01 23:32 13阅读 0赞

1.找到工程中的pom.xml文件，在dependencies中添加如下代码

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

2.创建NettyServer类

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    public class NettyServer {
        //绑定的端口号
        private int port;
    
        public NettyServer(int port) {
            this.port = port;
            bind();
        }
    
        private void bind() {
            //创建线程组
            EventLoopGroup boss = new NioEventLoopGroup();
            EventLoopGroup worker = new NioEventLoopGroup();
    
            try {
                //定义服务端启动类
                ServerBootstrap bootstrap = new ServerBootstrap();
                //设置线程组
                bootstrap.group(boss, worker);
                //使用NioServerSocketChannel
                bootstrap.channel(NioServerSocketChannel.class);
                //位每个连接添加NettyServerHandler 处理器
                bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel)
                            throws Exception {
                        ChannelPipeline p = socketChannel.pipeline();
                        // 添加NettyServerHandler，用来处理Server端接收和处理消息的逻辑
                        p.addLast(new NettyServerHandler());
                    }
                });
                //绑定端口并进行阻塞
                ChannelFuture channelFuture = bootstrap.bind(port).sync();
                if (channelFuture.isSuccess()) {
                    System.err.println("启动Netty服务成功，端口号：" + this.port);
                }
                // 关闭连接之前一直阻塞
                channelFuture.channel().closeFuture().sync();
    
            } catch (Exception e) {
                System.err.println("启动Netty服务异常，异常信息：" + e.getMessage());
                e.printStackTrace();
            } finally {
                //退出的时候关闭线程组
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            //启动服务器
            new NettyServer(10086);
        }
    }

3.新建NettyServerHandler类，用来实现Server端接收和处理消息的逻辑

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import java.io.UnsupportedEncodingException;
    
    //设置共享
    @ChannelHandler.Sharable
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 读取数据
         * @param ctx
         * @param msg
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            //将msg强转成ByteBuf
            ByteBuf buf = (ByteBuf) msg;
            //从ByteBuf中获取消息
            String recieved = getMessage(buf);
            System.err.println("服务器接收到客户端消息：" + recieved);
    
            try {
                //使用ctx进行发送消息
                ctx.writeAndFlush(getSendByteBuf("你好，客户端"));
                System.err.println("服务器回复消息：你好，客户端");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
    
        //客户端端连接
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            //super.channelActive(ctx);
            System.out.println("一台客户端已经连接");
        }
    
        /*
         * 从ByteBuf中获取信息 使用UTF-8编码返回
         */
        private String getMessage(ByteBuf buf) {
    
            byte[] con = new byte[buf.readableBytes()];
            //将buf中的内容读取到con
            buf.readBytes(con);
            try {
                return new String(con, "UTF8");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                return null;
            }
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("一个客户端的关闭了连接");
        }
    
        /**
         * 发生异常
         *
         * @param ctx
         * @param cause
         * @throws Exception
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("发生了一个异常：" + cause.getMessage());
        }
    
        //发送消息
        private ByteBuf getSendByteBuf(String message)
                throws UnsupportedEncodingException {
            //字符串转换成byte数组
            byte[] req = message.getBytes("UTF-8");
            //创建ByteBuf
            ByteBuf pingMessage = Unpooled.buffer();
            //向ByteBuf中写入消息
            pingMessage.writeBytes(req);
            //返回pingMessage
            return pingMessage;
        }
    }

4.启动Server后打印信息

启动Netty服务成功，端口号：10086

5.创建NettyClient类（导包同NettyServer）

public class NettyClient {
    
        /*
         * 服务器端口号
         */
        private int port;
    
        /*
         * 服务器IP
         */
        private String host;
    
        public NettyClient(int port, String host) throws InterruptedException {
            this.port = port;
            this.host = host;
            start();
        }
    
        private void start() throws InterruptedException {
            //创建线程组
            EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
    
            try {
                //定义客户端启动类
                Bootstrap bootstrap = new Bootstrap();
                //设置NioSocketChannel
                bootstrap.channel(NioSocketChannel.class);
                //设置线程组
                bootstrap.group(eventLoopGroup);
                //设置服务器的IP以及端口号
                bootstrap.remoteAddress(host, port);
                //设置处理类
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel)
                            throws Exception {
                        //添加处理类
                        socketChannel.pipeline().addLast(new NettyClientHandler());
                    }
                });
                //连接到服务器并进行阻塞
                ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
                if (channelFuture.isSuccess()) {
                    System.err.println("连接服务器成功");
                }
                //客户端关闭之前一直保持阻塞状态
                channelFuture.channel().closeFuture().sync();
            } finally {
                //客户端关闭的时候关闭线程组
                eventLoopGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new NettyClient(10086, "localhost");
        }
    }

6.新建NettyClientHandler类，用来实现Server端接收和处理消息的逻辑

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.io.UnsupportedEncodingException;
    
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 连接到服务器触发
         *
         * @param ctx
         * @throws Exception
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            byte[] data = "你好，服务器".getBytes();
            ByteBuf firstMessage = Unpooled.buffer();
            firstMessage.writeBytes(data);
            ctx.writeAndFlush(firstMessage);
            System.err.println("客户端发送消息:你好，服务器");
        }
    
        /**
         * 读取数据
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            ByteBuf buf = (ByteBuf) msg;
            String rev = getMessage(buf);
            System.err.println("客户端收到服务器消息:" + rev);
        }
    
    
        private String getMessage(ByteBuf buf) {
            byte[] con = new byte[buf.readableBytes()];
            buf.readBytes(con);
            try {
                return new String(con, "UTF8");
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
                return null;
            }
        }
    }

7.启动Client端，客户端输入如下

连接服务器成功
    客户端发送消息:你好，服务器
    客户端收到服务器消息:你好，客户端

8.服务端输出如下

启动Netty服务成功，端口号：10086
    一台客户端已经连接
    服务器接收到客户端消息：你好，服务器
    服务器回复消息：你好，客户端