netty入门

netty是什么？

Netty 是基于 Java NIO 的异步事件驱动的网络应用框架，使用 Netty 可以快速开发网络应用，Netty 提供了高层次的抽象来简化 TCP 和 UDP 服务器的编程，但是你仍然可以使用底层的 API。
Netty 的内部实现是很复杂的，但是 Netty 提供了简单易用的API从网络处理代码中解耦业务逻辑。Netty 是完全基于 NIO 实现的，所以整个 Netty 都是异步的。
Netty 是最流行的 NIO 框架，它已经得到成百上千的商业、商用项目验证，许多框架和开源组件的底层 rpc 都是使用的 Netty，如 Dubbo、Elasticsearch 等等

netty的应用

现在市面上Dubbo、zk、RocketMQ、ElasticSearch、Spring5(对 HTTP 协议的实现)、GRpc、Spark 等大型开源项目都在使用 Netty 作为底层通讯框架。

牛刀小试

在这里我们就不对底层的TCP、UDP传输协议做过多的阐述，直接进入代码
在这里我们先通过下面的程序达到的目的是，对 Netty 编程的基本结构及流程有所了解。该程序是通过 Netty 实现 HTTP 请求的处理，即接收 HTTP 请求，返回 HTTP 响应。

创建工程

首先创建一个普通的 Maven 的 Java 工程。

导入依赖

<dependencies>
    <dependency>
        <groupId>io.netty</groupId>
        <artifactId>netty-all</artifactId>
        <version>4.1.36.Final</version>
    </dependency>
    <!--lombok依赖-->
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <version>1.18.6</version>
        <scope>provided</scope>
    </dependency>
</dependencies>

创建服务器启动类

创建服务端启动类

/**
 * @Description: 服务启动类
 * @author: dy
 */
public class FirstServer {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup parentGroup = new NioEventLoopGroup();
        NioEventLoopGroup childGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(parentGroup, childGroup)
                     .channel(NioServerSocketChannel.class)
                     .option(ChannelOption.SO_BACKLOG, 1024)
                    //接收套接字缓冲区大小
                    .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                    //发送套接字缓冲区大小
                    .option(ChannelOption.SO_SNDBUF, 1024 * 1024)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // StringEncoder：字符串编码器，将String编码为将要发送到Channel中的ByteBuf
                            pipeline.addLast(new StringEncoder(Charset.forName("UTF-8")));
                            // StringDecoder：字符串解码器，将Channel中的ByteBuf数据解码为String
                            pipeline.addLast(new StringDecoder(Charset.forName("UTF-8")));
                            //绑定处理器（可绑定多个）
                            pipeline.addLast(new HeartServerHandler()); //处理心跳
                            pipeline.addLast(new BusinessServerHandler()); //处理业务
                        }
                    });
            ChannelFuture future = bootstrap.bind(8888).sync();
            future.channel().closeFuture().sync();
            System.out.println("=======>>服务器已启动");
        } finally {
            parentGroup.shutdownGracefully();
            childGroup.shutdownGracefully();
        }
    }
}

创建服务端心跳处理器

/**
 * @Description: 心跳处理器
 * @author: dy
 */
public class HeartServerHandler extends ChannelInboundHandlerAdapter {


    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        // 将来自于客户端的数据显示在服务端控制台
        System.out.println(ctx.channel().remoteAddress() + "==消息内容==>>" + msg);
        Message msgObj = JSONObject.parseObject(msg.toString(), Message.class);
        if(msgObj.getHeader().getMessageId() == 0){
            // 心跳消息
            // 向客户端返回心跳响应数据
            System.out.println("收到心跳请求======");
            Message heartBeat = Message.builder().header(Message.Header.builder().messageId(msgObj.getHeader().getMessageId()).build())
                    .body(Message.MessageBody.builder().data("我是服务端,收到了你的心跳请求").build())
                    .build();
            String json = JSON.toJSONString(heartBeat);
            ByteBuf buf = Unpooled.copiedBuffer(json.getBytes(Charset.forName("UTF-8")));
            ctx.channel().writeAndFlush(buf);
        }else{
            //不处理传递到下一个处理器
            System.out.println("这是心跳请求处理器，不处理此消息");
            ctx.fireChannelRead(msgObj);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

}

创建服务端业务处理器

/**
 * @Description:业务处理器
 * @author: dy
 */
public class BusinessServerHandler extends SimpleChannelInboundHandler<Message> {


    @Override
    public void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception {
        // 将来自于客户端的数据显示在服务端控制台
        System.out.println(ctx.channel().remoteAddress() + "收到业务消息====>>" + msg);
        //处理具体的业务消息
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }

}

创建消息实体类

/**
 * @Description: 消息实体类
 * @author: dy
 */
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class Message  implements Serializable {


    /**
     * 消息头
     */
    private Header header;

    /**
     * 消息体
     */
    private MessageBody body;

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Header implements Serializable {


        private static final long serialVersionUID = 6819375880619930921L;
        /**
         * 消息任务ID
         */
        private int messageId;
        /**
         * 协议版本号
         */
        private int version = 1;
        /**
         * 消息类型
         */
        private int type;
        /**
         * 消息来源地址
         */
        private String remoteAddress;

    }

    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public static class MessageBody implements Serializable {

        private static final long serialVersionUID = 6919544256603837227L;

        /**
         * 业务逻辑是否处理成功
         */
        private Boolean isSuccess ;
        /**
         * 业务处理消息
         */
        private String message;
        /**
         * 消息内容
         */
        private Object data;
    }
}

创建客户端服务启动类

创建客户端启动类

/**
 * @Description: 客户端启动类
 * @author: dy
 */
public class FirstClient {
    public static void main(String[] args) throws InterruptedException {
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    //接收套接字缓冲区大小
                    .option(ChannelOption.SO_RCVBUF, 1024 * 1024)
                    //发送套接字缓冲区大小
                    .option(ChannelOption.SO_SNDBUF, 1024 * 1024)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.TCP_NODELAY, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new StringDecoder(Charset.forName("UTF-8")));
                            pipeline.addLast(new StringEncoder(Charset.forName("UTF-8")));
                            pipeline.addLast(new FirstClientHandler());
                        }
                    });
            ChannelFuture future = bootstrap.connect("localhost", 8888).sync();
            future.channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

创建客户端消息处理器

/**
 * @Description:客户端消息处理类
 * @author: dy
 */
public class FirstClientHandler extends SimpleChannelInboundHandler<String> {

    // msg的消息类型与类中的泛型类型是一致的
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + "收到服务端请求====>>" + msg);
        Message busBeat = Message.builder().header(Message.Header.builder().messageId(1).build())
                .body(Message.MessageBody.builder().data("我是客户端,请处理xxx消息").build())
                .build();
        String json = JSON.toJSONString(busBeat);
        ByteBuf buf = Unpooled.copiedBuffer((json).getBytes(Charset.forName("UTF-8")));
        ctx.channel().writeAndFlush(buf);
    }

    // 当Channel被激活后会触发该方法的执行
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("========");
        Message heartBeat = Message.builder().header(Message.Header.builder().messageId(0).build())
                .body(Message.MessageBody.builder().data("我是客户端，我来了").build())
                .build();
        String json = JSON.toJSONString(heartBeat);
        ByteBuf buf = Unpooled.copiedBuffer((json).getBytes(Charset.forName("UTF-8")));
        ctx.channel().writeAndFlush(buf);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

通过上面的代码我们大体可以画出netty的执行流程如下图所示：

netty执行流程.png

从上面我们使用netty，我们来看下其中比较重要的几个组件，来简单介绍下

Channel

字面意思就是管道，它本质上是对 Socket 的封装，其包含了一组 API，大大简化了直接与 Socket 进行操作的复杂性。也可以理解为一个连接网络输入和IO处理的桥梁。你可以通过Channel来判断当前的状态，是open还是connected，还可以判断当前Channel支持的IO操作，还可以使用ChannelPipeline对Channel中的消息进行处理。

EventLoopGroup

EventLoopGroup 是一个 EventLoop 池，包含很多的 EventLoop。Netty 为每个 Channel 分配了一个 EventLoop，用于处理用户连接请求、对用户请求的处理等所有事件。EventLoop 本身只是一个线程驱动，在其生命周期内只会绑定一个线程，让该线程处理一个 Channel 的所有 IO 事件。一个 Channel 一旦与一个 EventLoop 相绑定，那么在 Channel 的整个生命周期内是不能改变的。一个 EventLoop 可以与多个 Channel 绑定。即 Channel 与 EventLoop 的关系是 n:1，而 EventLoop 与线程的关系是 1:1

ServerBootStrap

用于配置整个 Netty 代码，将各个组件关联起来。服务端使用的是 ServerBootStrap，而客户端使用的是则 BootStrap。

ChannelHandler与 ChannelPipeline

ChannelHandler 是对 Channel 中数据的处理器，这些处理器可以是系统本身定义好的编解码器，也可以是用户自定义的。这些处理器会被统一添加到一个 ChannelPipeline 的对象中，然后按照添加的顺序对 Channel 中的数据进行依次处理。

ChannelFuture

Netty 中所有的 I/O 操作都是异步的，即操作不会立即得到返回结果，所以 Netty 中定义了一个ChannelFuture 对象作为这个异步操作的“代言人”，表示异步操作本身。如果想获取到该异步操作的返回值，可以通过该异步操作对象的 addListener()方法为该异步操作添加监 NIO 网络编程框听器，为其注册回调：当结果出来后马上调用执行。Netty 的异步编程模型都是建立在 Future 与回调概念之上的

在这里我们着重说一下SimpleChannelInboundHandler

SimpleChannelInboundHandler

从SimpleChannelInboundHandler的源码我们可以看到SimpleChannelInboundHandler是继承自ChannelInboundHandlerAdapter，我们来看下SimpleChannelInboundHandler的channelRead方法：

//读取消息对象  
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    boolean release = true;
    try {
        if (acceptInboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            I imsg = (I) msg;
            //如果消息属于本Handler可以处理的消息类型，则委托给channelRead0  
            channelRead0(ctx, imsg);
        } else {
            release = false;
            //当前通道Handler，不可处理消息，通过通道上下文，通知管道线中的下一个通道处理器，接受到一个消息  
            ctx.fireChannelRead(msg);
        }
    } finally {
        if (autoRelease && release) {
            //如果autoRelease为自动释放消息，且消息已处理则释放消息(注意 如果消息实现了ReferenceCounted，则调用ReferenceCounted#release()，如果不是什么都不做) 
            ReferenceCountUtil.release(msg);
        }
    }
}
//ReferenceCountUtil的release方法
public static boolean release(Object msg) {
        if (msg instanceof ReferenceCounted) {
            return ((ReferenceCounted) msg).release();
        }
        return false;
}

从源码可以看出Inbound通道处理器SimpleChannelInboundHandler<I>，内部有连个变量一个为参数类型匹配器，用来判断通道是否可以处理消息，另一个变量autoRelease，用于控制是否在通道处理消息完毕时，释放消息。读取方法channelRead，首先判断跟定的消息类型是否可以被处理，如果是，则委托给channelRead0，channelRead0待子类实现；如果返回false，则将消息转递给Channel管道线的下一个通道处理器；最后，如果autoRelease为自动释放消息，且消息已处理则释放消息。

所以ChannelInboundHandlerAdapter和SimpleChannelInboundHandler最主要的区别是：

SimpleChannelInboundHandler中的channelRead()方法会自动释放接收到的来自于对方的msg

所占有的所有资源。

ChannelInboundHandlerAdapter 中的 channelRead()方法不会自动释放接收到的来自于对方的

msg

所以在用的时候我们分情况：

若对方没有向自己发送数据，则自定义处理器建议继承自ChannelInboundHandlerAdapter。因为若继承自 SimpleChannelInboundHandler 需要重写channelRead0()方法。而重写该方法的目的是对来自于对方的数据进行处理。因为对方根本就没有发送数据，所以也就没有必要重写 channelRead0()方法。
若对方向自己发送了数据，而自己又需要将该数据再发送给对方，则自定义处理器建议 NIO 网络编程框架 Netty继承自 ChannelInboundHandlerAdapter。因为 write()方法的执行是异步的，且SimpleChannelInboundHandler 中的 channelRead()方法会自动释放掉来自于对方的 msg。若 write()方法中正在处理 msg，而此时 SimpleChannelInboundHandler 中的 channelRead()方法执行完毕了，将 msg 给释放了。此时就会报错。

这篇文章我们先简单介绍下netty的使用和一些简单的介绍，下篇文章我们来介绍下netty的粘包和拆包处理

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 204,189评论 6赞 478
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 85,577评论 2赞 381
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 150,857评论 0赞 337
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 54,703评论 1赞 276
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 63,705评论 5赞 366
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 48,620评论 1赞 281
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 37,995评论 3赞 396
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 36,656评论 0赞 258
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 40,898评论 1赞 298
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,639评论 2赞 321
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 37,720评论 1赞 330
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,395评论 4赞 319
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 38,982评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 29,953评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,195评论 1赞 260
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 44,907评论 2赞 349
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,472评论 2赞 342