Tcp keepalive和netty心跳
什么是心跳机制？
心跳说的是在客户端和服务端在互相建立ESTABLISH状态的时候，如何通过发送一个最简单的包来保持连接的存活，还有监控另一边服务的可用性等。
心跳包的作用
我们的心跳包就是为了防止Socket断开连接，或是TCP的连接断开吗？
答案是否定的，TCP连接的通道是个虚拟的，连接的维持靠的是两端TCP软件对连接状态的维护。
TCP 连接自身有维护连接的机制，说白了就是自身有长时间没有数据包情况下的判断连接是否还存在的检测，清除死连接，即使在没有数据来往的时候，TCP也就可以（在启动TCP这个功能的前提下）自动发包检测是否连接正常，这个不需要我们处理。
设计心跳包的目的
探知对端应用是否存活，服务端客户端都可以发心跳包，一般都是客户端发送心跳包，服务端用于判断客户端是否在线，从而对服务端内存缓存数据进行清理（玩家下线等）；问题在于，通过TCP四次握手断开的设定，我们也是可以通过Socket的read方法来判断TCP连接是否断开，从而做出相应的清理内存动作，那么为什么我们还需要使用客户端发送心跳包来判断呢？
在Java的阻塞编程中：通过
ServerSocket ss = new ServerSocket(10021);
Socket so = ss.accept();
// 获取相关流对象
InputStream in = so.getInputStream();
byte[] bytes = new byte[1024];
int num = in.read(bytes);
if（num == -1）//表明读到了流的末尾，事实也就是client端断开了连接，比如调用close()
{
so.close();
}
在Java的非阻塞编程当中：通过
SelectionKey key = selector.register（socketChannel,ops,handle）;
SocketChannel socketChanel = （SocketChannel）key.channel();
ByteBuffer buffer = ByteBuffer.allocate（1024）；
int num = socketChannel.read(buffer);
if(num == -1)
{
key.channel().close();
}
上述连接处理方式，返回-1,都是收到了客户端的fin或者rst之后的反应，所以根据四次分手原则，我们调用close方法，发送fin给客户端。
上面这种策略通过TCP协议的返回值来得知客户端TCP断开，从而得知客户端掉线。
那么这种方式有什么不完美呢？或者说有什么缺陷呢？
**主要原因就是TCP的断开可能有时候是不能瞬时探知的，甚至是不能探知的，也可能有很长时间的延迟，如果客户端没有正常的断开TCP连接，四次握手没有发起，服务端无从得知客户端的掉线，那么就要依靠开启TCP的keep alive机制，but TCP协议的keep alive机制是不建议开启的，即使开启了默认的时间间隔是2h，泪奔！如果要服务端维持一个2h的死链接，那是可怕的，如果我们调整了时间间隔，也是有problem的，因为TCP本身就不建议TCP层的心跳检测，因为这可能导致一个完好的TCP连接在中间网络中断的情况下重启
**
什么是keepalive定时器
在一个空闲的（idle）TCP连接上，没有任何的数据流，许多TCP/IP的初学者都对此感到惊奇。也就是说，如果TCP连接两端没有任何一个进程在向对方发送数据，那么在这两个TCP模块之间没有任何的数据交换。你可能在其它的网络协议中发现有轮询（polling），但在TCP中它不存在。言外之意就是我们只要启动一个客户端进程，同服务器建立了TCP连接，不管你离开几小时，几天，几星期或是几个月，连接依旧存在。中间的路由器可能崩溃或者重启，电话线可能go down或者back up，只要连接两端的主机没有重启，连接依旧保持建立。
这就可以认为不管是客户端的还是服务器端的应用程序都没有应用程序级（application-level）的定时器来探测连接的不活动状态（inactivity），从而引起任何一个应用程序的终止。然而有的时候，服务器需要知道客户端主机是否已崩溃并且关闭，或者崩溃但重启。许多实现提供了存活定时器来完成这个任务。
存活定时器是一个包含争议的特征。许多人认为，即使需要这个特征，这种对对方的轮询也应该由应用程序来完成，而不是由TCP中实现。此外，如果两个终端系统之间的某个中间网络上有连接的暂时中断，那么存活选项（option）就能够引起两个进程间一个良好连接的终止。例如，如果正好在某个中间路由器崩溃、重启的时候发送存活探测，TCP就将会认为客户端主机已经崩溃，但事实并非如此。
存活（keepalive）并不是TCP规范的一部分。在Host Requirements RFC罗列有不使用它的三个理由：（1）在短暂的故障期间，它们可能引起一个良好连接（good connection）被释放（dropped），（2）它们消费了不必要的宽带，（3）在以数据包计费的互联网上它们（额外）花费金钱。然而，在许多的实现中提供了存活定时器。
我们称使用存活选项的那一段为服务器，另一端为客户端。也可以在客户端设置该选项，且没有不允许这样做的理由，但通常设置在服务器。如果连接两端都需要探测对方是否消失，那么就可以在两端同时设置（比如NFS）。
若在一个给定连接上，两小时之内无任何活动，服务器便向客户端发送一个探测段。
-
客户端主机依旧活跃（up）运行，并且从服务器可到达。从客户端TCP的正常响应，服务器知道对方仍然活跃。服务器的TCP为接下来的两小时复位存活定时器，如果在这两个小时到期之前，连接上发生应用程序的通信，则定时器重新为往下的两小时复位，并且接着交换数据。
-
客户端已经崩溃，或者已经关闭（down），或者正在重启过程中。在这两种情况下，它的TCP都不会响应。服务器没有收到对其发出探测的响应，并且在75秒之后超时。服务器将总共发送10个这样的探测，每个探测75秒。如果没有收到一个响应，它就认为客户端主机已经关闭并终止连接
-
客户端曾经崩溃，但已经重启。这种情况下，服务器将会收到对其存活探测的响应，但该响应是一个复位，从而引起服务器对连接的终止。
-
客户端主机活跃运行，但从服务器不可到达。这与状态2类似，因为TCP无法区别它们两个。它所能表明的仅是未收到对其探测的回复。
在第一种状态下，服务器应用程序不知道存活探测是否发生。凡事都是由TCP层处理的，存活探测对应用程序透明，直到后面2，3，4三种状态发生。在这三种状态下，通过服务器的TCP，返回给服务器应用程序错误信息。（通常服务器向网络发出一个读请求，等待客户端的数据。如果存活特征返回一个错误信息，则将该信息作为读操作的返回值返回给服务器。）在状态2，错误信息类似于“连接超时”。状态3则为“连接被对方复位”。第四种状态看起来像连接超时，或者根据是否收到与该连接相关的ICMP错误信息，而可能返回其它的错误信息。
在TCP协议的机制里面，本身的心跳包机制，也就是TCP协议中的SO_KEEPALIVE，系统默认是设置2小时的心跳频率。需要用要用setsockopt将SOL_SOCKET.SO_KEEPALIVE设置为1才是打开，并且可以设置三个参数tcp_keepalive_time/tcp_keepalive_probes/tcp_keepalive_intvl，分别表示连接闲置多久开始发keepalive的ACK包、发几个ACK包不回复才当对方死了、两个ACK包之间间隔多长。TCP协议会向对方发一个带有ACK标志的空数据包（KeepAlive探针），对方在收到ACK包以后，如果连接一切正常，应该回复一个ACK；如果连接出现错误了（例如对方重启了，连接状态丢失），则应当回复一个RST；如果对方没有回复，服务器每隔多少时间再发ACK，如果连续多个包都被无视了，说明连接被断开了。
那么既然有TCP的心跳机制，我们为什么还要在应用层实现自己的心跳检测机制呢？
keepalive设计初衷清除和回收死亡时间长的连接，不适合实时性高的场合，而且它会先要求连接一定时间内没有活动，周期长，这样其实已经断开很长一段时间，没有及时性。
应用层的心跳包
心跳包，通常是客户端每隔一小段时间向服务器发送的一个数据包，通知服务器自己仍然在线，服务器与客户端之间每隔一段时间 进行一次交互,来判断这个链接是否有效，并传输一些可能有必要的数据。通常是在建立了一个TCP的socket连接后无法保证这个连接是否持续有效，这时候两边应用会通过定时发送心跳包来保证连接是有效的。因按照一定的时间间隔发送，类似于心跳，所以叫做心跳包。事实上为了保持长连接（长连接指的是建立一次TCP连接之后，就认为连接有效，利用这个连接去不断传输数据，不断开TCP连接），至于包的内容，是没有特别规定的，不过一般都是很小的包，或者只是包含包头的一个空包。
那么心跳包的意义就在于方便的在服务端管理客户端的在线情况，并且可以防止TCP的死连接问题，避免出现长时间不在线的死链接仍然出现在服务端的管理任务中。
netty心跳
Netty 提供了 IdleStateHandler ，ReadTimeoutHandler，WriteTimeoutHandler 检测连接的有效性。当然，你也可以自己写个Handler。
IdleStateHandler： 当连接的空闲时间（读或者写）太长时，将会触发一个 IdleStateEvent 事件。然后，你可以通过你的 ChannelInboundHandler 中重写 userEventTrigged 方法来处理该事件。
IdleStateHandler 既是出站处理器也是入站处理器，继承了 ChannelDuplexHandler 。通常在 initChannel 方法中将 IdleStateHandler 添加到 pipeline 中。然后在自己的 handler 中重写 userEventTriggered 方法，当发生空闲事件（读或者写），就会触发这个方法，并传入具体事件。
源码解析
重要属性：
private final boolean observeOutput;// 是否考虑出站时较慢的情况。默认值是false（不考虑）。
private final long readerIdleTimeNanos; // 读事件空闲时间，0 则禁用事件
private final long writerIdleTimeNanos;// 写事件空闲时间，0 则禁用事件
private final long allIdleTimeNanos; //读或写空闲时间，0 则禁用事件
当该 handler 被添加到 pipeline 中时，则调用 initialize 方法：
private void initialize(ChannelHandlerContext ctx) {
switch (state) {
case 1:
case 2:
return;
}
state = 1;
initOutputChanged(ctx);
lastReadTime = lastWriteTime = ticksInNanos();
if (readerIdleTimeNanos > 0) {
// 这里的 schedule 方法会调用 eventLoop 的 schedule 方法，将定时任务添加进队列中
readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
readerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (writerIdleTimeNanos > 0) {
writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
writerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (allIdleTimeNanos > 0) {
allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
allIdleTimeNanos, TimeUnit.NANOSECONDS);
}
}
只要给定的参数大于0，就创建一个定时任务，每个事件都创建。同时，将 state 状态设置为 1，防止重复初始化。调用 initOutputChanged 方法，初始化 “监控出站数据属性”，代码如下：
private void initOutputChanged(ChannelHandlerContext ctx) {
if (observeOutput) {
Channel channel = ctx.channel();
Unsafe unsafe = channel.unsafe();
ChannelOutboundBuffer buf = unsafe.outboundBuffer();
// 记录了出站缓冲区相关的数据，buf 对象的 hash 码，和 buf 的剩余缓冲字节数
if (buf != null) {
lastMessageHashCode = System.identityHashCode(buf.current());
lastPendingWriteBytes = buf.totalPendingWriteBytes();
}
}
}
这个 observeOutput “监控出站数据属性” 有什么作用？
假设：当你的客户端应用每次接收数据是30秒，而你的写空闲时间是 25 秒，那么，当你数据还没有写出的时候，写空闲时间触发了。实际上是不合乎逻辑的。因为你的应用根本不空闲。
怎么解决呢？
Netty 的解决方案是：记录最后一次输出消息的相关信息，并使用一个值 firstXXXXIdleEvent 表示是否再次活动过，每次读写活动都会将对应的 first 值更新为 true，如果是 false，说明这段时间没有发生过读写事件。同时如果第一次记录出站的相关数据和第二次得到的出站相关数据不同，则说明数据在缓慢的出站，就不用触发空闲事件。
总的来说，这个字段就是用来对付 “客户端接收数据奇慢无比，慢到比空闲时间还多” 的极端情况。所以，Netty 默认是关闭这个字段的。
-
该类内部的 3 个定时任务类
image.png
这 3 个定时任务分别对应 读，写，读或者写 事件。共有一个父类。这个父类提供了一个模板方法：
image.png
当通道关闭了，就不执行任务了。反之，执行子类的 run 方法。
读事件：
protected void run(ChannelHandlerContext ctx) {
long nextDelay = readerIdleTimeNanos;
if (!reading) {
nextDelay -= ticksInNanos() - lastReadTime;
}
if (nextDelay <= 0) {
// Reader is idle - set a new timeout and notify the callback.
// 用于取消任务 promise
readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
boolean first = firstReaderIdleEvent;
firstReaderIdleEvent = false;
try {
// 再次提交任务
IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
// 触发用户 handler use
channelIdle(ctx, event);
} catch (Throwable t) {
ctx.fireExceptionCaught(t);
}
} else {
// Read occurred before the timeout - set a new timeout with shorter delay.
readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
}
}
该方法很简单：
- 得到用户设置的超时时间。
- 如果读取操作结束了（执行了 channelReadComplete 方法设置） ，就用当前时间减去给定时间和最后一次读操作的时间（执行了 channelReadComplete 方法设置），如果小于0，就触发事件。反之，继续放入队列。间隔时间是新的计算时间。
- 触发的逻辑是：首先将任务再次放到队列，时间是刚开始设置的时间，返回一个 promise 对象，用于做取消操作。然后，设置 first 属性为 false ，表示，下一次读取不再是第一次了，这个属性在 channelRead 方法会被改成 true。
- 创建一个 IdleStateEvent 类型的写事件对象，将此对象传递给用户的 UserEventTriggered 方法。完成触发事件的操作。
总的来说，每次读取操作都会记录一个时间，定时任务时间到了，会计算当前时间和最后一次读的时间的间隔，如果间隔超过了设置的时间，就触发 UserEventTriggered 方法。就是这么简单。
再看看写事件任务。
写事件的 run 方法
写任务的逻辑基本和读任务的逻辑一样，唯一不同的就是有一个针对 出站较慢数据的判断。
if (hasOutputChanged(ctx, first)) {
return;
}
如果这个方法返回 true，就不执行触发事件操作了，即使时间到了。看看该方法实现：
private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {
if (observeOutput) {
// 如果最后一次写的时间和上一次记录的时间不一样，说明写操作进行过了，则更新此值
if (lastChangeCheckTimeStamp != lastWriteTime) {
lastChangeCheckTimeStamp = lastWriteTime;
// 但如果，在这个方法的调用间隙修改的，就仍然不触发事件
if (!first) { // #firstWriterIdleEvent or #firstAllIdleEvent
return true;
}
}
Channel channel = ctx.channel();
Unsafe unsafe = channel.unsafe();
ChannelOutboundBuffer buf = unsafe.outboundBuffer();
// 如果出站区有数据
if (buf != null) {
// 拿到出站缓冲区的 对象 hashcode
int messageHashCode = System.identityHashCode(buf.current());
// 拿到这个 缓冲区的 所有字节
long pendingWriteBytes = buf.totalPendingWriteBytes();
// 如果和之前的不相等，或者字节数不同，说明，输出有变化，将 "最后一个缓冲区引用" 和 “剩余字节数” 刷新
if (messageHashCode != lastMessageHashCode || pendingWriteBytes != lastPendingWriteBytes) {
lastMessageHashCode = messageHashCode;
lastPendingWriteBytes = pendingWriteBytes;
// 如果写操作没有进行过，则任务写的慢，不触发空闲事件
if (!first) {
return true;
}
}
}
}
return false;
}
- 如果用户没有设置了需要观察出站情况。就返回 false，继续执行事件。
- 反之，继续向下， 如果最后一次写的时间和上一次记录的时间不一样，说明写操作刚刚做过了，则更新此值，但仍然需要判断这个 first 的值，如果这个值还是 false，说明在这个写事件是在两个方法调用间隙完成的 / 或者是第一次访问这个方法，就仍然不触发事件。
- 如果不满足上面的条件，就取出缓冲区对象，如果缓冲区没对象了，说明没有发生写的很慢的事件，就触发空闲事件。反之，记录当前缓冲区对象的 hashcode 和 剩余字节数，再和之前的比较，如果任意一个不相等，说明数据在变化，或者说数据在慢慢的写出去。那么就更新这两个值，留在下一次判断。
- 继续判断 first ，如果是 fasle，说明这是第二次调用，就不用触发空闲事件了。
这里有个问题，为什么第一次的时候一定要触发事件呢？假设，客户端开始变得很慢，这个时候，定时任务监听发现时间到了，就进入这里判断，当上次记录的缓冲区相关数据已经不同，这个时候难道触发事件吗？
实际上，这里是 Netty 的一个考虑：假设真的发生了很写出速度很慢的问题，很可能引发 OOM，相比叫连接空闲，这要严重多了。为什么第一次一定要触发事件呢？如果不触发，用户根本不知道发送了什么，当一次写空闲事件触发，随后出现了 OOM，用户可以感知到：可能是写的太慢，后面的数据根本写不进去，所以发生了OOM。所以，这里的一次警告还是必要的。
所有事件的 run 方法
这个类叫做 AllIdleTimeoutTask ，表示这个监控着所有的事件。当读写事件发生时，都会记录。代码逻辑和写事件的的基本一致，除了这里：
long nextDelay = allIdleTimeNanos;
if (!reading) {
// 当前时间减去 最后一次写或读 的时间 ，若大于0，说明超时了
nextDelay -= ticksInNanos() - Math.max(lastReadTime, lastWriteTime);
}
这里的时间计算是取读写事件中的最大值来的。然后像写事件一样，判断是否发生了写的慢的情况。最后调用 ctx.fireUserEventTriggered(evt) 方法。
通常这个使用的是最多的。构造方法一般是：
pipeline.addLast(new IdleStateHandler(0, 0, 30, TimeUnit.SECONDS));
读写都是 0 表示禁用，30 表示 30 秒内没有任务读写事件发生，就触发事件。注意，当不是 0 的时候，这三个任务会重叠。
小结
IdleStateHandler 可以实现心跳功能，当服务器和客户端没有任何读写交互时，并超过了给定的时间，则会触发用户 handler 的 userEventTriggered 方法。用户可以在这个方法中尝试向对方发送信息，如果发送失败，则关闭连接。
IdleStateHandler 的实现基于 EventLoop 的定时任务，每次读写都会记录一个值，在定时任务运行的时候，通过计算当前时间和设置时间和上次事件发生时间的结果，来判断是否空闲。
内部有 3 个定时任务，分别对应读事件，写事件，读写事件。通常用户监听读写事件就足够了。
同时，IdleStateHandler 内部也考虑了一些极端情况：客户端接收缓慢，一次接收数据的速度超过了设置的空闲时间。Netty 通过构造方法中的 observeOutput 属性来决定是否对出站缓冲区的情况进行判断。
如果出站缓慢，Netty 不认为这是空闲，也就不触发空闲事件。但第一次无论如何也是要触发的。因为第一次无法判断是出站缓慢还是空闲。当然，出站缓慢的话，OOM 比空闲的问题更大。
所以，当你的应用出现了内存溢出，OOM之类，并且写空闲极少发生（使用了 observeOutput 为 true），那么就需要注意是不是数据出站速度过慢。
默认 observeOutput 是 false，意思是，即使你的应用出站缓慢，Netty 认为是写空闲。
实战
所以在channelPipeline中加入IdleStateHandler，我们在handler中提示的是5秒读，所以我们服务端的配置的是：
ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
因为服务端必须5秒接受一次心跳请求， 那么客户端的配置:
ph.addLast( new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));
userEventTriggered是Netty 处理心跳超时事件，在IdleStateHandler设置超时时间，如果达到了，就会直接调用该方法。如果没有超时则不调用。我们重写该方法的话，就可以自行进行相关的业务逻辑处理了。
完整的代码如下:
服务端:
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;
/**
*
* Title: NettyServer
* Description: Netty服务端
*/
public class NettyServer {
private static final int port = 9876; //设置服务端端口
private static EventLoopGroup group = new NioEventLoopGroup(); // 通过nio方式来接收连接和处理连接
private static ServerBootstrap b = new ServerBootstrap();
/**
* Netty创建全部都是实现自AbstractBootstrap。
* 客户端的是Bootstrap，服务端的则是 ServerBootstrap。
**/
public static void main(String[] args) throws InterruptedException {
try {
b.group(group);
b.channel(NioServerSocketChannel.class);
b.childHandler(new NettyServerFilter()); //设置过滤器
// 服务器绑定端口监听
ChannelFuture f = b.bind(port).sync();
System.out.println("服务端启动成功,端口是:"+port);
// 监听服务器关闭监听
f.channel().closeFuture().sync();
} finally {
group.shutdownGracefully(); //关闭EventLoopGroup，释放掉所有资源包括创建的线程
}
}
}
服务端业务逻辑
业务逻辑这块，因为要重写userEventTriggered，所以继承ChannelInboundHandlerAdapter 。
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
/**
*
* Title: HelloServerHandler
* Description: 服务端业务逻辑
*/
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
/** 空闲次数 */
private int idle_count =1;
/** 发送次数 */
private int count = 1;
/**
* 超时处理
* 如果5秒没有接受客户端的心跳，就触发;
* 如果超过两次，则直接关闭;
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
if (obj instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) obj;
if (IdleState.READER_IDLE.equals(event.state())) { //如果读通道处于空闲状态，说明没有接收到心跳命令
System.out.println("已经5秒没有接收到客户端的信息了");
if (idle_count > 2) {
System.out.println("关闭这个不活跃的channel");
ctx.channel().close();
}
idle_count++;
}
} else {
super.userEventTriggered(ctx, obj);
}
}
/**
* 业务逻辑处理
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("第"+count+"次"+",服务端接受的消息:"+msg);
String message = (String) msg;
if ("hb_request".equals(message)) { //如果是心跳命令，则发送给客户端;否则什么都不做
ctx.write("服务端成功收到心跳信息");
ctx.flush();
}
count++;
}
/**
* 异常处理
*/
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
服务端过滤器
增加了心跳的相关设置。
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;
/**
*
* Title: HelloServerInitializer
* Description: Netty 服务端过滤器
*/
public class NettyServerFilter extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
// 以("\n")为结尾分割的 解码器
// ph.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
// 解码和编码，应和客户端一致
//入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式
ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
ph.addLast("decoder", new StringDecoder());
ph.addLast("encoder", new StringEncoder());
ph.addLast("handler", new NettyServerHandler());// 服务端业务逻辑
}
}
客户端
因为过滤器中没有使用DelimiterBasedFrameDecoder ，所以不必换行了。
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import java.io.IOException;
/**
*
* Title: NettyClient
* Description: Netty客户端
*/
public class NettyClient {
public static String host = "127.0.0.1"; //ip地址
public static int port = 9876; //端口
/// 通过nio方式来接收连接和处理连接
private static EventLoopGroup group = new NioEventLoopGroup();
private static Bootstrap b = new Bootstrap();
private static Channel ch;
/**
* Netty创建全部都是实现自AbstractBootstrap。
* 客户端的是Bootstrap，服务端的则是 ServerBootstrap。
**/
public static void main(String[] args) throws InterruptedException, IOException {
System.out.println("客户端成功启动...");
b.group(group);
b.channel(NioSocketChannel.class);
b.handler(new NettyClientFilter());
// 连接服务端
ch = b.connect(host, port).sync().channel();
star();
}
public static void star() throws IOException{
String str="Hello Netty";
ch.writeAndFlush(str);
// ch.writeAndFlush(str+ "\r\n");
System.out.println("客户端发送数据:"+str);
}
}
客户端业务逻辑处理
简单的在userEventTriggered 做了下相应的逻辑处理。
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import java.util.Date;
/**
*
* Title: NettyClientHandler
* Description: 客户端业务逻辑实现
*/
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
/** 客户端请求的心跳命令 */
private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("hb_request",
CharsetUtil.UTF_8));
/** 空闲次数 */
private int idle_count = 1;
/** 发送次数 */
private int count = 1;
/**循环次数 */
private int fcount = 1;
/**
* 建立连接时
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("建立连接时："+new Date());
ctx.fireChannelActive();
}
/**
* 关闭连接时
*/
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("关闭连接时："+new Date());
}
/**
* 心跳请求处理
* 每4秒发送一次心跳请求;
*
*/
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
System.out.println("循环请求的时间："+new Date()+"，次数"+fcount);
if (obj instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) obj;
if (IdleState.WRITER_IDLE.equals(event.state())) { //如果写通道处于空闲状态,就发送心跳命令
if(idle_count <= 3){ //设置发送次数
idle_count++;
ctx.channel().writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
}else{
System.out.println("不再发送心跳请求了！");
}
fcount++;
}
}
}
/**
* 业务逻辑处理
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("第"+count+"次"+",客户端接受的消息:"+msg);
count++;
}
}
客户端过滤器
几乎和服务端一致，除了心跳相关设置。
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;
/**
*
* Title: NettyClientFilter
* Description: Netty客户端 过滤器
*/
public class NettyClientFilter extends ChannelInitializer<SocketChannel> {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline ph = ch.pipeline();
/*
* 解码和编码，应和服务端一致
* */
// ph.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
//入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式
//因为服务端设置的超时时间是5秒，所以设置4秒
ph.addLast( new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));
ph.addLast("decoder", new StringDecoder());
ph.addLast("encoder", new StringEncoder());
ph.addLast("handler", new NettyClientHandler()); //客户端的逻辑
}
}
结果
服务端启动成功,端口是:9876
第1次,服务端接受的消息:Hello Netty
第2次,服务端接受的消息:hb_request
第3次,服务端接受的消息:hb_request
第4次,服务端接受的消息:hb_request
已经5秒没有接收到客户端的信息了
已经5秒没有接收到客户端的信息了
已经5秒没有接收到客户端的信息了
关闭这个不活跃的channel
客户端成功启动...
客户端发送数据:Hello Netty
建立连接时：Sun Nov 10 15:43:16 CST 2019
循环请求的时间：Sun Nov 10 15:43:20 CST 2019，次数1
第1次,客户端接受的消息:服务端成功收到心跳信息
循环请求的时间：Sun Nov 10 15:43:24 CST 2019，次数2
第2次,客户端接受的消息:服务端成功收到心跳信息
循环请求的时间：Sun Nov 10 15:43:28 CST 2019，次数3
第3次,客户端接受的消息:服务端成功收到心跳信息
循环请求的时间：Sun Nov 10 15:43:32 CST 2019，次数4
不再发送心跳请求了！
循环请求的时间：Sun Nov 10 15:43:36 CST 2019，次数5
不再发送心跳请求了！
循环请求的时间：Sun Nov 10 15:43:40 CST 2019，次数6
不再发送心跳请求了！
关闭连接时：Sun Nov 10 15:43:43 CST 2019
