netty学习（1）:1个客户端与服务器通信

╰+攻爆jí腚メ 2023-10-12 22:31 34阅读 0赞

### 1. 新建maven工程，添加netty依赖 ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>nettyTest</artifactId>
        <version>1.0-SNAPSHOT</version>
        <build>
            <plugins>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <configuration>
                        <source>8</source>
                        <target>8</target>
                    </configuration>
                </plugin>
            </plugins>
        </build>
    
        <dependencies>
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.34.Final</version>
            </dependency>
        </dependencies>
    </project>

### 2. 新建netty服务器 ###

package server;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    import java.util.concurrent.ConcurrentHashMap;
    
    public class NettyService {
        
        /**
         * 1.创建BossGroup线程组，处理网络连接事件
         * 2.创建workerGroup线程组 处理网络读写事件
         * 3.创建服务端启动助手，serverBootStrap
         * 4.设置服务端通道实现方式为NIO
         * 5.设置服务端options
         * 6.创建通道初始化对象
         * 8.向pipeline中添加自定义业务初处理逻辑handler
         * 9.启动服务端并绑定端口，将异步改为同步
         * 10.关闭通道和连接池
         */
    
        //监听的端口号
        private final int port;
        public NettyService(int port) {
        
            this.port = port;
        }
    
        public void createNettyService() {
        
            //创建两个线程组 boosGroup、workerGroup,这两个都是无限循环,boos是负责进行连接请求的接收accept事件，而worker则是负责业务处理
            //只是处理连接请求accept，含有的子线程有多少个呢，NioEventLoop个数，默认是cpu的核数*2
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            //真正的和客户端进行业务处理，含有的子线程有多少个呢，NioEventLoop个数，默认是cpu的核数*2
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
        
                //创建服务端的启动对象，设置参数
                ServerBootstrap bootstrap = new ServerBootstrap();
                //设置两个线程组boosGroup和workerGroup
                bootstrap.group(bossGroup, workerGroup)
                        //设置服务端通道实现类型
                        .channel(NioServerSocketChannel.class) //使用nioserversocketchannel作为通道实现
                        //设置线程队列得到连接个数
                        .option(ChannelOption.SO_BACKLOG, 128) //设置线程队列等待连接个数
                        //设置保持活动连接状态
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        //使用匿名内部类的形式初始化通道对象
                        .childHandler(new ChannelInitializer<SocketChannel>() {
        
                            //这里会进行客户端业务处理
                            //通过一个特殊的ChannelInboundHandler 来初始化注册到EventLoop的Channel
                            @Override
                            protected void initChannel(SocketChannel sc) throws Exception {
        
                                //给pipeline管道设置处理器
                                ChannelPipeline pipeline = sc.pipeline();
                                //添加一个基于行的解码器
                                pipeline.addLast(new LineBasedFrameDecoder(2048));
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(new StringEncoder());
                                //这里加入了自定义的handler
                                pipeline.addLast(new NettyServiceHandler());
                            }
                        });//给workerGroup的EventLoop对应的管道设置处理器
                //绑定端口号，启动服务端
                ChannelFuture channelFuture = bootstrap.bind(port).sync();
                System.out.println("netty服务端已经准备就绪,端口" + port);
                //对关闭通道进行监听
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } finally {
        
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

自定义handler

package server;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.*;
    import io.netty.channel.group.ChannelGroup;
    import io.netty.channel.group.DefaultChannelGroup;
    import io.netty.util.CharsetUtil;
    import io.netty.util.concurrent.GlobalEventExecutor;
    
    /**
     * 自定义处理Handler
     */
    public class NettyServiceHandler extends SimpleChannelInboundHandler<String> {
        
        // 创建一个ChannelGroup，其是一个线程安全的集合，其中存放着与当前服务器相连接的所有Active状态的Channel
        // GlobalEventExecutor是一个单例、单线程的EventExecutor，是为了保证对当前group中的所有Channel的处理
        // 线程是同一个线程
        //private static ChannelGroup group = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    
        // 只要有客户端Channel与服务端连接成功就会执行这个方法
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            // 获取到当前与服务器连接成功的channel
            Channel channel = ctx.channel();
            System.out.println(channel.remoteAddress() + "---上线");
            //group.writeAndFlush(channel.remoteAddress() + "---上线\n");
            // 将当前channel添加到group中
            //group.add(channel);
    
    //        NettyService.remoteAddressMap.put(channel.remoteAddress().toString(), ctx);
        }
    
        // 只要有客户端Channel断开与服务端的连接就会执行这个方法
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        
            // 获取到当前要断开连接的Channel
            Channel channel = ctx.channel();
            System.out.println(channel.remoteAddress() + "------下线");
            //group.writeAndFlush(channel.remoteAddress() + "下线，当前在线人数：" + group.size() + "\n");
    
            // group中存放的都是Active状态的Channel，一旦某Channel的状态不再是Active，
            // group会自动将其从集合中踢出，所以，下面的语句不用写
            // remove()方法的应用场景是，将一个Active状态的channel移出group时使用
            // group.remove(channel);
        }
    
        /**
         * channelRead，这个方法当有数据读写的时候，会触发，可以读取客户端的消息
         * 只要有客户端Channel给当前的服务端发送了消息，那么就会触发该方法的执行
         */
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        
            Channel channel = ctx.channel();
            System.out.println("netty客户端" + channel.remoteAddress() + "发送过来的消息:" + msg);
            channel.writeAndFlush("自己发的消息：" + msg + "\n");
            
    //        String[] split = msg.split(":");
    //        if (split.length >= 2) { //指定客户端时发送给指定客户
    //            System.out.println("发送给客户:" + split[0]);
    //            if (NettyService.userIdMap.get(split[0]) != null) {
        
    //                NettyService.userIdMap.get(split[0]).writeAndFlush(channel.remoteAddress() + "：" + split[1] + "\n");
    //            } else {
        
    //                NettyService.userIdMap.put(split[0], ctx);  //第一次发送消息时注册客户端的channel
    //            }
    //        } else { //否则发送给所有的客户端
    //            System.out.println("广播");
    //            // 遍历channelGroup，从而区分“我”和“别人”发出的消息，如果消息是自己发出的就显示“我”
    //            group.forEach(ch -> { // JDK8 提供的lambda表达式
    //                if (ch != channel) {
        
    //                    ch.writeAndFlush(channel.remoteAddress() + "：" + msg + "\n");
    //                } else {
        
    //                    channel.writeAndFlush("自己发的消息：" + msg + "\n");
    //                }
    //            });
    //        }
        }
        /**
         * channelReadComplete，数据读取完毕之后，需要做的业务操作，回消息
         */
    //    public void channelReadComplete(ChannelHandlerContext channelHandlerContext) throws Exception {
        
    //        //消息出站
    //        System.out.println("Netty服务端读取消息完毕");
    //        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("--over", CharsetUtil.UTF_8));
    //    }
    
        /**
         * exceptionCaught，发生异常的handler
         */
        public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        
            throwable.printStackTrace();
            channelHandlerContext.close();
        }
    }

### 3. 新建客户端 ###

package client;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.*;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    
    import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStreamReader;
    import java.nio.charset.StandardCharsets;
    
    public class NettyClient {
        
        /**
         * 1.创建线程组
         * 2.设置线程组启动助手 bootstrap
         * 3.设置客户端通道为NIO
         * 4.创建通道初始化对象
         * 5.向pipeline中添加自定义业务处理的handler
         * 6.启动客户端，等待链接服务端，同时将异步改为同步
         * 8.关闭通道和连接池
         */
        private final String host;
        private final int port;
        public NettyClient(String host, int port) {
        
            this.host = host;
            this.port = port;
        }
    
        public void createNettyClient() {
        
            //客户端跟服务端有些许区别，第一个就是不需要两个loopgroup了，不需要使用boosgroup了，所以只需要新建一个loopgroup来处理业务就行了
            EventLoopGroup group = new NioEventLoopGroup();
            try {
        
                //第二点就是不是使用serverBootstrap，而是使用bootstrap
                Bootstrap bootstrap = new Bootstrap();
                //设置线程组
                bootstrap.group(group)
                        .channel(NioSocketChannel.class) //设置客户端的通道实现类型
                        .handler(new ChannelInitializer<SocketChannel>() {
        
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
        
                                ChannelPipeline pipeline = ch.pipeline();
                                pipeline.addLast(new LineBasedFrameDecoder(2048));
                                pipeline.addLast(new StringDecoder());
                                pipeline.addLast(new StringEncoder());
                                pipeline.addLast(new NettyClientHandler()); //加入自己的处理器
                            }
                        });
                //启动客户端
                ChannelFuture future = bootstrap.connect(host, port).sync();
                System.out.println("netty客户端1准备就绪");
                // 获取键盘输入
                InputStreamReader is = new InputStreamReader(System.in, StandardCharsets.UTF_8);
                BufferedReader br = new BufferedReader(is);
                // 将输入的内容写入到Channel
                while (true) {
        
                    //br.readLine()中执行fill()方法获取输入数据，获取不到时会发生阻塞，直到获取到数据为止
                    future.channel().writeAndFlush(br.readLine() + "\r\n");
                }
    
                //监听通道关闭的状态事件
                //channel.closeFuture().sync();
            } catch (InterruptedException | IOException e) {
        
                e.printStackTrace();
            } finally {
        
                //关闭通道和连接池
                group.shutdownGracefully();
            }
        }
    }

自定义handler

package client;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /**
     * 自定义处理Handler
     */
    public class NettyClientHandler extends SimpleChannelInboundHandler<String> {
        
        /**
         * channelActive，当通道就绪，就会触发该方法
         */
    //    public void channelActive(ChannelHandlerContext channelHandlerContext) throws Exception {
        
    //        //发送消息到服务端
    //        channelHandlerContext.writeAndFlush(Unpooled.copiedBuffer("netty客户端1启动.", CharsetUtil.UTF_8));
    //    }
    
    //    public void channelInactive(ChannelHandlerContext channelHandlerContext) throws Exception {
        
    //
    //    }
        /**
         * channelRead，当通道有读取事件时，会触发
         */
        public void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        //接收服务端发送过来的消息
            System.out.println("收到服务端消息：" + channelHandlerContext.channel().remoteAddress() + "的消息：" + msg);
        }
    
        /**
         * exceptionCaught，当有异常的就会触发
         */
        public void exceptionCaught(ChannelHandlerContext channelHandlerContext, Throwable throwable) throws Exception {
        
            channelHandlerContext.close();
        }
    }

### 4. 测试 ###

启动服务器

package test;
    
    import server.NettyService;
    
    public class Service {
        
        public static void main(String[] args) {
        
            new NettyService(6666).createNettyService();
        }
    }

启动客户端

package test;
    
    import client.NettyClient;
    
    public class Client1 {
        
        public static void main(String[] args) {
        
            new NettyClient("127.0.0.1",6666).createNettyClient();
        }
    }

![在这里插入图片描述][281755f902f7485d84681827ffc9dbe9.png]  
![在这里插入图片描述][1c59a1357ef946ebbfbfc99be948049a.png]

[281755f902f7485d84681827ffc9dbe9.png]: https://img-blog.csdnimg.cn/281755f902f7485d84681827ffc9dbe9.png
[1c59a1357ef946ebbfbfc99be948049a.png]: https://img-blog.csdnimg.cn/1c59a1357ef946ebbfbfc99be948049a.png