Netty心跳
所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.在 Netty 中, 实现心跳机制的关键是IdleStateHandler, 看下它的构造器：
public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}
这里解释下三个参数的含义：
readerIdleTimeSeconds: 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLE 的IdleStateEvent 事件.
writerIdleTimeSeconds: 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE 的IdleStateEvent 事件.
allIdleTimeSeconds: 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个 ALL_IDLE 的 IdleStateEvent 事件.
注：这三个参数默认的时间单位是秒。若需要指定其他时间单位，可以使用另一个构造方法：
IdleStateHandler(boolean observeOutput, long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit)
Netty 服务端添加心跳
要实现Netty服务端心跳检测机制需要在服务器端的ChannelInitializer中加入如下的代码：
pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));
IdleStateHandler源码
初步地看下IdleStateHandler源码，先看下IdleStateHandler中的channelRead方法：
红框代码其实表示该方法只是进行了透传，不做任何业务逻辑处理，让channelPipe中的下一个handler处理channelRead方法我们再看看channelActive方法：
这里有个initialize的方法，这是IdleStateHandler的精髓，接着探究
这边会触发一个Task，ReaderIdleTimeoutTask，这个task里的run方法源码是这样的：
第一个红框代码是用当前时间减去最后一次channelRead方法调用的时间，假如这个结果是6s，说明最后一次调用channelRead已经是6s之前的事情了，你设置的是5s，那么nextDelay则为-1，说明超时了，那么第二个红框代码则会触发下一个handler的userEventTriggered方法：
如果没有超时则不触发userEventTriggered方法。
实现代码
服务端代码
package com.jony.netty.heartbeat;
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.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;
public class HeartBeatServer {
public static void main(String[] args) throws Exception {
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss, worker)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
//IdleStateHandler的readerIdleTime参数指定超过3秒还没收到客户端的连接，
//会触发IdleStateEvent事件并且交给下一个handler处理，下一个handler必须
//实现userEventTriggered方法处理对应事件
pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));
pipeline.addLast(new HeartBeatServerHandler());
}
});
System.out.println("netty server start。。");
ChannelFuture future = bootstrap.bind(9000).sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
}
客户端代码
package com.jony.netty.heartbeat;
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.Random;
public class HeartBeatClient {
public static void main(String[] args) throws Exception {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new HeartBeatClientHandler());
}
});
System.out.println("netty client start。。");
Channel channel = bootstrap.connect("127.0.0.1", 9000).sync().channel();
String text = "Heartbeat Packet";
Random random = new Random();
while (channel.isActive()) {
int num = random.nextInt(10);
Thread.sleep(num * 1000);
channel.writeAndFlush(text);
}
} catch (Exception e) {
e.printStackTrace();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(" client received :" + msg);
if (msg != null && msg.equals("idle close")) {
System.out.println(" 服务端关闭连接，客户端也关闭");
ctx.channel().closeFuture();
}
}
}
}
为了能让服务端心跳监测异常，服务端设置3s监测一次心跳，那么我们客户端做如下处理，随机产生一个随机码，就可以产生心跳异常了。
Random random = new Random();
while (channel.isActive()) {
int num = random.nextInt(10);
Thread.sleep(num * 1000);
channel.writeAndFlush(text);
}
消息收发处理器
package com.jony.netty.heartbeat;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> {
int readIdleTimes = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
System.out.println(" ====== > [server] message received : " + s);
if ("Heartbeat Packet".equals(s)) {
ctx.channel().writeAndFlush("ok");
} else {
System.out.println(" 其他信息处理 ... ");
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent) evt;
String eventType = null;
switch (event.state()) {
case READER_IDLE:
eventType = "读空闲";
readIdleTimes++; // 读空闲的计数加1
break;
case WRITER_IDLE:
eventType = "写空闲";
// 不处理
break;
case ALL_IDLE:
eventType = "读写空闲";
// 不处理
break;
}
System.out.println(ctx.channel().remoteAddress() + "超时事件：" + eventType);
if (readIdleTimes > 3) {
System.out.println(" [server]读空闲超过3次，关闭连接，释放更多资源");
ctx.channel().writeAndFlush("idle close");
ctx.channel().close();
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
}
}
如果心跳检测异常，则关闭客户端的管道连接。
输出结果
服务端
====== > [server] message received : Heartbeat Packet
/127.0.0.1:61306超时事件：读空闲
====== > [server] message received : Heartbeat Packet
====== > [server] message received : Heartbeat Packet
/127.0.0.1:61306超时事件：读空闲
====== > [server] message received : Heartbeat Packet
/127.0.0.1:61306超时事件：读空闲
/127.0.0.1:61306超时事件：读空闲
[server]读空闲超过3次，关闭连接，释放更多资源
客户端
client received :ok
client received :ok
client received :ok
client received :ok
client received :idle close
服务端关闭连接，客户端也关闭
通过以上输出信息，可以看到只要超过3此读空闲，服务就会自动关闭。
Netty零拷贝
Netty的接收和发送ByteBuffer采用DIRECT BUFFERS，使用堆外直接内存进行Socket读写，不需要进行字节缓冲区的二次拷贝。如果使用传统的JVM堆内存（HEAP BUFFERS）进行Socket读写，JVM会将堆内存Buffer拷贝一份到直接内存中，然后才能写入Socket中。JVM堆内存的数据是不能直接写入Socket中的。相比于堆外直接内存，消息在发送过程中多了一次缓冲区的内存拷贝。可以看下netty的读写源码，比如read源码NioByteUnsafe.read()
直接内存
直接内存（Direct Memory）并不是虚拟机运行时数据区的一部分，也不是Java虚拟机规范中定义的内存区域，某些情况下这部分内存也会被频繁地使用，而且也可能导致OutOfMemoryError异常出现。JavaDirectByteBuffer可以分配一块直接内存(堆外内存)，元空间对应的内存也叫作直接内存，它们对应的都是机器的物理内存。
注意：此时只有DirectByteBuffer对象在jvm内存中，它里面的数据在堆外内存，也就是是直接内存中。
直接内存和堆内存的区别
package com.jony.netty.directbuffer;
import java.nio.ByteBuffer;
/**
* 直接内存与堆内存的区别
*/
public class DirectMemoryTest {
public static void heapAccess() {
long startTime = System.currentTimeMillis();
//分配堆内存
ByteBuffer buffer = ByteBuffer.allocate(1000);
for (int i = 0; i < 100000; i++) {
for (int j = 0; j < 200; j++) {
buffer.putInt(j);
}
buffer.flip();
for (int j = 0; j < 200; j++) {
buffer.getInt();
}
buffer.clear();
}
long endTime = System.currentTimeMillis();
System.out.println("堆内存访问:" + (endTime - startTime));
}
public static void directAccess() {
long startTime = System.currentTimeMillis();
//分配直接内存
ByteBuffer buffer = ByteBuffer.allocateDirect(1000);
for (int i = 0; i < 100000; i++) {
for (int j = 0; j < 200; j++) {
buffer.putInt(j);
}
buffer.flip();
for (int j = 0; j < 200; j++) {
buffer.getInt();
}
buffer.clear();
}
long endTime = System.currentTimeMillis();
System.out.println("直接内存访问:" + (endTime - startTime));
}
public static void heapAllocate() {
long startTime = System.currentTimeMillis();
for (int i = 0; i < 100000; i++) {
ByteBuffer.allocate(100);
}
long endTime = System.currentTimeMillis();
System.out.println("堆内存申请:" + (endTime - startTime));
}
public static void directAllocate() {
long startTime = System.currentTimeMillis();
for (int i = 0; i < 100000; i++) {
ByteBuffer.allocateDirect(100);
}
long endTime = System.currentTimeMillis();
System.out.println("直接内存申请:" + (endTime - startTime));
}
public static void main(String args[]) {
for (int i = 0; i < 10; i++) {
heapAccess();
directAccess();
}
System.out.println();
for (int i = 0; i < 10; i++) {
heapAllocate();
directAllocate();
}
}
}
输出结果
堆内存访问:101
直接内存访问:63
堆内存访问:75
直接内存访问:46
堆内存访问:47
直接内存访问:67
堆内存访问:135
直接内存访问:43
堆内存访问:117
直接内存访问:54
堆内存访问:70
直接内存访问:52
堆内存访问:74
直接内存访问:39
堆内存访问:56
直接内存访问:40
堆内存访问:56
直接内存访问:36
堆内存访问:58
直接内存访问:39
堆内存申请:15
直接内存申请:50
堆内存申请:11
直接内存申请:41
堆内存申请:106
直接内存申请:57
堆内存申请:2
直接内存申请:30
堆内存申请:2
直接内存申请:112
堆内存申请:2
直接内存申请:31
堆内存申请:2
直接内存申请:25
堆内存申请:3
直接内存申请:27
堆内存申请:7
直接内存申请:30
堆内存申请:6
直接内存申请:185
可以看出
从程序运行结果看出直接内存申请较慢，但访问效率高。在java虚拟机实现上，本地IO会直接操作直接内存（直接内存=>系统调用=>硬盘/网卡），而非直接内存则需要二次拷贝（堆内存=>直接内存=>系统调用=>硬盘/网卡）
直接内存分配源码分析：
public static ByteBuffer allocateDirect(int capacity) {
return new DirectByteBuffer(capacity);3 }
DirectByteBuffer(int cap) { // package‐private
super(‐1, 0, cap, cap);
boolean pa = VM.isDirectMemoryPageAligned();
int ps = Bits.pageSize();
long size = Math.max(1L, (long)cap + (pa ? ps : 0));
//判断是否有足够的直接内存空间分配，可通过‐XX:MaxDirectMemorySize=<size>参数指定直接内存最大可分配空间，如果不指定默认为最
大堆内存大小，
//在分配直接内存时如果发现空间不够会显示调用System.gc()触发一次full gc回收掉一部分无用的直接内存的引用对象，同时直接内存也会
被释放掉
//如果释放完分配空间还是不够会抛出异常java.lang.OutOfMemoryError
Bits.reserveMemory(size, cap);
long base = 0;
try {
// 调用unsafe本地方法分配直接内存
base = unsafe.allocateMemory(size);
} catch (OutOfMemoryError x) {
/ 分配失败，释放内存
Bits.unreserveMemory(size, cap);
throw x;
}
unsafe.setMemory(base, size, (byte) 0);
if (pa && (base % ps != 0)) {
// Round up to page boundary
address = base + ps ‐ (base & (ps ‐ 1));
} else {
address = base;
}
// 使用Cleaner机制注册内存回收处理函数，当直接内存引用对象被GC清理掉时，
// 会提前调用这里注册的释放直接内存的Deallocator线程对象的run方法
cleaner = Cleaner.create(this, new Deallocator(base, size, cap));
}
// 申请一块本地内存。内存空间是未初始化的，其内容是无法预期的。
// 使用freeMemory释放内存，使用reallocateMemory修改内存大小
public native long allocateMemory(long bytes);
// openjdk8/hotspot/src/share/vm/prims/unsafe.cpp
UNSAFE_ENTRY(jlong, Unsafe_AllocateMemory(JNIEnv *env, jobject unsafe, jlong size))
UnsafeWrapper("Unsafe_AllocateMemory");
size_t sz = (size_t)size;
if (sz != (julong)size || size < 0) {
THROW_0(vmSymbols::java_lang_IllegalArgumentException());
}
if (sz == 0) {
return 0;
}
54 sz = round_to(sz, HeapWordSize);
// 调用os::malloc申请内存，内部使用malloc这个C标准库的函数申请内存
void* x = os::malloc(sz, mtInternal);
if (x == NULL) {
THROW_0(vmSymbols::java_lang_OutOfMemoryError());
}
//Copy::fill_to_words((HeapWord*)x, sz / HeapWordSize);
return addr_to_java(x);62 UNSAFE_END
使用直接内存的优缺点：
优点：
不占用堆内存空间，减少了发生GC的可能
java虚拟机实现上，本地IO会直接操作直接内存（直接内存=>系统调用=>硬盘/网卡），而非直接内存则需要二次拷贝（堆内存=>直接内存=>系统调用=>硬盘/网卡）
缺点：
初始分配较慢
没有JVM直接帮助管理内存，容易发生内存溢出。为了避免一直没有FULL GC，最终导致直接内存把物理内存被耗完。我们可以指定直接内存的最大值，通过-XX：MaxDirectMemorySize来指定，当达到阈值的时候，调用system.gc来进行一次FULL GC，间接把那些没有被使用的直接内存回收掉。