一，客户端代码
1 package bhz.netty.heartBeat; 2 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; 10 import io.netty.handler.timeout.ReadTimeoutHandler; 11 12 public class Client { 13 14 15 public static void main(String[] args) throws Exception{ 16 17 EventLoopGroup group = new NioEventLoopGroup(); 18 Bootstrap b = new Bootstrap(); 19 b.group(group) 20 .channel(NioSocketChannel.class) 21 .handler(new ChannelInitializer<SocketChannel>() { 22 @Override 23 protected void initChannel(SocketChannel sc) throws Exception { 24 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); 25 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); 26 sc.pipeline().addLast(new ReadTimeoutHandler(30)); 27 sc.pipeline().addLast(new ClienHeartBeattHandler()); 28 } 29 }); 30 31 ChannelFuture cf = b.connect("10.13.82.18", 8888).sync(); 32 33 cf.channel().closeFuture().sync(); 34 group.shutdownGracefully(); 35 } 36 }
二，客户端助手类代码
1 package bhz.netty.heartBeat; 2 3 import java.net.InetAddress; 4 import java.util.HashMap; 5 import java.util.concurrent.Executors; 6 import java.util.concurrent.ScheduledExecutorService; 7 import java.util.concurrent.ScheduledFuture; 8 import java.util.concurrent.TimeUnit; 9 10 import org.hyperic.sigar.CpuPerc; 11 import org.hyperic.sigar.Mem; 12 import org.hyperic.sigar.Sigar; 13 14 import io.netty.channel.ChannelHandlerContext; 15 import io.netty.channel.ChannelInboundHandlerAdapter; 16 import io.netty.util.ReferenceCountUtil; 17 18 19 public class ClienHeartBeattHandler extends ChannelInboundHandlerAdapter { 20 21 private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); 22 23 private ScheduledFuture<?> heartBeat; 24 //主动向服务器发送认证信息 25 private InetAddress addr ; 26 27 private static final String SUCCESS_KEY = "auth_success_key"; 28 29 @Override 30 public void channelActive(ChannelHandlerContext ctx) throws Exception { 31 addr = InetAddress.getLocalHost(); 32 String ip = addr.getHostAddress(); 33 System.err.println("ip:" + ip); 34 String key = "1234"; 35 String auth = ip + "," + key; 36 ctx.writeAndFlush(auth); 37 } 38 39 @Override 40 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 41 try { 42 if(msg instanceof String){ 43 String ret = (String)msg; 44 if(SUCCESS_KEY.equals(ret)){ 45 // 握手成功，主动发送心跳消息 46 this.heartBeat = this.scheduler.scheduleWithFixedDelay(new HeartBeatTask(ctx), 0, 8, TimeUnit.SECONDS); 47 System.out.println(msg); 48 } 49 else { 50 System.out.println(msg); 51 } 52 } 53 } finally { 54 ReferenceCountUtil.release(msg); 55 } 56 } 57 58 private class HeartBeatTask implements Runnable { 59 private final ChannelHandlerContext ctx; 60 61 public HeartBeatTask(final ChannelHandlerContext ctx) { 62 this.ctx = ctx; 63 } 64 65 @Override 66 public void run() { 67 try { 68 RequestInfo info = new RequestInfo(); 69 //ip 70 info.setIp(addr.getHostAddress()); 71 Sigar sigar = new Sigar(); 72 //cpu prec 73 CpuPerc cpuPerc = sigar.getCpuPerc(); 74 HashMap<String, Object> cpuPercMap = new HashMap<String, Object>(); 75 cpuPercMap.put("combined", cpuPerc.getCombined()); 76 cpuPercMap.put("user", cpuPerc.getUser()); 77 cpuPercMap.put("sys", cpuPerc.getSys()); 78 cpuPercMap.put("wait", cpuPerc.getWait()); 79 cpuPercMap.put("idle", cpuPerc.getIdle()); 80 // memory 81 Mem mem = sigar.getMem(); 82 HashMap<String, Object> memoryMap = new HashMap<String, Object>(); 83 memoryMap.put("total", mem.getTotal() / 1024L); 84 memoryMap.put("used", mem.getUsed() / 1024L); 85 memoryMap.put("free", mem.getFree() / 1024L); 86 info.setCpuPercMap(cpuPercMap); 87 info.setMemoryMap(memoryMap); 88 ctx.writeAndFlush(info); 89 90 } catch (Exception e) { 91 e.printStackTrace(); 92 } 93 } 94 95 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 96 cause.printStackTrace(); 97 if (heartBeat != null) { 98 heartBeat.cancel(true); 99 heartBeat = null; 100 } 101 ctx.fireExceptionCaught(cause); 102 } 103 104 } 105 }
1，连接成功后会先执行客户端的public void channelActive(ChannelHandlerContext ctx) throws Exception {方法注意这个时候还没有去执行服务器端的代码。
所以先打印出来ip:10.13.82.18这行代码。
然后现在去执行服务器端的代码
3，第40行代码这时候会返回auth_success_key，
4，第46行代码会执行，由于是异步的所以会先走第47行代码，所以客户端会打印出来auth_success_key这个字段
6，这时候服务端返回的会执行第50行代码所以打印出来info received!
7，由于一直处于一个连接的状态而且使用的是心跳连接每过5秒就会连接一次，这时候由于是一直连接的状态，所以就会一直返回info received!
三，封装的实体类对象
1 package bhz.netty.heartBeat; 2 3 import java.io.Serializable; 4 import java.util.HashMap; 5 6 public class RequestInfo implements Serializable { 7 8 private String ip ; 9 private HashMap<String, Object> cpuPercMap ; 10 private HashMap<String, Object> memoryMap; 11 //.. other field 12 13 public String getIp() { 14 return ip; 15 } 16 public void setIp(String ip) { 17 this.ip = ip; 18 } 19 public HashMap<String, Object> getCpuPercMap() { 20 return cpuPercMap; 21 } 22 public void setCpuPercMap(HashMap<String, Object> cpuPercMap) { 23 this.cpuPercMap = cpuPercMap; 24 } 25 public HashMap<String, Object> getMemoryMap() { 26 return memoryMap; 27 } 28 public void setMemoryMap(HashMap<String, Object> memoryMap) { 29 this.memoryMap = memoryMap; 30 } 31 32 33 }
四，编解码工具类
1 package bhz.netty.heartBeat; 2 3 import io.netty.handler.codec.marshalling.DefaultMarshallerProvider; 4 import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider; 5 import io.netty.handler.codec.marshalling.MarshallerProvider; 6 import io.netty.handler.codec.marshalling.MarshallingDecoder; 7 import io.netty.handler.codec.marshalling.MarshallingEncoder; 8 import io.netty.handler.codec.marshalling.UnmarshallerProvider; 9 10 import org.jboss.marshalling.MarshallerFactory; 11 import org.jboss.marshalling.Marshalling; 12 import org.jboss.marshalling.MarshallingConfiguration; 13 14 /** 15 * Marshalling工厂 16 * @author（alienware） 17 * @since 2014-12-16 18 */ 19 public final class MarshallingCodeCFactory { 20 21 /** 22 * 创建Jboss Marshalling解码器MarshallingDecoder 23 * @return MarshallingDecoder 24 */ 25 public static MarshallingDecoder buildMarshallingDecoder() { 26 //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。 27 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); 28 //创建了MarshallingConfiguration对象，配置了版本号为5 29 final MarshallingConfiguration configuration = new MarshallingConfiguration(); 30 configuration.setVersion(5); 31 //根据marshallerFactory和configuration创建provider 32 UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration); 33 //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度 34 MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1); 35 return decoder; 36 } 37 38 /** 39 * 创建Jboss Marshalling编码器MarshallingEncoder 40 * @return MarshallingEncoder 41 */ 42 public static MarshallingEncoder buildMarshallingEncoder() { 43 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); 44 final MarshallingConfiguration configuration = new MarshallingConfiguration(); 45 configuration.setVersion(5); 46 MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration); 47 //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组 48 MarshallingEncoder encoder = new MarshallingEncoder(provider); 49 return encoder; 50 } 51 }
五，服务端代码
1 package bhz.netty.heartBeat; 2 3 import io.netty.bootstrap.ServerBootstrap; 4 import io.netty.channel.ChannelFuture; 5 import io.netty.channel.ChannelInitializer; 6 import io.netty.channel.ChannelOption; 7 import io.netty.channel.EventLoopGroup; 8 import io.netty.channel.nio.NioEventLoopGroup; 9 import io.netty.channel.socket.SocketChannel; 10 import io.netty.channel.socket.nio.NioServerSocketChannel; 11 import io.netty.handler.logging.LogLevel; 12 import io.netty.handler.logging.LoggingHandler; 13 import io.netty.handler.timeout.ReadTimeoutHandler; 14 15 public class Server { 16 17 public static void main(String[] args) throws Exception{ 18 19 EventLoopGroup pGroup = new NioEventLoopGroup(); 20 EventLoopGroup cGroup = new NioEventLoopGroup(); 21 22 ServerBootstrap b = new ServerBootstrap(); 23 b.group(pGroup, cGroup) 24 .channel(NioServerSocketChannel.class) 25 .option(ChannelOption.SO_BACKLOG, 1024) 26 //设置日志 27 .handler(new LoggingHandler(LogLevel.INFO)) 28 .childHandler(new ChannelInitializer<SocketChannel>() { 29 protected void initChannel(SocketChannel sc) throws Exception { 30 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); 31 sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); 32 sc.pipeline().addLast(new ReadTimeoutHandler(30)); 33 sc.pipeline().addLast(new ServerHeartBeatHandler()); 34 } 35 }); 36 37 ChannelFuture cf = b.bind(8888).sync(); 38 39 cf.channel().closeFuture().sync(); 40 pGroup.shutdownGracefully(); 41 cGroup.shutdownGracefully(); 42 43 } 44 }
,
六，服务端助手类代码
1 package bhz.netty.heartBeat; 2 3 import java.util.HashMap; 4 5 import io.netty.channel.ChannelFutureListener; 6 import io.netty.channel.ChannelHandlerContext; 7 import io.netty.channel.ChannelInboundHandlerAdapter; 8 9 public class ServerHeartBeatHandler extends ChannelInboundHandlerAdapter { 10 11 /** key:ip value:auth */ 12 private static HashMap<String, String> AUTH_IP_MAP = new HashMap<String, String>(); 13 private static final String SUCCESS_KEY = "auth_success_key"; 14 15 static { 16 AUTH_IP_MAP.put("10.13.82.18", "1234"); 17 } 18 19 private boolean auth(ChannelHandlerContext ctx, Object msg){ 20 //System.out.println(msg); 21 String [] ret = ((String) msg).split(","); 22 String auth = AUTH_IP_MAP.get(ret[0]); 23 if(auth != null && auth.equals(ret[1])){ 24 ctx.writeAndFlush(SUCCESS_KEY); 25 return true; 26 } else { 27 ctx.writeAndFlush("auth failure !").addListener(ChannelFutureListener.CLOSE); 28 return false; 29 } 30 } 31 32 @Override 33 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 34 if(msg instanceof String){ 35 auth(ctx, msg); 36 } else if (msg instanceof RequestInfo) { 37 38 RequestInfo info = (RequestInfo) msg; 39 System.out.println("--------------------------------------------"); 40 System.out.println("当前主机ip为: " + info.getIp()); 41 System.out.println("当前主机cpu情况: "); 42 HashMap<String, Object> cpu = info.getCpuPercMap(); 43 System.out.println("总使用率: " + cpu.get("combined")); 44 System.out.println("用户使用率: " + cpu.get("user")); 45 System.out.println("系统使用率: " + cpu.get("sys")); 46 System.out.println("等待率: " + cpu.get("wait")); 47 System.out.println("空闲率: " + cpu.get("idle")); 48 49 System.out.println("当前主机memory情况: "); 50 HashMap<String, Object> memory = info.getMemoryMap(); 51 System.out.println("内存总量: " + memory.get("total")); 52 System.out.println("当前内存使用量: " + memory.get("used")); 53 System.out.println("当前内存剩余量: " + memory.get("free")); 54 System.out.println("--------------------------------------------"); 55 56 ctx.writeAndFlush("info received!"); 57 } else { 58 ctx.writeAndFlush("connect failure!").addListener(ChannelFutureListener.CLOSE); 59 } 60 } 61 62 63 }
2，第35行代码执行，这时候会返回auth_success_key，所以客户端会打印出来auth_success_key这个字段
5，由46行代码触发的回调函数，会触发38-55行的代码，这时候会返回客户端
七，结果
客户端结果
1 ip:10.13.82.18 2 auth_success_key 3 info received! 4 info received! 5 info received! 6 info received! 7 info received! 8 info received!
服务端结果
1 当前主机ip为: 10.13.82.18 2 当前主机cpu情况: 3 总使用率: 0.15106815869786366 4 用户使用率: 0.1185147507629705 5 系统使用率: 0.03255340793489318 6 等待率: 0.0 7 空闲率: 0.8489318413021363 8 当前主机memory情况: 9 内存总量: 3865836 10 当前内存使用量: 3537528 11 当前内存剩余量: 328308 12 -------------------------------------------- 13 -------------------------------------------- 14 当前主机ip为: 10.13.82.18 15 当前主机cpu情况: 16 总使用率: 0.15106815869786366 17 用户使用率: 0.1190233977619532 18 系统使用率: 0.03204476093591048 19 等待率: 0.0 20 空闲率: 0.8413021363173957 21 当前主机memory情况: 22 内存总量: 3865836 23 当前内存使用量: 3518000 24 当前内存剩余量: 347836 25 -------------------------------------------- 26 -------------------------------------------- 27 当前主机ip为: 10.13.82.18 28 当前主机cpu情况: 29 总使用率: 0.16954164613109907 30 用户使用率: 0.16165598817151305 31 系统使用率: 0.007885657959586003 32 等待率: 0.0 33 空闲率: 0.830458353868901 34 当前主机memory情况: 35 内存总量: 3865836 36 当前内存使用量: 3521124 37 当前内存剩余量: 344712 38 -------------------------------------------- 39 -------------------------------------------- 40 当前主机ip为: 10.13.82.18 41 当前主机cpu情况: 42 总使用率: 0.163265306122449 43 用户使用率: 0.14095870906502136 44 系统使用率: 0.022306597057427623 45 等待率: 0.0 46 空闲率: 0.8367346938775511 47 当前主机memory情况: 48 内存总量: 3865836 49 当前内存使用量: 3495956 50 当前内存剩余量: 369880 51 -------------------------------------------- 52 -------------------------------------------- 53 当前主机ip为: 10.13.82.18 54 当前主机cpu情况: 55 总使用率: 0.16366366366366367 56 用户使用率: 0.14814814814814814 57 系统使用率: 0.015515515515515516 58 等待率: 0.0 59 空闲率: 0.8363363363363363 60 当前主机memory情况: 61 内存总量: 3865836 62 当前内存使用量: 3493664 63 当前内存剩余量: 372172 64 -------------------------------------------- 65 -------------------------------------------- 66 当前主机ip为: 10.13.82.18 67 当前主机cpu情况: 68 总使用率: 0.17497456765005087 69 用户使用率: 0.1515768056968464 70 系统使用率: 0.023397761953204477 71 等待率: 0.0 72 空闲率: 0.8250254323499492 73 当前主机memory情况: 74 内存总量: 3865836 75 当前内存使用量: 3489476 76 当前内存剩余量: 376360 77 -------------------------------------------- 78 -------------------------------------------- 79 当前主机ip为: 10.13.82.18 80 当前主机cpu情况: 81 总使用率: 0.14637752587481517 82 用户使用率: 0.10004928536224741 83 系统使用率: 0.046328240512567766 84 等待率: 0.0 85 空闲率: 0.8457368161655988 86 当前主机memory情况: 87 内存总量: 3865836 88 当前内存使用量: 3472776 89 当前内存剩余量: 393060 90 -------------------------------------------- 91 -------------------------------------------- 92 当前主机ip为: 10.13.82.18 93 当前主机cpu情况: 94 总使用率: 0.22200488997555012 95 用户使用率: 0.1374083129584352 96 系统使用率: 0.08459657701711491 97 等待率: 0.0 98 空闲率: 0.7779951100244499 99 当前主机memory情况: 100 内存总量: 3865836 101 当前内存使用量: 3467608 102 当前内存剩余量: 398228 103 -------------------------------------------- 104 -------------------------------------------- 105 当前主机ip为: 10.13.82.18 106 当前主机cpu情况: 107 总使用率: 0.1416267942583732 108 用户使用率: 0.13444976076555024 109 系统使用率: 0.007177033492822967 110 等待率: 0.0 111 空闲率: 0.8583732057416268 112 当前主机memory情况: 113 内存总量: 3865836 114 当前内存使用量: 3470100 115 当前内存剩余量: 395736
注意红字的顺序及解释
