package com.cw.netty.high.protocol.bean;import java.util.HashMap;
import java.util.Map;/*** 消息头header定义** @author chenwei* @create 2018-07-02 11:20**/public class Header {private int crcCode=0xabef0101;private int length;//消息长度private long sessionID;//会话IDprivate byte type;//消息类型private byte priority;//消息优先级private Map<String, Object> attachment = new HashMap<String, Object>();//附件public int getCrcCode() {return crcCode;}public void setCrcCode(int crcCode) {this.crcCode = crcCode;}public int getLength() {return length;}public void setLength(int length) {this.length = length;}public long getSessionID() {return sessionID;}public void setSessionID(long sessionID) {this.sessionID = sessionID;}public byte getType() {return type;}public void setType(byte type) {this.type = type;}public byte getPriority() {return priority;}public void setPriority(byte priority) {this.priority = priority;}public Map<String, Object> getAttachment() {return attachment;}public void setAttachment(Map<String, Object> attachment) {this.attachment = attachment;}@Overridepublic String toString() {return "Header{" +"crcCode=" + crcCode +", length=" + length +", sessionID=" + sessionID +", type=" + type +", priority=" + priority +", attachment=" + attachment +'}';}
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;/*** netty消息类定义** @author chenwei* @create 2018-07-02 11:19**/public class NettyMessage {private Header header;//消息头private Object body;//消息体public Header getHeader() {return header;}public void setHeader(Header header) {this.header = header;}public Object getBody() {return body;}public void setBody(Object body) {this.body = body;}@Overridepublic String toString() {return "NettyMessage{" +"header=" + header +", body=" + body +'}';}
}

package com.cw.netty.high.protocol;/*** @author chenwei* @create 2018-07-02 17:05**/public class MessageType {//    public static final byte LOGIN_R2=0;//业务请求消息
//    public static final byte LOGIN_1=1;//业务响应消息
//    public static final byte LOGIN_2=2;//业务ONE WAY 消息public static final byte LOGIN_REQ=3;//握手请求消息public static final byte LOGIN_RESP=4;//握手应答消息public static final byte HEARTBEAT_REQ=5;//心跳请求消息public static final byte HEARTBEAT_RESP=6;//心跳应答消息
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.MarshallingCodecFactory;
import com.cw.netty.high.protocol.bean.NettyMarshallingEncoder;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageEncoder;import java.util.List;
import java.util.Map;/*** 消息编码器** @author chenwei* @create 2018-07-02 11:32**/public class NettyMessageEncoder extends MessageToMessageEncoder<NettyMessage> {private NettyMarshallingEncoder marshallingEncoder;public NettyMessageEncoder(){marshallingEncoder= MarshallingCodecFactory.buildMarshallingEncoder();}@Overrideprotected void encode(ChannelHandlerContext channelHandlerContext, NettyMessage nettyMessage, List<Object> out)  throws Exception{if (nettyMessage == null || nettyMessage.getHeader()==null) {throw new Exception("The encode message is null");}ByteBuf buffer = Unpooled.buffer();System.out.println("开始编码："+nettyMessage);//按顺利编码后，根据定义的字段数据类型写入ByteBuf,解码时也要按顺序挨个取出
        buffer.writeInt(nettyMessage.getHeader().getCrcCode());buffer.writeInt(nettyMessage.getHeader().getLength());buffer.writeLong(nettyMessage.getHeader().getSessionID());buffer.writeByte(nettyMessage.getHeader().getType());buffer.writeByte(nettyMessage.getHeader().getPriority());buffer.writeInt(nettyMessage.getHeader().getAttachment().size());String key=null;Object value=null;byte[] keyArray=null;//针对header中的附件编码for (Map.Entry<String,Object> param : nettyMessage.getHeader().getAttachment().entrySet()) {key=param.getKey();keyArray= key.getBytes("UTF-8");value= param.getValue();buffer.writeInt(keyArray.length);buffer.writeBytes(keyArray);marshallingEncoder.encode(channelHandlerContext,value,buffer);}if (nettyMessage.getBody() != null) {//使用MarshallingEncoder编码消息体
            marshallingEncoder.encode(channelHandlerContext,nettyMessage.getBody(),buffer);}else {//没有消息体的话，就赋予0值buffer.writeInt(0);}//更新消息长度字段的值，至于为什么-8，是因为8是长度字段后的偏移量，LengthFieldBasedFrameDecoder的源码中//对长度字段和长度的偏移量之和做了判断，如果不-8，会导致LengthFieldBasedFrameDecoder解码返回null//这是 《Netty权威指南》中的写错的地方buffer.setInt(4, buffer.readableBytes()-8);//书中此处没有add，也即没有将ByteBuf加入到List中，也就没有消息进行编码了，所以导致运行了没有效果……
        out.add(buffer);}}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;
import com.cw.netty.high.protocol.bean.MarshallingCodecFactory;
import com.cw.netty.high.protocol.bean.NettyMarshallingDecoder;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;import java.util.HashMap;
import java.util.Map;/*** Netty消息解码类* @author chenwei* @create 2018-07-02 16:31** 继承LengthFieldBasedFrameDecoder是为了更好了使用它对tcp的粘包和半包处理，* 只需要给我表示消息长度的字段偏移量和消息长度自身所占的字节数，该解码器就能* 自动实现对半包的处理，调用父类LengthFieldBasedFrameDecoder的decode方法后，* 返回的就是整包消息或者为null,**/public class NettyMessageDecoder extends LengthFieldBasedFrameDecoder {NettyMarshallingDecoder marshallingDecoder;public NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {super(maxFrameLength, lengthFieldOffset, lengthFieldLength);marshallingDecoder = MarshallingCodecFactory.buildMarshallingDecoder();}@Overrideprotected Object decode(ChannelHandlerContext ctx, ByteBuf in1) throws Exception {//父类解码后的消息，后续就针对处理后的消息体进行解码，这也是//《Netty权威指南》的另一处书写错误，书中仍对原ByteBuf进行读取//由于父类decode后，读指针已经到达了消息头总长度处，此后再对原消息//进行读取后报处下标越界的异常ByteBuf frame=(ByteBuf) super.decode(ctx, in1);if (frame == null) {return null;}NettyMessage message =new NettyMessage();Header header=new Header();header.setCrcCode(frame.readInt());header.setLength(frame.readInt());header.setSessionID(frame.readLong());header.setType(frame.readByte());header.setPriority(frame.readByte());int size= frame.readInt();if (size > 0) {int keySize=0;byte[] keyArray=null;String key=null;Map<String, Object> attch = new HashMap<String, Object>();for (int i = 0; i <size ; i++) {keySize= frame.readInt();keyArray = new byte[keySize];frame.readBytes(keyArray);key = new String(keyArray, "UTF-8");attch.put(key, marshallingDecoder.decode(ctx,frame));}header.setAttachment(attch);}//readableBytes即为判断剩余可读取的字节数（ this.writerIndex - this.readerIndex）//大于4说明有消息体（无消息体时readableBytes=4），故进行解码if (frame.readableBytes() > 4) {message.setBody(marshallingDecoder.decode(ctx, frame));}message.setHeader(header);return message;}
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;/*** @author chenwei* @create 2018-07-02 17:02**/public class LoginAuthReqHandler extends ChannelHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//建立连接后，发送认证消息NettyMessage message=buildLoginReq();System.out.println("client 发送 认证消息：message="+message);ctx.writeAndFlush(message);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message=(NettyMessage)msg;//若是握手应答消息，判断是否认证成功if (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_RESP) {byte loginResult= (byte)message.getBody();if (loginResult != 0) {//握手失败，关闭连接
                ctx.close();}else {System.out.println("login is ok :"+message);ctx.fireChannelRead(msg);}}else{ctx.fireChannelRead(msg);}}private NettyMessage buildLoginReq(){NettyMessage message=new NettyMessage();Header header=new Header();header.setType((byte) MessageType.LOGIN_REQ);message.setHeader(header);return message;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.fireExceptionCaught(cause);}
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;/*** @author chenwei* @create 2018-07-02 17:19**/public class LoginAuthRespHandler extends ChannelHandlerAdapter {private Map<String, Boolean> nodeCheck = new ConcurrentHashMap<>();private String [] whiteList={"127.0.0.1","10.155.33.113"};@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message=(NettyMessage)msg;//若为握手认证消息，则校验并返回响应，否则传递到下一个handlerif (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_REQ) {String nodeIndex= ctx.channel().remoteAddress().toString();NettyMessage loginResp=null;if (nodeCheck.containsKey(nodeIndex)) {//重复登陆，拒绝loginResp=buildResponse((byte)-1);System.out.println("重复登陆，拒绝 ：ip="+nodeIndex);}else{boolean isOk=true;InetSocketAddress address=(InetSocketAddress) ctx.channel().remoteAddress();String ip=  address.getAddress().getHostAddress();for (String wip:whiteList){if (wip.equals(ip)) {isOk=true;nodeCheck.put(ip, true);System.out.println("通过白名单检测 ip="+ip);break;}}loginResp=isOk?buildResponse((byte) 0):buildResponse((byte)-1);}ctx.writeAndFlush(loginResp);}else {ctx.fireChannelRead(msg);}}private NettyMessage buildResponse(byte result){NettyMessage message=new NettyMessage();Header header=new Header();header.setType(MessageType.LOGIN_RESP);message.setHeader(header);message.setBody(result);return message;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {nodeCheck.remove(ctx.channel().remoteAddress().toString());ctx.close();ctx.fireExceptionCaught(cause);}
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;/*** 客户端心跳检测* @author chenwei* @create 2018-07-02 17:36**/public class HeartBeatReqHandler extends ChannelHandlerAdapter {private volatile ScheduledFuture heartBeaet;@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message= (NettyMessage)msg;//认证成功后，定时发送心跳检测if (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_RESP) {heartBeaet=ctx.executor().scheduleAtFixedRate(new HeartBeatTask(ctx), 0, 5000, TimeUnit.MILLISECONDS);} else if (message.getHeader() != null && message.getHeader().getType() == MessageType.HEARTBEAT_RESP) {System.out.println("client receive server heart beat message :-->"+message);}else {ctx.fireChannelRead(msg);}}private class HeartBeatTask implements Runnable{ChannelHandlerContext ctx;public HeartBeatTask(ChannelHandlerContext ctx) {this.ctx=ctx;}@Overridepublic void run() {NettyMessage nettyMessage=buildHeatBeat();System.out.println("client send heart beat message to server :--->"+nettyMessage);ctx.writeAndFlush(nettyMessage);}}//心跳检测仅消息头就够了private NettyMessage buildHeatBeat(){NettyMessage message=new NettyMessage();Header header=new Header();header.setType(MessageType.HEARTBEAT_REQ);message.setHeader(header);return message;}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {if (heartBeaet != null) {heartBeaet.cancel(true);heartBeaet=null;}ctx.fireExceptionCaught(cause);}
}

package com.cw.netty.high.protocol;import com.cw.netty.high.protocol.bean.Header;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;/*** 服务器端心跳handler* @author chenwei* @create 2018-07-02 18:00**/public class HeartBeatRespHandler extends ChannelHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {NettyMessage message=(NettyMessage)msg;//收到心跳消息后，构造心跳应答消息返回if (message.getHeader() != null && message.getHeader().getType() == MessageType.HEARTBEAT_REQ) {System.out.println("receive client heart beat message :-->");NettyMessage heartBeat=buildHeatBeat();System.out.println("send heart beat response message to client:-->"+heartBeat);ctx.writeAndFlush(heartBeat);}else {ctx.fireChannelRead(msg);}}private NettyMessage buildHeatBeat(){NettyMessage message=new NettyMessage();Header header=new Header();header.setType(MessageType.HEARTBEAT_RESP);message.setHeader(header);return message;}}

package com.cw.netty.high.protocol.bean;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.marshalling.MarshallingDecoder;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;/*** @author chenwei* @create 2018-07-02 16:12**/public class NettyMarshallingDecoder extends MarshallingDecoder {public NettyMarshallingDecoder(UnmarshallerProvider provider) {super(provider);}public NettyMarshallingDecoder(UnmarshallerProvider provider, int maxObjectSize) {super(provider, maxObjectSize);}public Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception{return super.decode(ctx,in);}
}

package com.cw.netty.high.protocol.bean;import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallingEncoder;/*** @author chenwei* @create 2018-07-02 16:15**/public class NettyMarshallingEncoder extends MarshallingEncoder {public NettyMarshallingEncoder(MarshallerProvider provider) {super(provider);}@Overridepublic void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {super.encode(ctx, msg, out);}
}

package com.cw.netty.high.protocol.bean;import io.netty.handler.codec.marshalling.DefaultMarshallerProvider;
import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider;
import io.netty.handler.codec.marshalling.MarshallerProvider;
import io.netty.handler.codec.marshalling.UnmarshallerProvider;
import org.jboss.marshalling.MarshallerFactory;
import org.jboss.marshalling.Marshalling;
import org.jboss.marshalling.MarshallingConfiguration;/*** @author chenwei* @create 2018-07-02 14:15**/public class MarshallingCodecFactory {public static NettyMarshallingDecoder buildMarshallingDecoder(){final MarshallerFactory factory= Marshalling.getProvidedMarshallerFactory("serial");final MarshallingConfiguration configuration=new MarshallingConfiguration();configuration.setVersion(5);UnmarshallerProvider provider = new DefaultUnmarshallerProvider(factory, configuration);NettyMarshallingDecoder decoder = new NettyMarshallingDecoder(provider, 1024);return decoder;}public static NettyMarshallingEncoder buildMarshallingEncoder(){final MarshallerFactory factory= Marshalling.getProvidedMarshallerFactory("serial");final MarshallingConfiguration configuration=new MarshallingConfiguration();configuration.setVersion(5);MarshallerProvider provider = new DefaultMarshallerProvider(factory, configuration);NettyMarshallingEncoder encoder = new NettyMarshallingEncoder(provider);return encoder;}}

package com.cw.netty.high.protocol;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.ReadTimeoutHandler;import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;/*** @author chenwei* @create 2018-07-02 18:15**/public class NettyClient {private ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);public void connect(final int port, final String host)throws Exception{try {Bootstrap boot=new Bootstrap();EventLoopGroup group = new NioEventLoopGroup();boot.group(group).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<Channel>() {@Overrideprotected void initChannel(Channel channel) throws Exception {channel.pipeline().addLast(new NettyMessageDecoder(1024*1024,4,4));channel.pipeline().addLast("messageEncoder",new NettyMessageEncoder());//添加超时处理handler，规定时间内没有收到消息则关闭链路channel.pipeline().addLast("readTimeoutHandler", new ReadTimeoutHandler(50));channel.pipeline().addLast("loginAuthHandler", new LoginAuthReqHandler());channel.pipeline().addLast("heartbeatHandler", new HeartBeatReqHandler());}});ChannelFuture future = boot.connect(new InetSocketAddress(host, port),new InetSocketAddress("127.0.0.1",8889)).sync();System.out.println("client is start……");future.channel().closeFuture().sync();}finally {//释放完毕后，清空资源，再次发起重连操作executorService.execute(new Runnable() {@Overridepublic void run() {try {System.out.println("重连……");TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}try {connect(port,host);} catch (Exception e) {e.printStackTrace();}}});}}public static void main(String[] args) throws Exception {new NettyClient().connect(8000,"127.0.0.1");}}

package com.cw.netty.high.protocol;import io.netty.bootstrap.ServerBootstrap;
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;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;/*** @author chenwei* @create 2018-07-02 18:36**/public class NettyServer {public void bind() throws Exception{EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workGroup = new NioEventLoopGroup();ServerBootstrap bootstrap=new ServerBootstrap();bootstrap.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {socketChannel.pipeline().addLast(new NettyMessageDecoder(1024*1024,4,4));socketChannel.pipeline().addLast(new NettyMessageEncoder());socketChannel.pipeline().addLast("readTimeoutHandler",new ReadTimeoutHandler(50));socketChannel.pipeline().addLast(new LoginAuthRespHandler());socketChannel.pipeline().addLast("HeartBeathandler",new HeartBeatRespHandler());}});bootstrap.bind("127.0.0.1",8000).sync();System.out.println("netty server start ok:"+("127.0.0.1"+8000));}public static void main(String[] args) throws Exception {new NettyServer().bind();}}