Netty -- 简单的服务器

不念不忘少年蓝@ 2024-02-19 18:14 66阅读 0赞

Netty框架，类似于tomcat，把java的socket通信变得简单了，提供了java的tcp通信的很多读数据，返回数据，处理数据的封装方法，是个很好用的搭建TCP服务器的框架，类似于Mina。

下面是网上摘取的很通俗的Netty的架构图：

Netty是典型的Reactor模型结构，在实现上，Netty中的Boss类充当mainReactor，NioWorker类充当subReactor（默认NioWorker的个数是当前服务器的可用核数）。

在处理新来的请求时，NioWorker读完已收到的数据到ChannelBuffer中，之后触发ChannelPipeline中的ChannelHandler流。

Netty是事件驱动的，可以通过ChannelHandler链来控制执行流向。因为ChannelHandler链的执行过程是在subReactor中同步的，所以如果业务处理handler耗时长，将严重影响可支持的并发数。

![19172727-038d2796569a40d5b2c61dc414ed5ab4.png][]

简单的理解就是，原来的socket通信的顺序是，服务器接收数据——服务器处理数据——服务器返回数据。系统在处理数据的时候花的时间比其接收和返回数据多很多，如果中间阻塞了会花更长的时间。这时候我们要优化的话，就再接收到数据的时候，就开启一个新的线程去处理数据，然后服务器就又可以接收新的连接请求，用并行的方法，让服务器可以处理更多的请求，如果我们要自己去实现这些功能，可能要花很多时间，而且隐藏的问题也可能很多，所有Netty框架帮我们做好了这些功能，安全也稳定，可以节约服务器的开发时间和成本。

## 搭建Netty服务器： ##

### 使用maven：pom.xml，添加jar包依赖 ###

<dependencies>
          <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>5.0.0.Alpha2</version>
          </dependency>
      </dependencies>

### 创建服务器Handler类：ServerHandler.java ###

import java.io.UnsupportedEncodingException;
    import java.nio.charset.Charset;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.ReferenceCountUtil;
    
    /**
     * 处理客户端的请求
     * @author zhb
     */
    public class ServerHandler extends ChannelInboundHandlerAdapter {
    
        // 读取数据这,个方法会在收到消息时被调用
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {       
            ByteBuf bb = (ByteBuf)msg;
            // 创建一个和客户端信息同等长度的字节数组
            byte[] reqByte = new byte[bb.readableBytes()];
            // 将客户端信息中的数据读取到数组中
            bb.readBytes(reqByte);
            String reqStr = new String(reqByte, Charset.forName("utf-8"));
            System.err.println("服务器接收到客户端信息： " + reqStr);
            String respStr = "服务端已接收到客户端信息，返回客户端信息！";
            System.out.println("服务器返回客户端信息：" + respStr);
            
            // 返回给客户端响应     
            ctx.writeAndFlush(Unpooled.copiedBuffer(respStr.getBytes("utf-8")));//.addListener(ChannelFutureListener.CLOSE);
            // ReferenceCountUtil.release(msg);//抛弃msg
        }
    
        
        // 数据读取完毕的处理
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.err.println("服务器读取数据完成");
        }
        
        // 出现异常的处理
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.err.println("服务器读取数据异常");
            ctx.close();
        }
    
        
    }

###  ###

### 服务器规则类：ServerNetty.java ###

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;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    /**
     * tcp/ip 服务端用netty实现
     * @author zhb
     *
     */
    public class ServerNetty {
        
        private int port;   //服务器端口
        public ServerNetty(int port){
            this.port = port;
        }
        
        // netty 服务器启动
        public void serverStart() throws InterruptedException{
            
            // 用来接收进来的连接
            EventLoopGroup bossGroup = new NioEventLoopGroup(); 
            // 用来处理已经被接收的连接，一旦bossGroup接收到连接，就会把连接信息注册到workerGroup上
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            
            try {
                // nio服务的启动类
                ServerBootstrap sbs = new ServerBootstrap();
                // 配置nio服务参数
                sbs.group(bossGroup, workerGroup)
                   .channel(NioServerSocketChannel.class) // 说明一个新的Channel如何接收进来的连接
                   .option(ChannelOption.SO_BACKLOG, 128) // tcp最大缓存链接个数
                   .childOption(ChannelOption.SO_KEEPALIVE, true) //保持连接
                   .handler(new LoggingHandler(LogLevel.INFO)) // 日志级别
                   .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {                     
                            // 解码器
    //                      socketChannel.pipeline().addLast(MarshallingCodefactory.buildDecoder());
                            // 编码器
    //                      socketChannel.pipeline().addLast(MarshallingCodefactory.buildEncoder());
                             // 设置网络超时时间
    //                      socketChannel.pipeline().addLast(new ReadTimeoutHandler(5));
                            // 处理接收到的请求
                            socketChannel.pipeline().addLast(new ServerHandler()); // 这里相当于过滤器，可以配置多个
                        }
                   });
                
                System.err.println("server 开启--------------");
                // 绑定端口，开始接受链接
                ChannelFuture cf = sbs.bind(port).sync();
                
                // 开多个端口
    //          ChannelFuture cf2 = sbs.bind(3333).sync();
    //          cf2.channel().closeFuture().sync();
                
                // 等待服务端口的关闭；在这个例子中不会发生，但你可以优雅实现；关闭你的服务
                cf.channel().closeFuture().sync();
            } finally{
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }           
        }
        
            
        // 开启netty服务器
        public static void main(String[] args) throws InterruptedException {
            new ServerNetty(8080).serverStart();
        }
            
    
    }

### 右键运行main方法，启动Netty服务器： ###

server 开启--------------
    三月 04, 2019 5:14:43 下午 io.netty.handler.logging.LoggingHandler channelRegistered
    信息: [id: 0xe8854b52] REGISTERED
    三月 04, 2019 5:14:43 下午 io.netty.handler.logging.LoggingHandler bind
    信息: [id: 0xe8854b52] BIND: 0.0.0.0/0.0.0.0:8080
    三月 04, 2019 5:14:43 下午 io.netty.handler.logging.LoggingHandler channelActive
    信息: [id: 0xe8854b52, /0:0:0:0:0:0:0:0:8080] ACTIVE

## 创建客户端： ##

**客户端就不用Netty了，直接用socket创建，个人感觉用Netty创建客户端意义不大，节约不了多少工作量，还不通用。**

**TestClient.java：**

import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.io.IOException;
    import java.net.Socket;
     
    public class TestClient {
     
    	public static final String IP_ADDR = "localhost";//服务器地址   
        public static final int PORT = 8080;//服务器端口号    
          
        public static void main(String[] args) {    
            System.out.println("客户端启动...");      
           // while (true) {    
                Socket socket = null;  
                try {  
                    //创建一个流套接字并将其连接到指定主机上的指定端口号  
                    socket = new Socket(IP_ADDR, PORT);    
                  
                    //向服务器端发送数据    
                    DataOutputStream out = new DataOutputStream(socket.getOutputStream());    
                    String Msg = "client send to server";
                    System.out.println("客户端数据长度" + Msg.length());
                    System.out.println("客户端发送数据：" + Msg);
                    out.write(Msg.getBytes() , 0 , Msg.length());
                    out.flush(); 
                    //out.close(); 
                    
                    //读取服务器端数据    
                    DataInputStream input = new DataInputStream(socket.getInputStream());    
                    
                    StringBuffer returnMsg = new StringBuffer();
                    int len = 0;
                    byte[] b = new byte[1024]; //容器，存放数据
                    System.out.println("开始接收服务端返回数据。。。");
                    while ((len = input.read(b)) != -1) {//一直读，读到没数据为止
                    	returnMsg.append(new String(b, 0, len, "utf-8"));
                    	if (len < 1024) {//如果读的长度小于1024，说明是最后一次读,后面已经没有数据，跳出循环
    						break;
    					}
                    }  
                    System.out.println("服务器端返回过来的是: " + returnMsg.toString());    
                    out.close(); 
                    input.close();  
                } catch (Exception e) {  
                    System.out.println("客户端异常:" + e.getMessage());   
                } finally {  
                    if (socket != null) {  
                        try {  
                            socket.close();  
                        } catch (IOException e) {  
                            socket = null;   
                            System.out.println("客户端 finally 异常:" + e.getMessage());   
                        }  
                    }  
                }  
        }   
    }

### 右键运行main方法：测试结果： ###

#### 客户端： ####

客户端启动...
    客户端数据长度21
    客户端发送数据：client send to server
    开始接收服务端返回数据。。。
    服务器端返回过来的是: 服务端已接收到客户端信息，返回客户端信息！

#### 服务端： ####

server 开启--------------
    三月 04, 2019 5:17:58 下午 io.netty.handler.logging.LoggingHandler channelRegistered
    信息: [id: 0xae2aa3a7] REGISTERED
    三月 04, 2019 5:17:58 下午 io.netty.handler.logging.LoggingHandler bind
    信息: [id: 0xae2aa3a7] BIND: 0.0.0.0/0.0.0.0:8080
    三月 04, 2019 5:17:58 下午 io.netty.handler.logging.LoggingHandler channelActive
    信息: [id: 0xae2aa3a7, /0:0:0:0:0:0:0:0:8080] ACTIVE
    三月 04, 2019 5:18:02 下午 io.netty.handler.logging.LoggingHandler channelRead
    信息: [id: 0xae2aa3a7, /0:0:0:0:0:0:0:0:8080] RECEIVED: [id: 0xe1d5ea02, /127.0.0.1:9505 => /127.0.0.1:8080]
    服务器接收到客户端信息： client send to server
    服务器返回客户端信息：服务端已接收到客户端信息，返回客户端信息！
    服务器读取数据完成
    服务器读取数据完成

[19172727-038d2796569a40d5b2c61dc414ed5ab4.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/0686eb6d625249d29a966bdfbd7cf1e7.png