Netty通信

小咪咪 2022-07-16 11:43 161阅读 0赞

## Netty简介 ##

Netty是用来开发高性能、高可靠性的网络服务器和客户端程序，主要是用作通信接口  
特点：

*  吞吐大数据量
 *  I/O超时和idle状态检测

对比网络传输方式：传统的RPC框架或者基于RMI等方式的远程服务（过程）调用采用了同步阻塞IO，当客户端的并发压力或者网络时延增大之后，同步阻塞IO会由于频繁的wait导致IO线程经常性的阻塞，由于线程无法高效的工作，IO处理能力自然下降。

对比序列化方式问题：  
Java序列化存在如下几个典型问题：

1) Java序列化机制是Java内部的一种对象编解码技术，无法跨语言使用；例如对于异构系统之间的对接，Java序列化后的码流需要能够通过其它语言反序列化成原始对象（副本），目前很难支持；

2) 相比于其它开源的序列化框架，Java序列化后的码流太大，无论是网络传输还是持久化到磁盘，都会导致额外的资源占用；

3) 序列化性能差（CPU资源占用高）。

线程模型问题：由于采用同步阻塞IO，这会导致每个TCP连接都占用1个线程，由于线程资源是JVM虚拟机非常宝贵的资源，当IO读写阻塞导致线程无法及时释放时，会导致系统性能急剧下降，严重的甚至会导致虚拟机无法创建新的线程。

## 准备工作 ##

*  方式一 ： 进入Netty官网的下载页面下载：  
    [http://netty.io/downloads.html][http_netty.io_downloads.html]  
    下载以后解压，有JAR包全部分开的，也有all-in-one的，这里用all-in-one就行了
 *  方式二： 使用Maven，如下配置

<dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty</artifactId> 
        <version>X.Y.Z.Q</version>
        <scope>compile</scope>
      </dependency>

## 实例 ##

首先需要先有个Handler继承SimpleChannelInboundHandler  
TcpServerHandler.java

package org.netty.test2;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /** * @author NICK * */
    public class TcpServerHandler extends SimpleChannelInboundHandler<Object> { 
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            System.out.println("SERVER接收到消息:" + msg);
            ctx.channel()
                    .writeAndFlush("yes, server is accepted you ,nice !" + msg);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            System.out.println("Unexpected exception from downstream.");
            ctx.close();
        }
    
    }

编写服务端  
TcpServer.java

package org.netty.test2;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelInitializer;
    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;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    
    /** * @author NICK * */
    public class TcpServer { 
    
        private static final String IP = "127.0.0.1";
        private static final int PORT = 8088;
        /** 用于分配处理业务线程的线程组个数 */
        protected static final int BIZGROUPSIZE = Runtime.getRuntime()
                .availableProcessors() * 2; // 默认
        /** 业务出现线程大小 */
        protected static final int BIZTHREADSIZE = 4;
        /* * NioEventLoopGroup实际上就是个线程池, * NioEventLoopGroup在后台启动了n个NioEventLoop来处理Channel事件, * 每一个NioEventLoop负责处理m个Channel, * NioEventLoopGroup从NioEventLoop数组里挨个取出NioEventLoop来处理Channel */
        private static final EventLoopGroup bossGroup = new NioEventLoopGroup(
                BIZGROUPSIZE);
        private static final EventLoopGroup workerGroup = new NioEventLoopGroup(
                BIZTHREADSIZE);
    
        /** * 运行 * @throws Exception */
        protected static void run() throws Exception {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup);
            b.channel(NioServerSocketChannel.class);
            b.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("frameDecoder",
                            new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0,
                                    4, 0, 4));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("decoder",
                            new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("encoder",
                            new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast(new TcpServerHandler());
                }
            });
    
            b.bind(IP, PORT).sync();
            System.out.println("TCP服务器已启动");
        }
    
        /** * 关闭 */
        protected static void shutdown() {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    
        public static void main(String[] args) throws Exception {
            System.out.println("开始启动TCP服务器...");
            TcpServer.run();
            // TcpServer.shutdown();
        }
    
    }

启动Main函数，这时候服务端已经启动起来了，验证是否成功，可以在DOS下面执行命令：  
telnet 127.0.0.1 PORT

客户端

TcpClientHandler.java

TcpClient.java

package org.netty.test2;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    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.nio.NioSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.codec.string.StringEncoder;
    import io.netty.util.CharsetUtil;
    
    /** * @author NICK * */
    public class TcpClient { 
        public static String HOST = "127.0.0.1";
        public static int PORT = 8088;
    
        public static Bootstrap bootstrap = getBootstrap();
        public static Channel channel = getChannel(HOST, PORT);
    
        /** * 初始化 * * @return */
        public static final Bootstrap getBootstrap() {
            EventLoopGroup group = new NioEventLoopGroup();
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class);
            b.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    //初始化通道
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("frameDecoder",
                            new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0,
                                    4, 0, 4));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("decoder",
                            new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("encoder",
                            new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("handler", new TcpClientHandler());
                }
            });
            b.option(ChannelOption.SO_KEEPALIVE, true);
            return b;
        }
    
    
        /** * @param host IP * @param port 端口 * @return 连接，获取通道 */
        public static final Channel getChannel(String host, int port) {
            Channel channel = null;
            try {
                channel = bootstrap.connect(host, port).sync().channel();
            } catch (Exception e) {
                e.printStackTrace();
                System.out.println(String.format("连接Server(IP[%s],PORT[%s])失败",
                        host, port));
                return null;
            }
            return channel;
        }
    
        /** * @param msg 数据 * @throws Exception 异常 */
        public static void sendMsg(String msg) throws Exception {
            if (channel != null) {
                channel.writeAndFlush(msg).sync();
            } else {
                System.out.println("消息发送失败,连接尚未建立!");
            }
        }
    
        public static void main(String[] args) throws Exception {
            try {
                long t0 = System.nanoTime();
                for (int i = 0; i < 100000; i++) {
                    TcpClient.sendMsg(i + "你好1");
                    Thread.sleep(3000);
                }
                long t1 = System.nanoTime();
                System.out.println((t1 - t0) / 1000000.0);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

运行Main函数，打印信息。  
表示链接服务端，可以运行多个Main函数，表示有多个客户端，这点跟Socket类似

[http_netty.io_downloads.html]: http://netty.io/downloads.html