Netty(三)_核心模块组件与实战

拼搏现实的明天。 2022-11-21 11:45 347阅读 0赞

Netty核心模块组件与实战

文章目录

  • Netty核心模块组件与实战
    • Bootstrap、ServerBootstrap
    • Future、ChannelFuture
    • Channel
    • ChannelHandler
    • Pipeline 和 ChannelPipeline
    • ChannelHandlerContext
    • EventLoopGroup
    • Selector
    • Option
    • Unpooled
    • NettyBS-TCP简单通信
    • Netty-Http服务编写
    • Netty-群聊系统
    • Netty-心跳检测机制
    • WebSocket 编程实现BS长连接

回顾上一篇讲IO线程模型最终抛出的Netty线程模型示意图如下。

img

本文紧接上篇，正式介绍netty，上图是由上一章在最后抛出来的netty线程模型示意图。

这里篇文章篇幅还是挺长的，前面部分先简单介绍下netty的最少必备API，后面再有五个小程序带出netty的应用

前半部分听个响，在应用的时候自然就懂了！

在下一篇的源码分析中，我们会更清楚的明白netty这个机器是如何动起来的。

Bootstrap、ServerBootstrap

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

ServerBootstrap常见方法

  1. //指定具体的EventLoopGroup，即服务端的父子线程组BossEventGroup和WorkEventGroup
  2. public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)    
  5. //该方法为父类AbstractBootstrap所定义，用来设置一个服务器端的通道   
  6. public B channel(Class<? extends C> channelClass) 
  8. //父类定义，用来给ServerChannel添加配置
  9. public <T> B option(ChannelOption<T> option, T value) 
  11. //对在acceptor接收到的channel进行配置
  12. public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value) 
  14. //设置自定义的Hanler用来处理accept接收的channel的事件   
  15. public ServerBootstrap childHandler(ChannelHandler childHandler) 
  17. //服务端channel绑定的监听端口    
  18. public ChannelFuture bind(int inetPort)

服务端的Boostact先不介绍，但是逻辑都是类似的

Future、ChannelFuture

Netty 中所有的 IO 操作都是异步的，不能立刻得知消息是否被正确处理。

但是可以通过监听机制，一旦有结果返回，可以监听到。

具体的实现就是通过 Future 和 ChannelFutures。

他们可以注册一个监听，当操作执行成功或失败时，执行操作的线程会调用你监听者的函数来进行通知！

常见的异步方法有 Channel channel()返回当前正在进行 IO 操作的通道。

ChannelFuture sync()等待异步操作执行完毕，该方法立刻返回一个Future对象。

通过返回Future，Future代表一个“未来”。

Channel

Channel是Netty网络通信的核心组件，意为通道/频道。能够用于提供执行网络 I/O 操作（read/write）的场所。

通过 Channel对象可获得当前网络连接的通道的状态、配置参数 （如Buffer缓冲区大小）

Channel 提供异步的网络 I/O 操作(如建立连接，读写，绑定端口)

异步调用意味着任何 I/O 调用都将立即返回，所以不保证在调用结束时所请求的 I/O 操作已完成，调用立即返回一个 ChannelFuture 实例，通过注册监听器到 ChannelFuture 上，可以在I/O 操作成功、失败或取消时回调通知调用方

Channel支持关联 I/O 操作与对应的处理程序（Handler）

不同协议、不同的阻塞类型的连接都有不同的 Channel 类型与之对应，常用的 Channel 类型有：

  • NioSocketChannel，异步的客户端 TCP Socket 连接。
  • NioServerSocketChannel，异步的服务器端 TCP Socket 连接。
  • NioDatagramChannel，异步的 UDP 连接。
  • NioSctpChannel，异步的客户端 Sctp 连接。
  • NioSctpServerChannel，异步的 Sctp 服务器端连接，这些通道涵盖了 UDP 和 TCP 网络 IO 以及文件 IO。

你可以例用netty编写一个高度定制化的网络服务器！

ChannelHandler

ChannelHandler 是一个接口，用于真正处理 I/O 事件或拦截 I/O 操作，并将其转发到其 ChannelPipeline(业务处理链) 中的下一个Handler。这里涉及到我们后面要讲的pipeline。

Handler都在pipeline管道中进行维护。服务端处理IO事件要从pipeline中调用相应的handler进行处理。

ChannelHandler 本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类

ChannelHandler及其实现类

img

  • ChannelInboundHandler用于处理入站（从客户端到服务端）IO事件。
  • 反之，ChannelOutboundHandler用于处理出站IO事件。
  • ChannelInitializer用于初始化一个ChannelHandler。

我们经常需要自定义一个 Handler 类去继承 ChannelInboundHandlerAdapter，然后通过重写相应方法实现对接收的channel的业务处理，一般需要重写以下方法

  1. //Invoked when the current Channel has read a message from the peer.
  2. //当该通道需要读取事件时，触发该方法。（事件驱动）
  3. //ChannelHandlerContext为上下文对象，携带该通道的所有信息。
  4. void channelRead(ChannelHandlerContext ctx, java.lang.Object msg)
  6. //通道连接就绪时触发该方法。    
  7. void channelActive(ChannelHandlerContext ctx)
  8. The Channel of the ChannelHandlerContext is now active    
  10.     ...

Pipeline 和 ChannelPipeline

ChannelPipeline 是一个 Handler 的集合，对应图中的pipeline，它负责处理和拦截 inbound 或者 outbound 的事件和操作，相当于 一个贯穿 Netty 的链。

ChannelPipeline 实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及 Channel 中各个的 ChannelHandler 如何相互交互

在 Netty 中每个 Channel 都有且仅持有一个 ChannelPipeline 与之对应，它们的组成关系如下

在这里插入图片描述

一个Channel包含一个ChannelPipeline，ChannelPipeline中维护了一个由ChannelHandler组成的双向链表

入站事件从head到tail逐个被Handler处理，出站事件反之。

通过ChannelPipeline addFirst(ChannelHandler... handlers)，或ChannelPipeline addLast(ChannelHandler... handlers)向ChannelPipeline中添加Handler。

ChannelHandlerContext

ChannelHandlerContext用于保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象。

即在pipeline链式传递的过程中进行链式传递。

即 ChannelHandlerContext 中 包 含 一 个 具 体 的 事 件 处 理 器 ChannelHandler ， 同 时 ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息，方便对 ChannelHandler 进行调用。

在下一篇的源码分析中，你可以认为，ChannelHandlerContext就是Handler

常用方法：

  1. ChannelFuture close(); //关闭通道
  3. ChannelOutboundInvoker flush(); //刷新挂起的消息
  5. ChannelFuture writeAndFlush(Object msg); // ..

EventLoopGroup

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-vFSzf3ss-1604217943497)(C:\Users\20335\AppData\Roaming\Typora\typora-user-images\image-20201027165808059.png)]

EventLoopGroup 是一组 EventLoop 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop 同时工作，每个 EventLoop 维护着一个 Selector 实例。

通常一个服务端口即一个 ServerSocketChannel 对应一个 Selector 和一个 EventLoop 线程。

BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理。

EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop 来处理任务。

在 Netty 服 务 器 端 编 程 中 ， 我 们 一 般 都 需 要 提 供 两 个 EventLoopGroup ， 例 如 ： BossEventLoopGroup 和 WorkerEventLoopGroup。

Selector

Netty 基于 Selector 对象实现 I/O 多路复用，通过 Selector 一个线程可以监听多个 Chan绑定在其上的channnel 事件。

当向一个 Selector 中注册 Channel 后，Selector 内部的机制就可以自动不断地查询(Select) 这些注册的 Channel 是否有已就绪的 I/O 事件（例如可读，可写，网络连接完成等），这样程序就可以很简单地使用一个线程高效地管理多个 Channel

Option

Netty 在创建 Channel 实例后,一般都需要设置 ChannelOption 参数。

ChannelOption 参数如下:

ChannelOption.SO_BACKLOG:对应TCP/IP协议listen函数中的backlog参数，用于初始化服务器的可连接队列容量。

服务器处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接。多个客户端来的时候，服务器将暂时无法处理的客户端连接放在队列中等待处理，backlog参数用于指定大小。

ChannelOption.SO_KEEPALIVE：设置一直保持连接活动状态。

Unpooled

Unpooled是Netty 提供一个专门用来操作缓冲区(即 Netty 的数据容器)的工具类

一般用来创建Netty中的缓存区，相当于NIO中的Buffer… 在Netty中为ByteBuf

Creates a new ByteBuf by allocating new space…

最常用的方法为

  1. /**
  2.      * Creates a new big-endian buffer whose content is the specified
  3.      * string encoded in the specified charset.
  4.      */
  5.     public static ByteBuf copiedBuffer(CharSequence string, Charset charset) {
  7.     }

下面再上四个实例

NettyBS-TCP简单通信

创建 Netty 项目

  • Netty 服务器在 6668 端口监听，客户端能发送消息给服务器 “hello, 服务器~”
  • 服务器可以回复消息给客户端 “hello, 客户端~”
  • 目的：对 Netty 线程模型有一个初步认识，便于理解 Netty 模型理论

配置netty的Maven依赖：

  1. <dependency>
  2.             <groupId>io.netty</groupId>
  3.             <artifactId>netty-all</artifactId>
  4.             <version>4.1.42.Final</version>
  5.         </dependency>

服务端编写：

  1. package netty.simple;
  3. import io.netty.bootstrap.ServerBootstrap;
  4. import io.netty.channel.*;
  5. import io.netty.channel.nio.NioEventLoopGroup;
  6. import io.netty.channel.socket.SocketChannel;
  7. import io.netty.channel.socket.nio.NioServerSocketChannel;
  8. import io.netty.channel.socket.nio.NioSocketChannel;
  10. public class NettyServer {
  12.     public static void main(String[] args) throws Exception {
  16.         /*
  17.         * 创建BossGroup 和 WorkerGroup
  18.         * 说明
  19.         * 1. 创建两个线程组 bossGroup 和 workerGroup
  20.         * 2. bossGroup 只是处理连接请求 , 真正的对客户端业务处理，会交给 workerGroup完成
  21.         * bossGroup 和 workerGroup 含有的子线程(NioEventLoop)的个数默认是 cpu核数 * 2
  22.         * */
  24.         //只准备一个接收连接的线程组即可
  25.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  26.         EventLoopGroup workerGroup = new NioEventLoopGroup(); //8
  28.         try {
  30.             //创建服务器端的启动对象，配置参数
  31.             ServerBootstrap bootstrap = new ServerBootstrap();
  33.             //使用链式编程来对线程组进行设置
  34.             bootstrap.group(bossGroup, workerGroup) //设置两个线程组
  35.                     .channel(NioServerSocketChannel.class) //使用netty的NioSocketChannel作为服务器的通道
  36.                     .option(ChannelOption.SO_BACKLOG, 128) // 设置服务器通道连接队列容量
  37.                     .childOption(ChannelOption.SO_KEEPALIVE, true) //通信通道保持活动连接状态
  38.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  40.                         //向SocketChannel添加Handler
  41.                         //以下方法在channel实例化后会被调用
  42.                         //目的是为了加入我们自定义的Handler 
  43.                         @Override
  44.                         protected void initChannel(SocketChannel ch) throws Exception {
  46.                             //通过channel向管道添加自定义handler
  47.                             ch.pipeline().addLast(new NettyServerHandler());
  48.                         }
  49.                     });
  51.             //异步绑定一个端口, 生成了一个 ChannelFuture 对象
  52.             //启动服务器(并绑定端口)
  53.             ChannelFuture cf = bootstrap.bind(6668).sync();
  55.             //给cf注册监听器，监控我们关心的事件
  56.             cf.addListener(new ChannelFutureListener() {
  58.                 @Override
  59.                 //以下方法在bind操作完成时会被操作线程进行回调
  60.                 public void operationComplete(ChannelFuture future) throws Exception {
  62.                     if (cf.isSuccess()) {
  64.                         System.out.println("监听端口 6668 成功...");
  65.                     } else {
  67.                         System.out.println("监听端口 6668 失败...");
  68.                     }
  69.                 }
  70.             });
  72.             //对关闭通道进行监听
  73.             cf.channel().closeFuture().sync();
  74.         }finally {
  76.             //优雅得关闭
  77.             bossGroup.shutdownGracefully();
  78.             workerGroup.shutdownGracefully();
  79.         }
  80.     }
  81. }

客户端编写：

  1. package netty.simple;
  3. import io.netty.bootstrap.Bootstrap;
  4. import io.netty.channel.ChannelFuture;
  5. import io.netty.channel.ChannelInitializer;
  6. import io.netty.channel.EventLoopGroup;
  7. import io.netty.channel.nio.NioEventLoopGroup;
  8. import io.netty.channel.socket.SocketChannel;
  9. import io.netty.channel.socket.nio.NioSocketChannel;
  11. public class NettyClient {
  13.     public static void main(String[] args) throws Exception {
  16.         //客户端只需要一个事件循环组
  17.         EventLoopGroup group = new NioEventLoopGroup();
  18.         try {
  20.             //创建客户端启动对象
  21.             //注意客户端使用的不是 ServerBootstrap 而是 Bootstrap
  22.             Bootstrap bootstrap = new Bootstrap();
  23.             //设置相关参数
  24.             bootstrap.group(group) //设置线程组
  25.                     .channel(NioSocketChannel.class) // 设置客户端通道的实现类(反射)
  26.                     .handler(new ChannelInitializer<SocketChannel>() {
  28.                         @Override
  29.                         //向客户端channel添加自定义handler
  30.                         protected void initChannel(SocketChannel ch) throws Exception {
  32.                             ch.pipeline().addLast(new NettyClientHandler()); 
  33.                         }
  34.                     });
  36.             System.out.println("客户端 ok..");
  37.             //启动客户端去连接服务器端
  38.             //注意这也是个异步操作
  39.             ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
  40.             channelFuture.channel().closeFuture().sync();
  41.         }finally {
  43.             group.shutdownGracefully();
  44.         }
  45.     }
  46. }

自定义服务端Handler：

  1. package netty.simple;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.Channel;
  6. import io.netty.channel.ChannelHandlerContext;
  7. import io.netty.channel.ChannelInboundHandlerAdapter;
  8. import io.netty.channel.ChannelPipeline;
  9. import io.netty.util.CharsetUtil;
  11. import java.util.concurrent.TimeUnit;
  12. /**
  13.  * 服务端需要继承ChannelInboundHandlerAdapter
  14.  * */
  15. public class NettyServerHandler extends ChannelInboundHandlerAdapter {
  19.     /** 
  20.      * 当服务端通道有Read事件发生时，即客户端向服务端通道写的时候，服务端的读自动触发。
  21.      * 读取客户端发送的数据
  22.      * 1. ChannelHandlerContext ctx:上下文对象, 含有管道pipeline , 通道channel, 地址等信息
  23.      * 2. Object msg: 就是客户端发送的数据 默认Object
  24.      */
  25.     @Override
  26.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  29.         //从上下文对象中获取通道
  30.         Channel channel = ctx.channel();
  31.         //将 msg 转成一个 ByteBuf
  32.         //ByteBuf 是 Netty 提供的，不是 NIO 的 ByteBuffer.
  33.         ByteBuf buf = (ByteBuf) msg;
  34.         System.out.println("客户端发送消息是:" + buf.toString(CharsetUtil.UTF_8));
  35.         System.out.println("客户端地址:" + channel.remoteAddress());
  36.     }
  38.     //数据读取完毕触发
  39.     //这里我们将数据写入缓存
  40.     @Override
  41.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  44.         //writeAndFlush 是 write + flush
  45.         //将数据写入到缓存，并刷新
  46.         //一般讲，我们对这个发送的数据进行编码
  47.         ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端~(>^ω^<)喵1", CharsetUtil.UTF_8));
  48.     }
  50.     //处理异常, 一般是需要关闭通道
  51.     @Override
  52.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  54.         ctx.close();
  55.     }
  56. }

客户自定义端Handler代码：

  1. package netty.simple;
  3. import io.netty.buffer.ByteBuf;
  4. import io.netty.buffer.Unpooled;
  5. import io.netty.channel.ChannelHandlerContext;
  6. import io.netty.channel.ChannelInboundHandlerAdapter;
  7. import io.netty.util.CharsetUtil;
  9. public class NettyClientHandler extends ChannelInboundHandlerAdapter {
  12.     //当通道就绪就会触发该方法
  13.     @Override
  14.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  16.         System.out.println("client " + ctx);
  17.         //向客户端写数据
  18.         ctx.writeAndFlush(Unpooled.copiedBuffer("hello, server: (>^ω^<)喵", CharsetUtil.UTF_8));
  19.     }
  21.     //当通道有读取事件时，即客户端向该客户端发送消息时，会触发
  22.     @Override
  23.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  25.         ByteBuf buf = (ByteBuf) msg;
  26.         System.out.println("服务器回复的消息:" + buf.toString(CharsetUtil.UTF_8));
  27.         System.out.println("服务器的地址： "+ ctx.channel().remoteAddress());
  28.     }
  30.     //发生异常时
  31.     @Override
  32.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  34.         cause.printStackTrace();
  35.         ctx.close();
  36.     }
  37. }

Netty-Http服务编写

Netty 服务器在 6668 端口监听，浏览器发出请求 “http://localhost:6668/ “

服务器可以回复消息给客户端 “Hello! 我是服务器 “ , 并对特定请求资源进行过滤.

编写服务端代码：

  1. public class TestServer {
  3.     public static void main(String[] args) throws Exception {
  6.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  7.         EventLoopGroup workerGroup = new NioEventLoopGroup();
  9.         try {
  11.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  12.             //ServerBootstrap引导
  13.             serverBootstrap.group(bossGroup, workerGroup) //设置线程组
  14.                     .channel(NioServerSocketChannel.class) //设置ServerChannel
  15.                     .childHandler(new TestServerInitializer()); //设置接收的Channel的Handler
  16.             //异步绑定 省略注册监听
  17.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  18.             if (channelFuture.isDone()){
  20.                 System.out.println("服务器已绑定端口:7000");
  21.             }
  23.             //监听关闭通道
  24.             channelFuture.channel().closeFuture().sync();
  26.         }finally {
  28.             //优雅关闭服务器资源
  29.             bossGroup.shutdownGracefully();
  30.             workerGroup.shutdownGracefully();
  31.         }
  32.     }
  33. }

编写自定义Handler：

  1. /**
  2.  * 说明
  3.  * 1. SimpleChannelInboundHandler 是 ChannelInboundHandlerAdapter（适配器）
  4.  * 2. HttpObject 客户端和服务器端相互通讯的数据被封装成 HttpObject
  5.  */
  6. public class TestHttpServerHandler extends SimpleChannelInboundHandler<HttpObject> {
  9.     //读事件处理
  10.     @Override
  11.     protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
  14.         //判断 msg 是不是 httprequest请求
  15.         if(msg instanceof HttpRequest) {
  17.             System.out.println("服务端" + ctx.handler() + "开始处理...");
  18.             System.out.println("  " + "客户端地址" + ctx.channel().remoteAddress() + "发出请求...");
  19.             //获得请求
  20.             HttpRequest httpRequest = (HttpRequest) msg;
  21.             //获取uri, 过滤指定的资源
  22.             URI uri = new URI(httpRequest.uri());
  23.             //对请求网站图标的请求做过滤
  24.             if("/favicon.ico".equals(uri.getPath())) {
  26.                 System.out.println("请求了 favicon.ico, 不做响应...");
  27.                 return;
  28.             }
  30.             //回复信息给浏览器 [http协议]
  31.             ByteBuf content = Unpooled.copiedBuffer("hello, 我是服务器", CharsetUtil.UTF_8);
  32.             //构造一个http的相应，即 httpresponse
  33.             //HTTP_1_1代表HTTP版本，HttpResponseStatus.OK代表200状态码
  34.             FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, content);
  35.             //对HTTP头设置
  36.             response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain");
  37.             response.headers().set(HttpHeaderNames.CONTENT_LENGTH, content.readableBytes());
  38.             //将构建好的response写回客户端
  39.             ctx.writeAndFlush(response);
  40.         }
  41.     }
  42. }

自定义的Handler需要ChannelInitializer进行加入

  1. public class TestServerInitializer extends ChannelInitializer<SocketChannel> {
  4.     /**
  5.      * 在Channel初始化的时候设置好。
  6.      * 每次请求配置一次管道。
  7.      * */
  8.     @Override
  9.     protected void initChannel(SocketChannel ch) throws Exception {
  11.         //通过Channel得到管道
  12.         ChannelPipeline pipeline = ch.pipeline();
  13.         //向管道加入Handler
  14.         //HttpServerCodec是netty提供的处理http的编-解码器
  15.         System.out.println("管道加入HttpServerCodec...");
  16.         pipeline.addLast("MyHttpServerCodec",new HttpServerCodec());
  17.         System.out.println("管道加入自定义Handler...");
  18.         pipeline.addLast("MyTestHttpServerHandler", new TestHttpServerHandler());
  19.         System.out.println("管道处理器配置成功....");
  20.     }
  21. }

Netty-群聊系统

编写一个 Netty 群聊系统，实现服务器端和客户端之间的数据简单通讯（非阻塞）

要求：

1.实现多人群聊

2.服务器端：可以监测用户上线，离线，并实现消息转发功能

3.客户端：通过 channel 可以无阻塞发送消息给其它所有用户，同时可以接受其它用户发送的消息

4.使用TCP实现

服务端编写：

  1. public class GroupChatServer {
  3.     private int port; //监听端口
  4.     public GroupChatServer(int port) {
  6.         this.port = port;
  7.     }
  9.     //编写run方法，处理客户端的请求
  10.     public void run() throws Exception{
  13.         //创建两个线程组
  14.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  15.         EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop
  17.         try {
  19.             //服务端Netty引导
  20.             ServerBootstrap b = new ServerBootstrap();
  21.             b.group(bossGroup, workerGroup)
  22.                     .channel(NioServerSocketChannel.class) //基于TCP
  23.                     .option(ChannelOption.SO_BACKLOG, 128) //连接队列 128
  24.                     .childOption(ChannelOption.SO_KEEPALIVE, true) //配置客户端通道保持通道活跃
  25.                     .childHandler(new GroupChatServerInitializer());
  27.             ChannelFuture channelFuture = b.bind(port).sync();
  28.             if (channelFuture.isDone()){
  30.                 System.out.println("服务端绑定端口" + port + "成功...");
  31.             }
  34.             //监听关闭通道
  35.             channelFuture.channel().closeFuture().sync();
  36.         }finally {
  38.             bossGroup.shutdownGracefully();
  39.             workerGroup.shutdownGracefully();
  40.         }
  42.     }
  44.     public static void main(String[] args) throws Exception {
  47.         new GroupChatServer(7000).run();
  48.     }
  49. }

服务端GroupChatClientInitializer配置：

  1. /**
  2.  * 管道配置 Channel初始化时被调用
  3.  * */
  4. public class GroupChatClientInitializer extends ChannelInitializer<SocketChannel> {
  7.     @Override
  8.     protected void initChannel(SocketChannel ch) throws Exception {
  10.         //得到pipeline
  11.         ChannelPipeline pipeline = ch.pipeline();
  12.         //客户端收到消息要解码 发送的消息要编码
  13.         //注意在出站事件和入站事件，这两个Handler的处理顺序是不同的，这由netty帮我们完成
  14.         pipeline.addLast("decoder", new StringDecoder());
  15.         pipeline.addLast("encoder", new StringEncoder());
  16.         //向管道加入自定义的handler
  17.         pipeline.addLast(new GroupChatClientHandler());
  18.     }
  19. }

服务端Handler编写

  1. public class GroupChatServerHandler extends SimpleChannelInboundHandler<String> {
  3.     //这是一个负责群聊的核心Handler
  4.     //定义一个channel 组，neety支持Handler对部分通道进行直接管理
  5.     //GlobalEventExecutor.INSTANCE) 是全局的事件执行器，是一个单例对象
  6.     private static ChannelGroup  channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
  7.     //定义时间
  8.     SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  11.     //handlerAdded 表示连接建立，一旦连接，第一个被执行
  12.     @Override
  13.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  15.         Channel channel = ctx.channel();
  16.         //将当前channel 加入到  channelGroup
  17.         channelGroup.add(channel);
  19.         //推送所有在线用户上线人信息
  20.         //该方法会将 channelGroup 中所有的channel 遍历，并发送消息，
  21.         channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 加入聊天" + sdf.format(new java.util.Date()) + " \n");
  22.     }
  24.     //断开连接, 将xx客户离开信息推送给当前在线的客户
  25.     @Override
  26.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  28.         Channel channel = ctx.channel();
  29.         channelGroup.writeAndFlush("[客户端]" + channel.remoteAddress() + " 离开了\n");
  30.         //一旦某个客户端断开，自动从channel组中去除
  31.     }
  33.     //表示channel 处于活动状态, 提示 xx上线
  34.     @Override
  35.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  37.         System.out.println(ctx.channel().remoteAddress() + " 上线了~");
  38.     }
  40.     //表示channel 处于不活动状态, 提示 xx离线了
  41.     @Override
  42.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  44.         System.out.println(ctx.channel().remoteAddress() + " 离线了~");
  45.     }
  47.     //读取数据 当客户端发送聊天信息时触发
  48.     @Override
  49.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  51.         //获取到当前channel
  52.         Channel channel = ctx.channel();
  53.         //这时我们遍历channelGroup, 根据不同的情况，回送不同的消息
  54.         channelGroup.forEach(ch -> {
  56.             //转发消息
  57.             if(channel != ch) {
  59.                 ch.writeAndFlush("[客户]" + channel.remoteAddress() + " 发送了消息" + msg + "\n");
  60.             }else {
  62.                 //回显给发送了消息的客户端
  63.                 ch.writeAndFlush("[自己]发送了消息" + msg + "\n");
  64.             }
  65.         });
  66.     }
  68.     @Override
  69.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  71.         //关闭通道
  72.         ctx.close();
  73.     }
  74. }

客户端编写：

  1. import java.util.Scanner;
  3. public class GroupChatClient {
  6.     //用户连接目标ip+端口
  7.     private final String host;
  8.     private final int port;
  10.     public GroupChatClient(String host, int port) {
  12.         this.host = host;
  13.         this.port = port;
  14.     }
  16.     //启动客户端
  17.     public void run() throws Exception{
  19.         //客户端一个事件循环组
  20.         EventLoopGroup group = new NioEventLoopGroup();
  21.         try {
  23.         Bootstrap bootstrap = new Bootstrap()
  24.                 .group(group)
  25.                 .channel(NioSocketChannel.class) 
  26.                 .handler(new GroupChatClientInitializer());
  27.         //异步连接
  28.         ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
  29.         ///拿到连接后的通道，由于上面连接操作是异步，这里必然是阻塞    
  30.         Channel channel = channelFuture.channel();
  31.         System.out.println("客户端channel已建立:-------" + channel.localAddress()+ "--------");
  32.         //客户端需要输入信息
  33.         Scanner scanner = new Scanner(System.in);
  34.         while (scanner.hasNextLine()) {
  36.             //换行即发送
  37.             String msg = scanner.nextLine();
  38.             //通过channel 发送到服务器端
  39.             channel.writeAndFlush(msg + "\r\n");
  40.             }
  41.         }finally {
  43.             group.shutdownGracefully();
  44.         }
  45.     }
  47.     public static void main(String[] args) throws Exception {
  49.         new GroupChatClient("127.0.0.1", 7000).run();
  50.     }
  51. }

客户端管道配置：

  1. /**
  2.  * 管道配置 Channel初始化时被调用
  3.  * */
  4. public class GroupChatClientInitializer extends ChannelInitializer<SocketChannel> {
  7.     @Override
  8.     protected void initChannel(SocketChannel ch) throws Exception {
  10.         //得到pipeline
  11.         ChannelPipeline pipeline = ch.pipeline();
  12.         //编解码器
  13.         pipeline.addLast("decoder", new StringDecoder());
  14.         pipeline.addLast("encoder", new StringEncoder());
  15.         //向管道加入自定义的handler
  16.         pipeline.addLast(new GroupChatClientHandler());
  17.     }
  18. }

自定义客户端Handler：

  1. public class GroupChatClientHandler extends SimpleChannelInboundHandler<String> {
  3.     @Override
  4.     //直接读就完事
  5.     protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
  7.         System.out.println(msg.trim());
  8.     }
  9. }

Netty-心跳检测机制

当一个连接进来，长时间没有操作，我们可以使用心跳机制得知该连接是否已经断连。

如未断连，我们可以直接给它断了不让它长时间持有服务器的资源

netty提供的IdleStateHandler可以帮助我们完成心跳检测机制

IdleStateHandler 是netty 提供的处理空闲状态的处理器

  1. 1. long readerIdleTime : 表示多长时间没有读, 就会发送一个心跳检测包检测是否连接
  2.   2. long writerIdleTime : 表示多长时间没有写, 就会发送一个心跳检测包检测是否连接
  3.   3. long allIdleTime : 表示多长时间没有读写, 就会发送一个心跳检测包检测是否连接

当 IdleStateEvent 触发后 , 就会传递给管道 的下一个handler去处理

通过调用(触发)下一个handler 的 userEventTiggered , 在该方法中去处理 IdleStateEvent(读空闲，写空闲，读写空闲)

  1. public class MyServer {
  3.     public static void main(String[] args) throws Exception{
  7.         //创建两个线程组
  8.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  9.         EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop
  10.         try {
  13.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  14.             serverBootstrap.group(bossGroup, workerGroup)
  15.                     .channel(NioServerSocketChannel.class)
  16.                     .handler(new LoggingHandler(LogLevel.INFO))
  17.                     .childHandler(new ChannelInitializer<SocketChannel>() {
  19.                 @Override
  20.                 protected void initChannel(SocketChannel ch) throws Exception {
  22.                     ChannelPipeline pipeline = ch.pipeline();
  23.                     //7s不读则读空闲，7秒不写则写空闲，10秒不读写则读写空闲
  24.                     pipeline.addLast(new IdleStateHandler(3,5,7, TimeUnit.SECONDS));
  25.                     //加入一个对空闲检测进一步处理的handler(自定义)
  26.                     pipeline.addLast(new MyServerHandler());
  27.                 }
  28.             });
  30.             //启动服务器
  31.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  32.             channelFuture.channel().closeFuture().sync();
  34.         }finally {
  36.             bossGroup.shutdownGracefully();
  37.             workerGroup.shutdownGracefully();
  38.         }
  39.     }
  40. }

对空闲事件的处理Handler：

  1. public class MyServerHandler extends ChannelInboundHandlerAdapter {
  3.     @Override
  4.     public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  7.         //如果事件属于IdleStateEvent
  8.         if(evt instanceof IdleStateEvent) {
  10.             //将evt 向下转型 IdleStateEvent
  11.             IdleStateEvent event = (IdleStateEvent) evt;
  12.             String eventType = null;
  13.             //判断到底是哪个空间
  14.             switch (event.state()) {
  16.                 case READER_IDLE:
  17.                   eventType = "读空闲";
  18.                   break;
  19.                 case WRITER_IDLE:
  20.                     eventType = "写空闲";
  21.                     break;
  22.                 case ALL_IDLE:
  23.                     eventType = "读写空闲";
  24.                     break;
  25.             }
  26.             System.out.println(ctx.channel().remoteAddress() + "--长时间未操作--" + eventType);
  27.             //如果发生空闲，我们关闭通道
  28.            // ctx.channel().close();
  29.         }
  30.     }
  31. }

WebSocket 编程实现BS长连接

Http 协议是无状态的, 浏览器和服务器间的请求响应一次，下一次会重新创建连接

如果要实现Http的长连接，就得基于webSocket编程

通过改变并提供新的Handler添加到我们的Channel就行了

服务端编写：

  1. public class MyServer {
  3.     public static void main(String[] args) throws Exception{
  5.         //创建两个线程组
  6.         EventLoopGroup bossGroup = new NioEventLoopGroup(1);
  7.         EventLoopGroup workerGroup = new NioEventLoopGroup(); //8个NioEventLoop
  8.         try {
  10.             ServerBootstrap serverBootstrap = new ServerBootstrap();
  11.             serverBootstrap.group(bossGroup, workerGroup);
  12.             serverBootstrap.channel(NioServerSocketChannel.class);
  13.             serverBootstrap.handler(new LoggingHandler(LogLevel.INFO));
  14.             serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
  17.                 @Override
  18.                 protected void initChannel(SocketChannel ch) throws Exception {
  20.                     ChannelPipeline pipeline = ch.pipeline();
  21.                     //因为基于http协议，使用http的编码和解码器
  22.                     pipeline.addLast(new HttpServerCodec());
  23.                     //以块方式写，添加ChunkedWriteHandler处理器
  24.                     pipeline.addLast(new ChunkedWriteHandler());
  26.                     //1. http数据在传输过程中是分段, HttpObjectAggregator，可以将多个段聚合
  27.                     //2. 这也就是为什么，当浏览器发送大量数据时，就会发出多次http请求
  28.                     pipeline.addLast(new HttpObjectAggregator(8192));
  30.                     //websocket数据是以 帧(frame) 形式传递
  32.                     //浏览器请求时 ws://localhost:7000/hello 表示请求的uri
  33.                     //WebSocketServerProtocolHandler 核心功能是将 http协议升级为ws协议,保持长连接
  34.                     pipeline.addLast(new WebSocketServerProtocolHandler("/hello"));
  36.                     //自定义的handler ，处理业务逻辑
  37.                     pipeline.addLast(new MyTextWebSocketFrameHandler());
  38.                 }
  39.             });
  41.             //启动服务器
  42.             ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
  43.             channelFuture.channel().closeFuture().sync();
  45.         }finally {
  47.             bossGroup.shutdownGracefully();
  48.             workerGroup.shutdownGracefully();
  49.         }
  50.     }
  51. }

Handler：

  1. //这里 TextWebSocketFrame 类型，数据在Handler链中以文本帧(frame)传递 
  2. public class MyTextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{
  4.     @Override
  5.     protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
  7.         System.out.println("服务器收到消息 " + msg.text());
  8.         //回复消息
  9.         ctx.channel().writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " " + msg.text()));
  10.     }
  11.     //当web客户端连接后，立即触发
  12.     @Override
  13.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
  15.         //id 表示唯一的值，LongText 是唯一的 ShortText 不是唯一
  16.         System.out.println("handlerAdded 被调用" + ctx.channel().id().asLongText());
  17.         System.out.println("handlerAdded 被调用" + ctx.channel().id().asShortText());
  18.     }
  19.     @Override
  20.     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
  22.         System.out.println("handlerRemoved 被调用" + ctx.channel().id().asLongText());
  23.     }
  24.     @Override
  25.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  27.         System.out.println("异常发生 " + cause.getMessage());
  28.         ctx.close(); //关闭连接
  29.     }
  30. }

客户端_JS

  1. <!DOCTYPE html>
  2. <html lang="en">
  3. <head>
  4.     <meta charset="UTF-8"请求qet="UTF-8"ung'qiuqet="UTF-8"请求qet="UTF-8">
  5.     <title>Title</title>
  6. </head>
  7. <body>
  8. <script>
  9.     var socket;
  10.     //判断当前浏览器是否支持websocket
  11.     if(window.WebSocket) {
  13.         //访问服务端指定的ws路径，返回socket
  14.         socket = new WebSocket("ws://localhost:7000/hello");
  16.         //收到服务器端回送的消息后进行回调
  17.         socket.onmessage = function (ev) {
  19.             //获得文本框
  20.             var rt = document.getElementById("responseText");
  21.             //将服务器回送的消息放上去
  22.             rt.value = rt.value + "\n" + ev.data;
  23.         }
  25.         //连接开启(感知到连接开启)后回调
  26.         socket.onopen = function (ev) {
  28.             var rt = document.getElementById("responseText");
  29.             rt.value = "连接开启了.."
  30.         }
  32.         //连接关闭(感知到连接关闭)后回调
  33.         socket.onclose = function (ev) {
  35.             var rt = document.getElementById("responseText");
  36.             rt.value = rt.value + "\n" + "连接关闭了.."
  37.         }
  38.     } else {
  40.         alert("当前浏览器不支持websocket")
  41.     }
  43.     //点击发送消息后触发
  44.     function send(message) {
  46.         //先判断socket是否创建好
  47.         if(!window.socket) {
  49.             return;
  50.         }
  51.         //socket准备好了
  52.         if(socket.readyState == WebSocket.OPEN) {
  54.             //通过socket 发送消息
  55.             socket.send(message)
  56.         } else {
  58.             alert("连接没有开启");
  59.         }
  60.     }
  61. </script>
  62.     <form onsubmit="return false">
  63.         <textarea name="message" style="height: 300px; width: 300px"></textarea>
  64.         <input type="button" value="发生消息" onclick="send(this.form.message.value)">
  65.         <textarea id="responseText" style="height: 300px; width: 300px"></textarea>
  66.         <input type="button" value="清空内容" onclick="document.getElementById('responseText').value=''">
  67.     </form>
  68. </body>
  69. </html>

发表评论

表情：
评论列表 （有 0 条评论，347人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Netty 核心组件

    Netty 核心组件 1. 前言 本节我们主要从整体上了解 Netty 有哪些核心组件，很多同学学习完 Netty 虽然会使用，但是只知道如何自定义 Handler

    清疚清疚/ 0 赞/ 259 阅读

    相关 Netty核心组件

    Netty的核心组件 　　Channel 　　回调 　　Future 　　事件 　　ChannelHandler 　　这些模块代表了不同的类型：资源（事件）、逻

    傷城~傷城~/ 0 赞/ 338 阅读