网络编程之Netty

原文链接

概述

Netty 是一个高性能、异步事件驱动的 NIO 框架

优点

1、高性能、高吞吐量、低延迟
2、使用方便，详细的Javadoc,用户指南和示例
3、社区活跃，版本迭代周期短
4、设计优雅，适用于各种传输类型的统一API阻塞和非阻塞Socket；基于灵活且可扩展的事件模型

线程模型介绍

目前的线程模型有：传统阻塞I/O服务模型、Reactor模型；
根据Reactor数量和处理资源线程的的数量不同，有3种形式：
1、单Reactor
2、单Reactor多线程
3、主从Reactor多线程
Netty主要基于主从Reactor多线程模型做了一定的改进，其中主从Reactor多线程模型有多个Reactor

传统阻塞I/O服务模型

特点
1、采用阻塞IO模式获取数据输入的数据
2、每个连接都需要独立的线程完成数据的输入，业务处理，数据返回
问题
1、当并发数很大时，会创建大量的线程，占用很大的系统资源
2、连接创建后，如果当前线程暂时没有数据可读，该线程会阻塞在read操作，造成资源浪费

注：黄色框表示对象，蓝色框表示线程，白色框表示方法

Reactor模式（反应器模式/分发者模式/通知者模式）

针对传统阻塞I/O服务模型的2个缺点的解决方案：
1、基于I/O 复用模型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有的连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。
2、基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。
复用结合线程池，就是Reactor模式基本设计思想

说明
1、Reactor模式，通过一个或多个输入同时传递给服务处理器的模式（基于事件驱动）
2、服务端程序处理传入的多个请求，并将它们同步分派到相应的处理器线程，因此Reactor模式也叫Dispatcher模式
3、Reactor模式使用IO复用监听事件，收到事件后，分发给某个线程(进程)，这就是网络服务器处理高并发的关键

Reactor模式的优点

1、响应快，不必为单个同步时间所阻塞，虽然Reactor本身依然是同步的
2、可以最大程度的避免复杂的多线程问题及同步问题，并且避免了多线程/进程的切换开销
3、扩展性好，可以方便的通过增加Reactor实例个数来充分利用CPU资源
4、复用性好，Reactor模型本身与事件处理逻辑无关，具有很高的复用性

单Reactor单线程

1、Select 是前面I/O复用模型的标准网络编程API，可以实现应用程序通过一个阻塞对象监听多路连接请求
2、Reactor对象通过Select监控客户端请求事件，收到事件后通过Dispatch进行分发
3、如果是建立连接请求事件，则由Accepor通过Accept处理连接请求，然后创建一个Handler对象连接完成后的后续业务处理
4、如果不是建立连接事件，则Reactor会分发调用连接对象对应的Handler来响应
5、Handler会完成Read->业务处理->send的完整业务流程
优点
模型简单，没有多线程、进程通讯、竞争的问题，全都在一个进程中完成。
缺点
1、性能问题，只有一个线程，无法完全发挥多核CPU的性能，Handler在处理某个连接的业务时，整个进程无法处理其他连接事件，很容易导致性能瓶颈
2、可靠性问题，线程意外终止，或者进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障
使用场景
客户端的数量有限，业务处理非常快速，比如Redis在业务处理的时间复杂度O（1）的情况

单Reactor多线程

1、Reactor对象通过select监控客户端请求事件，收到事件后，通过dispatch进行分发
2、如果建立连接请求，则由Acceptor通过accetp处理连接请求，然后创建一个Handler对象处理完成连接后的各种事件
3、如果不是连接请求，则由Reactor分发调用连接对应的handler来处理
4、handler只负责响应事件，不做具体的业务处理，通过read读取数据后，会分发给后面的worker线程池的某个线程处理业务
5、worker线程池会分配独立线程完成真正的业务，并将结果返回给handler
6、hanlder收到响应后，通过send将结果返回给client
优点
可以充分利用多核CPU的处理能力
缺点
多线程下数据共享和访问比较复杂，reactor处理所有事件的监听和响应。单线程运行，高并发场景下容易出现性能瓶颈

主从Reactor多线程

1、Reactor主线程MainReactor对象通过select监听连接事件，收到事件后，通过Acceptor处理连接事件
2、当Acceptor处理连接事件后，MainReactor将连接分配给SubReactor
3、SubReactor将连接加入到连接队列进行监听，并创建handler进行各种事件处理
4、当有事件发生时，SubReactor就会调用相应的handler处理
5、handler通过read读取数据，分发给worker线程处理
6、worker线程池分配独立的worker线程进行业务处理，并返回结果
7、handler收到响应的结果后，再通过send将结果返回给client
8、Reactor主线程可以对应多个Reactor子线程，即MainReactor可以关联多个SubReactor
优点
1、父线程与子线程的数据交互简单职责明确，父线程只需要接收新连接，子线程完成后续的业务处理
2、父线程与子线程的数据交互简单，Reactor只需要把新连接传给子线程，子线程无需返回数据
缺点
编程复杂度较高

Netty模型简单版

1、BossGroup线程维护Selector，只关注Accept
2、当接收到Accept事件，获取到对应的AocketChannel，封装成NIOSocketChannel并注册到Worker线程（事件循环），并进行维护
3、当Worker线程监听到selector中通道发生自己感兴趣的事件后，就进行处理(hanlder),此时handler已经加入到通道

进阶版

1、有两组线程池：BossGroup和WorkerGroup, BossGroup中的线程（可以有多个），专门负责客户端连接，WorkerGroup中的线程专门负责连接上的读写。
2、BossGroup和WorkerGroup含有多个不断循环的执行事件处理的线程，每个线程都包含一个Selector, 用于监听注册在其上的Channel
3、每个BossGroup中的线程循环执行以下三个步骤：
3.1 轮询注册在其上的ServerSocketChannel的accept事件（OP_ACCEPT事件）
3.2 处理accept事件，与客户端建立连接，生成一个NioSocketChannel，并将其注册到WorkerGroup中某一个线程上的Selector上
3.3 再去以此循环处理任务队列中的下一个事件
4、每个WokerGroup中的线程循环执行以下三个步骤：
4.1 轮询注册在其上的NIOSocketChannel 的read/write事件（OP_READ/OP_WRITE事件）
4.2 在对应的NIOSocketChannel上处理read/write事件
4.3 再去以此循环处理任务队列中的下一个事件

详细版

1、Netty抽象出两组线程池BossGroup专门负责接收客户端的连接，WorkerGroup专门负责网络的读写
2、BossGroup和WorkerGroup类型都是NioEventLoopGroup
3、NioEventLoopGroup相当于一个事件循环组，这个组中含有多个事件循环，每一个事件循环是NioEventLoop
4、NioEventLoop表示一个不断循环的执行处理任务的线程，每个NioEventLoop都有一个Selector，用于监听绑定在其上的socket的网络通讯
5、NioEventLoopGroup可以有多个线程，即可以含有多个NioEventLoop
6、每个BossNioEventLoop循环执行的步骤有3个：
6.1 轮询accept事件
6.2 处理accept事件，与client建立连接，生成NioSocketChannel，并将其注册到某个WorkerNioEventLoop上的Selector
6.3 处理任务队列的任务，即runAllTasks
7、每个WorkerNioEventLoop循环执行的步骤
7.1 轮询read,write事件
7.2 处理i/o事件，即read,write事件，在对应NioSocketChannel处理
7.3处理任务队列的任务，即runAllTasks
8、每个WorkerNioEventLoop处理业务时，会使用pipeLine(管道)，pipeline中包含了channel，即通过pipeline可以获取到对应的通道，管道中维护了很多的处理器

入门实例

要求
1、Netty在 6668端口监听，客户端能发送消息给服务端 “hello, server”
2、服务器可以回复消息给客户端 “hello, client”

服务端

package com.tk.netty.netty;

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;


public class NettyServer {

    public static void main(String[] args) {
        //创建BossGroup和WorkerGroup
        //1、创建两个线程组BossGroup和WorkerGroup
        //2、bossGroup知识2处理连接请求，真正的和客户端处理业务，会交给workerGroup完成
        //3、两个都是无线循环
        //4、bossGroup和workerGroup含有的子进程（NioEventLoop）的个数
        //默认实际cpu核数 *2
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用链式编程进行设置
            bootstrap.group(bossGroup, workerGroup)//设置两个线程组
                    .channel(NioServerSocketChannel.class)//使用NioSocketChannel作为服务器的通道
                    .option(ChannelOption.SO_BACKLOG, 128)//设置线程队列得到连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态
                    .childHandler(new ChannelInitializer() {//设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("server is ready。。。");
            //绑定一个端口并且同步，生成了一个ChannelFuture对象
            //启动服务器（并绑定端口）
            ChannelFuture cf = bootstrap.bind(6668).sync();
            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端处理器

package com.tk.netty.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelPipeline;
import io.netty.util.CharsetUtil;


public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        System.out.println("服务器读取线程：" + Thread.currentThread().getName());
        System.out.println("server ctx=" + ctx);
        System.out.println("channel 和 pipeline的关系");
        ChannelPipeline pipeline = ctx.pipeline();//本质是一个双向链表
        Channel channel = ctx.channel();

        //将msg转成一个ByteBuf
        //ByteBuf是Netty提供的,不是Nio的ByteBuf
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("客户端发送消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址：" + channel.remoteAddress());
    }

    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        //相当于 write+flush
        //将数据写入缓存，并刷新
        //对这个发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,client", CharsetUtil.UTF_8));
    }

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

客户端

package com.tk.netty.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;


public class NettyClient {

    public static void main(String[] args) {
        //客户端需要一个事件循环组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建客户端启动对象
            //注意客户端使用的不是ServerBootStrap而是BootStrap
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)//设置线程组
                    .channel(NioSocketChannel.class)//设置客户端通道的实现类（反射）
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) {
                            socketChannel.pipeline().addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("client is ready。。。");
            //启动客户端去连接服务器
            //关于ChannelFuture要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端处理器

package com.tk.netty.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("client:" + ctx);
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server...", CharsetUtil.UTF_8));
    }

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务端回复的消息：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址：" + ctx.channel().remoteAddress());
    }

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

运行结果
服务端启动，在6668端口监听

客户端进行连接

任务队列(Task) 使用场景

1、用户程序自定义的普通任务
2、用户自定义定时任务
3、非当前Reactor线程调用Channel的各种方法。例如在推送系统的业务线程里面，根据用户的标识，找到对应的Channel引用，然后调用Write类方法向该用户推送消息，就会进入到这种场景。最终的Write会提交到任务队列中后被异步消费。

代码演示

package com.tk.netty.netty;

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

import java.util.concurrent.TimeUnit;


public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        //比如这里我们有一个耗时的操作-> 异步执行-> 提交该channel对应的NioEventLoop的taskQueue中
        //解决方案1 用户程序自定义的普通任务
        ctx.channel().eventLoop().execute(() -> {
            try {
                Thread.sleep(5000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("hello client2", CharsetUtil.UTF_8));
                System.out.println("channel hashcode=" + ctx.channel().hashCode());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        ctx.channel().eventLoop().execute(() -> {
            try {
                Thread.sleep(5000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("hello client3", CharsetUtil.UTF_8));
                System.out.println("channel hashcode=" + ctx.channel().hashCode());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });

        //解决方案2：用户自定义定时任务->该任务是提交到scheduledTaskQueue中
        ctx.channel().eventLoop().schedule(() -> {
            try {
                Thread.sleep(5000);
                ctx.writeAndFlush(Unpooled.copiedBuffer("hello client4", CharsetUtil.UTF_8));
                System.out.println("channel hashcode=" + ctx.channel().hashCode());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, 5, TimeUnit.SECONDS);

        System.out.println("go on 。。。");

    }


    
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        //相当于 write+flush
        //将数据写入缓存，并刷新
        //对这个发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,client", CharsetUtil.UTF_8));
    }

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

运行结果
服务端

客户端

说明
1、Netty抽象出两组线程池，BossGroup专门负责接收客户端连接，WorkerGroup专门负责网络读写操作
2、NioEventLoop标识一个不断循环执行任务的线程组，每个NioEventLoop都有一个Selector，用于监听绑定在其上的socket网络通道
3、NioEventLoop内部采用串行化设计，从消息的读取->解码->处理->编码->发送，始终由IO线程NioEventLoop负责

NioEventLoopGroup下包含多个NioEventLoop
每个NioEventLoop中包含有一个Selector，一个taskWueue
NioEventLoop的Selector上可以注册监听多个NioChannel
每个NioChannel只会绑定在唯一的NioEventLoop上
每个NioChannel都绑定有一个自己的ChannelPipeline

异步模型基本介绍

1、异步的概念和同步相对，当一个异步过程调用发出后，调用者不能立刻得到结果，实际处理这个调用的组件在完成之后，通过状态，通知，和回调来通知调用者
2、Netty中的I/O操作是异步的，包括Bind、Write、Connetct等操作会简单的返回液体个ChannelFuture
3、调用者并不能立刻获得结果，而是通过Future-Listener机制，用户可以方便的主动获取或者通过通知机制获得IO操作结果
4、Netty的异步模型是建立在future和callback上的。callback就是回调，重点说Future，它的核心思想是：假设一个方法fun,计算过程可能非常耗时，等待fun返回显然不合适，那么可以在调用fun的时候，立马返回一个Future，后续可以通过Future去监控方法fun的处理过程（即Future-Listener机制）

Future说明

1、表示异步的执行结果，可以通过它提供的方法来检测执行是否完成，比如检索计算等等
2、ChannelFuture是一个接口：public interface ChnnaelFuture extends Future 我们可以添加监听器，当监听的事件发生时，就会通知到监听器

Future-Listener机制

1、当Future对象刚刚创建时，处于非完成状态，调用者可以通过返回的ChannelFuture来获取操作执行的状态，注册监听函数来执行完成后的操作
2、常见有如下的操作

通过isDone方法来判断当前操作是否完成；
通过isSuccess方法来判断已完成的当前操作是否成功
通过getCause方法来获取已完成的当前操作失败的原因
通过isCanceled方法判断已完成的当前操作是否被取消
通过addListener方法来注册监听器，当操作已完成（isDone方法返回完成），将会通知指定的监听器；如果Future对象已完成，则通知指定的监听器

举例说明

绑定端口是异步操作，当绑定操作处理完，将会调用相应的监听器处理逻辑

 ChannelFuture cf = bootstrap.bind(6668).sync();
            cf.addListener((ChannelFutureListener) channelFuture -> {
                if(cf.isSuccess()){
                    System.out.println("监听端口 6668 成功！");
                }else{
                    System.out.println("监听端口 6668 失败!");
                }
            });

实例-HTTP服务

1、Netty服务器在6668监听，浏览器发出请求 http://localhost:6668
2、服务器可以回复给客户端，“Hello，我是服务端”，并对特定请求资源进行过滤
目的：Netty可以做Http服务开发，并且理解Hanlder实例和客户端及请求的关系

服务端

package com.tk.netty.netty;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;


public class TestServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new TestServerInitializer());

            ChannelFuture channelFuture = b.bind(8081).sync();
            channelFuture.addListener((ChannelFutureListener) -> {
                if (channelFuture.isSuccess()) {
                    System.out.println("监听端口 8081 成功！");
                } else {
                    System.out.println("监听端口 8081 失败!");
                }
            });
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}

初始化处理器

package com.tk.netty.netty;

import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.http.HttpServerCodec;


public class TestServerInitializer extends ChannelInitializer {

    @Override
    protected void initChannel(SocketChannel ch) {
        //向管道加入处理器
        //得到管道
        ChannelPipeline pipeline = ch.pipeline();
        //加入一个netty提供的httpServerCodec codec=>[coder-decoder]
        //1、HttpServerCodec 是netty提供的处理http的 编-解码器
        pipeline.addLast("MyHttpServerCodec", new HttpServerCodec());
        //2、增加一个自定的以的handler
        pipeline.addLast("myTestHttpServerHandler", new TestHttpServerHandler());
    }
}

自定义处理器

package com.tk.netty.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;

import java.net.URI;


public class TestHttpServerHandler extends SimpleChannelInboundHandler {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
        //判断msg是不是httpRequest请求
        if (msg instanceof HttpRequest) {
            System.out.println("pipeline hashcode=" + ctx.pipeline().hashCode());
            System.out.println("TestHttpServerHandler hashcode=" + this.hashCode());
            System.out.println("msg 类型=" + msg.getClass());
            System.out.println("客户端地址=" + ctx.channel().remoteAddress());

            HttpRequest httpRequest = (HttpRequest) msg;
            //获取uri，过滤指定的资源
            URI uri = new URI(httpRequest.uri());
            if ("/favicon.ico".equals(uri.getPath())) {
                System.out.println("请求了 favicon.ico，不做响应");
                return;
            }
            //回复信息给浏览器（http协议）
            ByteBuf content = Unpooled.copiedBuffer("hello,我是服务器", CharsetUtil.UTF_8);
            //构造一个http的响应，即httpResponse
            DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
            response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
            response.headers().set(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());

            //将构建好response 返回
            ctx.writeAndFlush(response);
        }
    }
}

Netty核心组件 Bootstrap、ServerBootStrap

BootStrap意思是引导，一个Netty应用通常由一个BootStrap开始，主要作用是配置整个Netty程序，串联各个组件，Netty中Bootstrap类是客户端程序的启动引导类，ServerBootstrap是服务端启动引导类
常用方法

public ServerBootstrap group(EventLoopGroup parentGroup,EventLoopGroup childGroup)//该方法用于服务器端，用来设置两个EventLoop
public B group(EventLoopGroup)//该方法用于客户端，用来设置一个EventLoop
public B channel(Class channelClass)//该方法用来设置一个服务器端的通道实现
public  B option(ChannelOption option,T value)//用来给ServerChannel添加配置
public  ServerBootstrap childOption(ChannelOption childOption,T value)//用来给接收到的通道添加配置
public ServerBootstrap childHandler(ChannelHandler childHandler)//该方法用来设置业务处理类（自定义的handler）
public ChannelFuture bind(int inetPort)//该方法用于服务端，用来设置占用的端口号
public ChannelFuture connect(String inetHost,int inetPort)//该方法用于客户端用来连接服务器端

Future、ChannelFuture

Netty中所有的IO操作都是异步的，不能立刻得知消息是否被正确处理，但是可以过一会儿等他执行完成或者直接注册一个监听，具体的实现就是通过Future和ChannelFuture，他们可以注册一个监听，当操作执行成功或者失败会自动触发注册的监听事件

常用方法

Cahnnel channel()//返回当前正在进行IO操作的通道
ChannelFuture sync() //等待异步操作执行完毕

Channel

1、Netty网络通信的组件，能够用于执行网络IO操作
2、通过Channel可获得当前网络连接的通道状态
3、通过Channel可获得网络连接的配置参数（例如接收缓冲区的大小）
4、Channel提供异步的网络I/O操作（如建立连接，读写，绑定端口），异步调用意味着任何I/O调用都将立即返回，并且不保证在调用结束时锁清秋的I/O操作已完成
5、调用立即返回一个ChannelFuture实例，通过注册监听器到ChannelFuture上，可以I/O操作成功、失败或取消时回调通知调用方
6、支持关联I/O操作与对应的处理程序
7、不同协议，不同的阻塞类型的连接都有不同的Channel类型与之对应，常用的Channel类型
NioSocketChannel：异步的客户端TCP Socket连接
NioServerSocketChannel：异步的服务器端TCP Socket连接
NioDatagramChannel：异步的UDP连接
NioSctpChannel：异步的Sctp连接
NioSctpServerChannel：异步的Sctp服务端连接，这些通道涵盖了UDP和TCP网路IO以及文件

Selector

1、Netty基于Selector对象实现I/O多路复用，通过Selector一个线程可以监听多个连接的Channel事件
2、当向一个Selector中注册Channel后，Selector内部的机制就可以自动不断地查询这些注册的Channel是否有已就绪的I/O事件（例如可读，可写，网络连接等），这样程序就可以很简单的使用一个线程高效的管理多个Channel

ChannnelHanlder

1、ChannelHandler是一个接口，处理I/O事件或拦截I/O操作，并将其转发到ChannelPipeline(业务处理链)中的下一个处理程序
2、ChannelHandler本身没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类
ChannelHandler子类及其实现类

常用方法

package com.tk.netty.netty;

import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandler;

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
    public ChannelInboundHandlerAdapter() {
    }
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();
    }
    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
    }
    //通道就绪事件
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }
    //通道读取数据事件
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
    }
    //数据读取完毕事件
    public void channelReadComplete(ChannelHandlerContext channelHandlerContext) throws Exception {
    }
}

Pipeline和ChannelPipeline

1、ChannelPipeline是一个Handler的集合，它负责处理和拦截inbound或者outbound的事件和操作，相当于贯穿Netty的链。（也可以这样理解：ChannelPipeline是保存ChannelHandler的List，用于处理或拦截Channel的入站事件和出战操作）
2、ChannelPipeline实现了一种高级形式的拦截过滤模式，使用户可以完全控制事件的处理方式，以及Channel中各个的ChannelHandler如何交互
3、在Netty中每个Channel都有且仅有一个ChannelPipeline与之对应，它们的组成关系如下：

说明：
1、一个Channel包含了一个ChannelPipeline,而ChannelPipeline中维护了一个由ChannelHnadlerContext组成的双向链表，并且每个ChannelHandlerContext中又关联着一个ChannelHandler
2、入站事件和出站事件在一个双向链表中，入站事件和会从链表head往后传递到最后一个入站的handler,出站事件会从链表tail往前传递到最前一个出站的handler，两种类型的handler互不干扰
常用方法

ChannelPipeline addFirst(ChannelHandler... handlers)//把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addLast(ChannelHandler... handlers)//把一个业务处理类（handler）添加到链中的最后一个位置

ChannelHandlerContext

1、保存Channel相关的所有上下文信息，同时关联一个ChannelHandler对象
2、即ChannelHandlerContext中包含一个具体的事件处理器ChannelHandler,同时ChannelHandlerContext中也绑定了对应的pipeline和Channel的信息，方便对ChannelHandler进行调用

常用方法

ChannelFuture close()//关闭通道
ChannelOutboundInvoker flush()//刷新
ChannelFuture writeAndFlush(Object msg)//将数据写到ChannelPipeline中当前ChannelHandler的下一个ChannelHandler开始处理（出站）

ChannelOption

1、Netty在创建Channel实例后，一般都需要设置ChannelOption参数
2、ChannelOption参数如下：
ChannelOption.SO_BACKLOG
对应TCP/IP协议 listen函数中的backlog参数，用来初始化服务器可连接队列大小。服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小。
ChannelOption.SO_KEEPALIVE
一直保持连接活动状态

EventLoopGroup和其实现类NioEventLoopGroup

1、EventLoopGroup是一组EventLoop的抽象，Netty为了更好的利用多核CPU资源，一般会有多个EventLoop同时工作，每个EventLoop维护着一个Selector实例
2、EventLoopGroup提供next接口，可以从组里面按照一定规则获取一个EventLoop来处理任务，在Netty服务器端编程中，我们一般需要两个EventLoopGroup，例如：BossEventLoopGroup和WorkerEventLoopGroup
3、通常一个服务端即一个ServerSocketChannel对应一个Selector和一个EventLoop线程。BossEventLoop负责接收客户端的连接并将SocketChannel交给WorkerEventLoopGroup来进行IO处理

Unpooled

Netty提供一个专门用来操作缓冲区（即Netty的数据容器）的工具类
常用方法

//通过给定的数据和字符编码返回一个ByteBuf对象（类似于Nio中的ByteBuffer但由区别）
public static ByteBuf copiedBuffer(CharSequence String,Charset charset)

案例1

package com.tk.netty.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;


public class NettyByteBuf01 {

    public static void main(String[] args) throws Exception {
        //创建一个ByteBuf
        //1、创建对象，该对象包含一个数组att，是一个byte[10]
        //2、在netty的buffer中，不需要使用flip进行反转
        //底层维护了readerIndex和writeIndex
        //3、通过readerIndex和WriteIndex和capacity，将buffer分成三个区域
        //0->readerIndex 已经读取的区域
        //readerIndex->writerIndex 可读的区域
        //writerIndex->capacity 可写的区域
        ByteBuf buffer = Unpooled.buffer(10);
        for (int i = 0; i < 10; i++) {
            buffer.writeByte(i);
        }
        System.out.println("capacity=" + buffer.capacity());
        for (int i = 0; i < buffer.capacity(); i++) {
            System.out.println(buffer.getByte(i));
        }
//        for (int i = 0; i < buffer.capacity(); i++) {
//            System.out.println(buffer.readByte());
//        }
    }
}

案例2

package com.tk.netty.netty;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.nio.charset.StandardCharsets;


public class NettyByteBuf02 {

    public static void main(String[] args) {
        ByteBuf byteBuf = Unpooled.copiedBuffer("hello,world!", StandardCharsets.UTF_8);
        //相关的使用方法
        if (byteBuf.hasArray()) {
            byte[] content = byteBuf.array();
            //将content转成字符串
            System.out.println(new String(content, StandardCharsets.UTF_8));
            System.out.println("byteBuf=" + byteBuf);
            System.out.println("offset=" + byteBuf.arrayOffset());
            System.out.println("readerIndex=" + byteBuf.readerIndex());
            System.out.println("writerIndex=" + byteBuf.writerIndex());
            System.out.println("capacity=" + byteBuf.capacity());
            System.out.println("getByte(0)=" + byteBuf.getByte(0));
            //可读的字节数
            int len = byteBuf.readableBytes();
            System.out.println("len=" + len);
            //使用for取出各个字节
            for (int i = 0; i < len; i++) {
                System.out.println((char) byteBuf.getByte(i));
            }
            //按照某个范围读取
            System.out.println(byteBuf.getCharSequence(0, 4, StandardCharsets.UTF_8));
            System.out.println(byteBuf.getCharSequence(4, 6, StandardCharsets.UTF_8));
        }
    }
}

Netty应用实例-群聊系统

实例要求：
1、编写一个Netty群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）
2、实现多人群聊
3、服务器端：可以检测用户上线，离线，并实现消息转发功能
4、客户端：通过channel可以无阻塞发送消息给其他所有用户，同时可以接受其他用户发送的消息（由服务器转发得到）
5、目的：进一步理解Netty非阻塞网络编程机制

服务端

package com.tk.netty.netty.groupchat;

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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;


public class GroupChatServer {


    public static void main(String[] args) throws Exception {
        new GroupChatServer(7000).run();
    }

    
    private final int port;

    public GroupChatServer(int port) {
        this.port = port;
    }

    public void run() throws Exception {
        //创建两个线程组
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            //获取到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //向pipeline加入解码器
                            pipeline.addLast("decoder", new StringDecoder());
                            //向pipeline加入编码器
                            pipeline.addLast("encoder", new StringEncoder());
                            //加入自己的业务处理
                            pipeline.addLast(new GroupChatServerHandler());
                        }
                    });
            System.out.println("netty 服务启动了。。。");
            ChannelFuture channelFuture = b.bind(port).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端处理器

package com.tk.netty.netty.groupchat;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;

import java.text.SimpleDateFormat;
import java.util.Date;


public class GroupChatServerHandler extends SimpleChannelInboundHandler {

    //定义一个channel组，管理所有的channel
    //GlobalEventExecutor.INSTANCE 是全局的事件执行器，是一个单例
    private static final ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        //将客户加入聊天的信息推给其他在线的客户端
        //该方法会将channelGroup中所有的channel遍历，并发送消息，我们自己不需要遍历
        channelGroup.writeAndFlush("[客户端] " + channel.remoteAddress() + " 加入聊天 " + sdf.format(new Date()) + "n");
        channelGroup.add(channel);
    }

    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        Channel channel = ctx.channel();
        channelGroup.writeAndFlush("[客户端] " + channel.remoteAddress() + " 离开了");
        System.out.println("channelGroup size=" + channelGroup.size());
    }

    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println(ctx.channel().remoteAddress() + " 上线了~");
    }

    
    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        System.out.println(ctx.channel().remoteAddress() + " 离线了~");
    }

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

    
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) {
        Channel channel = ctx.channel();
        //这时我们遍历channelGroup，根据不同情况，回送不同的消息
        channelGroup.forEach(ch -> {
            if (ch != channel) {//不是当前的通道
                ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息:" + msg + "n");
            } else {
                ch.writeAndFlush("[自己]发送了消息：" + msg + "n");
            }
        });
    }
}

客户端

package com.tk.netty.netty.groupchat;

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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;

import java.util.Scanner;


public class GroupChatClient {

    private final String host;
    private final int port;

    public static void main(String[] args) throws Exception {
        new GroupChatClient("127.0.0.1", 7000).run();
    }

    public GroupChatClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public void run() throws Exception {
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            //得到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //加入相关handler
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            //加入自定义handler
                            pipeline.addLast(new GroupChatClientHandler());
                        }
                    });
            ChannelFuture channelFuture = b.connect(host, port).sync();
            //得到channel
            Channel channel = channelFuture.channel();
            System.out.println("---" + channel.localAddress() + "---");
            //客户端需要输入信息
            Scanner sc = new Scanner(System.in);
            while (sc.hasNextLine()) {
                String msg = sc.nextLine();
                //通过channel发送到服务器端
                channel.writeAndFlush(msg + "rn");
            }

        } finally {
            group.shutdownGracefully();
        }
    }

}

客户端处理器

package com.tk.netty.netty.groupchat;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;


public class GroupChatClientHandler extends SimpleChannelInboundHandler {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) {
        System.out.println(s.trim());
    }
}

运行结果

服务端启动

启动客户端1

启动客户端2

客户端1发送消息

客户端1离线了

Netty心跳检测机制-案例实例要求

1、编写一个Netty心跳检测机制案例，当服务器超过3s没有读时，就提示读空闲
2、当服务器超过5s没有写操作时，就提示写空闲
3、实现服务器超过7s没有读或者写时，就提示读写空闲

服务器端

package com.tk.netty.netty.readwriteidle;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.concurrent.TimeUnit;


public class TestReadWriteIdle {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            //加入一个netty提供IdleStateHandler
                            //1、IdleStateHandler是netty提供的处理空闲状态的处理器
                            //2、long readerIdleTime:表示多长时间没有读，就会发送一个心跳检测包检测是否连接
                            //3、long writerIdleTime: 表示多长时间没有写，就会发送一个心跳检测包检测是否连接
                            //4、long allIdleTime: 表示多长时间没有续写，就会发送一个心跳检测包检测是否有连接
                            //5、当IdleStateEvent出发后，就会传递给管道的下一个handler去处理
                            //通过调用（触发）下一个handler的userEventTriggered，在该方法中去处理IdleStateEvent(读空闲，写空闲，读写空闲)
                            pipeline.addLast(new IdleStateHandler(3, 5, 7, TimeUnit.SECONDS));
                            //加入一个对空闲检测进一步处理的handler(自定义)
                            pipeline.addLast(new MyServerHandler());
                        }
                    });
            ChannelFuture channelFuture = b.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

处理器

package com.tk.netty.netty.readwriteidle;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleStateEvent;


public class MyServerHandler extends ChannelInboundHandlerAdapter {

    
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
        if (evt instanceof IdleStateEvent) {
            //将evt向下转型
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + "--超时事件--" + eventType);
            System.out.println("服务器做相应处理");
            //如果发生空闲，关闭通道
            //ctx.channel().close():
        }
    }
}

Netty 通过WebSocket 实现长连接实例要求

1、Http协议是无状态的，浏览器和服务器间的请求相应一次，下一次会重新创建连接
2、实现基于WebSocket的长连接的全双工的交互
3、改变Http协议多次请求的约束，实现长连接了，服务器可以发送消息给浏览器
4、客户端浏览器和服务器会相互感应，比如服务器关闭了，浏览器会感知，同样浏览器关闭了，服务器会感知

服务器端

package com.tk.netty.websocket;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.stream.ChunkedWriteHandler;


public class WebSocketServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            //因为基于http协议，使用的http的编码和解码器
                            pipeline.addLast(new HttpServerCodec());
                            //是以块的方式写，添加ChunkedWriteHandler处理器
                            pipeline.addLast(new ChunkedWriteHandler());
                            //http 数据在传输过程中是分段，HttpObjectAggregator，就是可以将多个段聚合
                            //这就是为什么当浏览器发送大量数据时，就会发出多次Http请求
                            pipeline.addLast(new HttpObjectAggregator(8192));
                            //1、对应websocket,它的数据是以帧(frame)形式传递
                            //2、可以看到WebSocketFrame下面又六个子类
                            //3、浏览器请求时，ws://localhost:7000/hello表示请求的uri
                            //4、WebSocketProtocolHandler核心功能是将http协议升级为ws协议，保持长连接
                            //5、保持长连接是通过一个状态码 101
                            pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
                            //自定义协议
                            pipeline.addLast(new MyTextWebSocketFrameHandler());
                        }
                    });
            ChannelFuture channelFuture = b.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }
}

自定义处理器

package com.tk.netty.websocket;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import java.time.LocalDateTime;


public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler {


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) {
        System.out.println("服务端收到消息：" + msg.text());
        //回复消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间：" + LocalDateTime.now() + "t" + msg.text()));
    }

    
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        //id表示唯一的值，LongText是唯一的ShortText不是唯一
        System.out.println("handlerAdded 被调用，longId=" + ctx.channel().id().asLongText());
        System.out.println("handlerAdded 被调用，shortId=" + ctx.channel().id().asShortText());
    }

    
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        System.out.println("handlerRemove被调用，longId=" + ctx.channel().id().asLongText());
    }

    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        System.out.println("异常发生：" + cause.getMessage());
        ctx.close();
    }
}

浏览器




    
    Title

运行结果

1、服务器端启动，浏览器显示连接成功

2、浏览器发送消息，并收到服务端响应

3、服务器端关闭连接

4、客户端关闭连接

Protobuf

protocol buffers 是一种语言无关、平台无关、可扩展的序列化结构数据的方法，它可用于（数据）通信协议、数据存储等。
Protocol Buffers 是一种灵活，高效，自动化机制的结构数据序列化方法－可类比 XML，但是比 XML 更小（3 ~> 10倍）、更快（20 ~ 100倍）、更为简单。

入门实例1

要求
1、客户端可以发送一个Student POJO 对象到服务器（通过Protobuf编码）
2、服务器端能接收一个StudentPOJO对象，并显示信息（通过Protobuf编码）

.proto文件

syntax = "proto3";//版本
option java_outer_classname = "StudentPOJO";//生成的外部类名，同时也是文件名
//protobuf使用message 管理数据
message Student{//会在StudentPOJO外部类生成一个内部类Student，他是真正发送的POJO对象
  int32 id = 1;//Student 类中有一个属性名字为 id 类型为 int32 (protobuf类型)1表示属性序号，不是值
  string name = 2;
}

protobuf 可执行文件

下载地址

执行代码

.protoc-3.6.1.exe .Student.proto --java_out=.

生成的文件

核心代码

客户端加入编码器，注意编码器的顺序

处理器

服务器端加入解密器

处理器

运行结果
分别启动服务器和客户端

服务器端收到结果

入门实例2

要求
1、客户端可以随机发送Student ，Worker对象到服务器（通过Protobuf编码）
2、服务端能接收Student，Worker对象（需要判断是哪种类型），并显示信息（通过protobuf编码）

关键代码

syntax = "proto3";
option optimize_for = SPEED;//加快解析
option java_package = "com.tk.netty.protobuf.message";//指定生成到哪个包下
option java_outer_classname = "MyDataInfo";//外部类名，文件名
//protobuf可以使用message管理其他的message
message MyMessage{
  //定义一个枚举类型
  enum DataType{
    StudentType = 0;//在proto3要求enum的编号从0开始
    WorkerType = 1;
  }
  //用Data_type来标识传的是哪一个枚举类型
  DataType data_type = 1;
  //表示每次枚举类型最多的只能出现其中的一个，节省空间
  oneof dataBody{
    Student student = 2;
    Worker worker = 3;
  }

  message Student{
    int32 id = 1;
    string name = 2;
  }

  message Worker{
    string name = 1;
    int32 age = 2;
  }
}

TCP粘包和拆包基本介绍

1、TCP是面向连接的，面向流的，提供高可靠性服务，收发两端（客户端和服务端）都要有一一成对的socket,因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法（Nagle算法），将多次间隔较小且数量小的数据，合并成一个大的数据块，然后进行封包。这样虽然提高了效率，但是接收端的难度就难于分辨出完整的数据包了，因为面向流的通信是无消息保护边界的。

2、由于TCP无消息保护边界的，需要在接收端处理消息边界问题，也就是我们所说的粘包、拆包问题，如图展示TCP粘包，拆包问题

说明
假设客户端分别发送了两个数据包D1和D2给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下四种情况：
1、服务端分两次读取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包
2、服务端一次接收到了两个数据包，D1和D2粘合到了一起，称之为TCP粘包
3、服务端分两次读取到了数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2包的剩余部分,这称之为TCP拆包
4、服务端分两次读取到了数据包，第一次读取到了D1包的部分内容D1_1，第二次读取到了D1包的剩余内容D1_2和完整的D2包，这也称之为TCP拆包

代码实例（粘包、拆包问题）

服务端

package com.tk.netty.tcp;

import com.tk.netty.protobuf.message.MyDataInfo;
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.protobuf.ProtobufDecoder;


public class NettyServer {

    public static void main(String[] args) {
        //创建BossGroup和WorkerGroup
        //1、创建两个线程组BossGroup和WorkerGroup
        //2、bossGroup知识2处理连接请求，真正的和客户端处理业务，会交给workerGroup完成
        //3、两个都是无线循环
        //4、bossGroup和workerGroup含有的子进程（NioEventLoop）的个数
        //默认实际cpu核数 *2
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用链式编程进行设置
            bootstrap.group(bossGroup, workerGroup)//设置两个线程组
                    .channel(NioServerSocketChannel.class)//使用NioSocketChannel作为服务器的通道
                    .option(ChannelOption.SO_BACKLOG, 128)//设置线程队列得到连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true)//设置保持活动连接状态
                    .childHandler(new ChannelInitializer() {//设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("server is ready。。。");
            //绑定一个端口并且同步，生成了一个ChannelFuture对象
            //启动服务器（并绑定端口）
            ChannelFuture cf = bootstrap.bind(6668).sync();
            cf.addListener((ChannelFutureListener) channelFuture -> {
                if (cf.isSuccess()) {
                    System.out.println("监听端口 6668 成功！");
                } else {
                    System.out.println("监听端口 6668 失败!");
                }
            });
            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端处理器

package com.tk.netty.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.StandardCharsets;
import java.util.UUID;


public class NettyServerHandler extends SimpleChannelInboundHandler {

    private int count;

    
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        String message = new String(buffer, StandardCharsets.UTF_8);
        System.out.println("收到客户端发送的消息：" + message);
        System.out.println("接收到客户端的数据量：" + (++count) + "n");

        //服务器回送数据给客户端，发送一个随机id
        ByteBuf byteBuf = Unpooled.copiedBuffer(
                UUID.randomUUID().toString().replaceAll("-", "") + "t", StandardCharsets.UTF_8);
        ctx.writeAndFlush(byteBuf);
    }

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

客户端

package com.tk.netty.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
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.NioSocketChannel;


public class NettyClient {

    public static void main(String[] args) {
        //客户端需要一个事件循环组
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建客户端启动对象
            //注意客户端使用的不是ServerBootStrap而是BootStrap
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)//设置线程组
                    .channel(NioSocketChannel.class)//设置客户端通道的实现类（反射）
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("client is ready。。。");
            //启动客户端去连接服务器
            //关于ChannelFuture要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

客户端处理器

package com.tk.netty.tcp;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

import java.nio.charset.StandardCharsets;

public class NettyClientHandler extends SimpleChannelInboundHandler {

    private int count;

    
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        //客户端发送10条数据
        for (int i = 0; i < 10; i++) {
            ByteBuf buffer = Unpooled.copiedBuffer("hello,world!", StandardCharsets.UTF_8);
            ctx.writeAndFlush(buffer);
        }
    }

    
    @Override
    public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        byte[] buffer = new byte[msg.readableBytes()];
        msg.readBytes(buffer);
        //将buffer转成字符串
        String message = new String(buffer, StandardCharsets.UTF_8);
        System.out.println("接收到服务器的数据：" + message + "t");
        System.out.println("接收到服务器数的数据量：" + (++count));

    }

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

}

运行结果
启动服务端和客户端

启动两个客户端

解决方案

1、使用自定义协议+编码器来解决
2、关键就是要解决服务器端每次读取数据长度的问题，这个问题解决就不会出现服务器多读或者少读数据的问题，从而避免TCP粘包，拆包。
自定义协议

package com.tk.netty.tcp.protocol;


public class MessageProtocol {

    
    private int len;
    private byte[] content;

    public int getLen() {
        return len;
    }

    public void setLen(int len) {
        this.len = len;
    }

    public byte[] getContent() {
        return content;
    }

    public void setContent(byte[] content) {
        this.content = content;
    }
}

编码器

package com.tk.netty.tcp.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;


public class MyMessageEncoder extends MessageToByteEncoder {

    @Override
    protected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) {
        System.out.println("MyMessageEncoder 方法被调用。。。");
        out.writeInt(msg.getLen());
        out.writeBytes(msg.getContent());
    }
}

解码器

package com.tk.netty.tcp.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ReplayingDecoder;

import java.util.List;


public class MyMessageDecoder extends ReplayingDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List

网络编程之Netty

Java相关栏目本月热门文章