netty自定义协议

男娘i 2022-06-12 12:45 197阅读 0赞

《netty权威指南》一书中关于自定义协议开发的源码中有一部分错误导致代码无法运行，加了一点改变可以完美运行了，

package nettyAgreement.decoder;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import nettyAgreement.Header;
 import nettyAgreement.MarshallingCodecFactory;
 import nettyAgreement.NettyMessage;
 
 /**
 * 消息解码类
 * @author Gong.YiYang
 * @Date 2017年7月19日
 * @Version 
 * @Description
 */
 public class NettyMessageDecoder extends LengthFieldBasedFrameDecoder {
 	
 	NettyMarshallingDecoder marshallingDecoder;
 	
 	public NettyMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
 		super(maxFrameLength, lengthFieldOffset, lengthFieldLength);
 		// TODO Auto-generated constructor stub
 		marshallingDecoder = MarshallingCodecFactory.buMarshallingDecoder();
 	}
 
 	public NettyMessageDecoder(int maxFrameLength,
 int lengthFieldOffset, int lengthFieldLength,
 int lengthAdjustment, int initialBytesToStrip) {
 		// TODO Auto-generated constructor stub
 		super( maxFrameLength,
 	 lengthFieldOffset, lengthFieldLength,
 	 lengthAdjustment, initialBytesToStrip);
 		marshallingDecoder = MarshallingCodecFactory.buMarshallingDecoder();
 	}
 
 	@Override
 	protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
 		// TODO Auto-generated method stub
 		ByteBuf frame = (ByteBuf)super.decode(ctx, in);
 		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) {
 			Map<String, Object> attch = new HashMap<String,Object>(size);
 			int keySize = 0;
 			byte[] keyArray = null;
 			String key = null;
 			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));
 			}
 			keyArray = null;
 			key = null;
 			header.setAttachment(attch);
 		}
 		if (frame.readableBytes() > 4) {
 			message.setBody(marshallingDecoder.decode(ctx, frame));
 		}
 		message.setHeader(header);
 		return message;
 	}
 }

package nettyAgreement.encoder;
 
 import java.util.List;
 import java.util.Map.Entry;
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.MessageToMessageEncoder;
 import nettyAgreement.MarshallingCodecFactory;
 import nettyAgreement.NettyMessage;
 
 /**
 * 消息编码
 * @author Gong.YiYang
 * @Date 2017年7月19日
 * @Version 
 * @Description
 */
 public class NettyMessageEncoder extends MessageToMessageEncoder<NettyMessage>{
 
 	private NettyMarshallingEncoder marshallingEncoder;
 	
 	public NettyMessageEncoder() {
 		this.marshallingEncoder = MarshallingCodecFactory.buildMarshallingEncoder();
 	}
 
 	@Override
 	public void encode(ChannelHandlerContext ctx, NettyMessage msg, List<Object> out) throws Exception {
 		// TODO Auto-generated method stub
 		if (msg == null || msg.getHeader() == null) {
 			throw new Exception("This encode message is null");
 		}
 		
 		ByteBuf sendBuffer = Unpooled.buffer();
 		sendBuffer.writeInt(msg.getHeader().getCrcCode());
 		sendBuffer.writeInt(msg.getHeader().getLength());
 		sendBuffer.writeLong(msg.getHeader().getSessionID());
 		sendBuffer.writeByte(msg.getHeader().getType());
 		sendBuffer.writeByte(msg.getHeader().getPriority());
 		sendBuffer.writeInt(msg.getHeader().getAttachment().size());
 		
 		String key = null;
 		byte[] keyArray = null;
 		Object value = null;
 		for (Entry<String, Object> param : msg.getHeader().getAttachment().entrySet()) {
 			 key = param.getKey();
 			 keyArray = key.getBytes("UTF-8");
 			 sendBuffer.writeInt(keyArray.length);
 			 sendBuffer.writeBytes(keyArray);
 			 value = param.getValue();
 			 marshallingEncoder.encode(ctx, value, sendBuffer);
 		}
 		
 		key = null;
 		keyArray = null;
 		value = null;
 		if (msg.getBody() != null) {
 			marshallingEncoder.encode(ctx, msg.getBody(), sendBuffer);
 		}else
 			sendBuffer.writeInt(0); 
 			// 在第4个字节出写入Buffer的长度 
 			int readableBytes = sendBuffer.readableBytes(); 
 			sendBuffer.setInt(4, readableBytes); 
 			
 			// 把Message添加到List传递到下一个Handler 
 			out.add(sendBuffer); 
 	}
 	
 }

package nettyAgreement;
 /**
 * 定义消息头
 * @author Gong.YiYang
 * @Date 2017年7月19日
 * @Version 
 * @Description
 */
 
 import java.util.HashMap;
 import java.util.Map;
 
 public final class Header {
 	
 	private int crcCode = 0xabef0101;//长度32，
 	private int length;//消息长度32
 	private long sessionID;//会话ID64
 	private byte type;//消息类型8
 	private byte priority;//消息优先级8
 	private Map<String, Object> attachment = new HashMap<String,Object>();//附件
 	public final int getCrcCode() {
 		return crcCode;
 	}
 	public final void setCrcCode(int crcCode) {
 		this.crcCode = crcCode;
 	}
 	public final int getLength() {
 		return length;
 	}
 	public final void setLength(int length) {
 		this.length = length;
 	}
 	public final long getSessionID() {
 		return sessionID;
 	}
 	public final void setSessionID(long sessionID) {
 		this.sessionID = sessionID;
 	}
 	public final byte getType() {
 		return type;
 	}
 	public final void setType(byte type) {
 		this.type = type;
 	}
 	public final byte getPriority() {
 		return priority;
 	}
 	public final void setPriority(byte priority) {
 		this.priority = priority;
 	}
 	public final Map<String, Object> getAttachment() {
 		return attachment;
 	}
 	public final void setAttachment(Map<String, Object> attachment) {
 		this.attachment = attachment;
 	}
 	@Override
 	public String toString() {
 		return "Header [crcCode=" + crcCode + ", length=" + length + ", sessionID=" + sessionID + ", type=" + type
 				+ ", priority=" + priority + ", attachment=" + attachment + "]";
 	}
 	
 	
 }

package nettyAgreement;
 
 import java.util.concurrent.TimeUnit;
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.util.concurrent.ScheduledFuture;
 import nettyAgreement.Header;
 import nettyAgreement.MessageType;
 import nettyAgreement.NettyMessage;
 
 /**
 * 客户端心跳消息
 * @author Gong.YiYang
 * @Date 2017年7月20日
 * @Version 
 * @Description
 */
 public class HeartBeatReqHandler extends ChannelHandlerAdapter {
 	
 	private volatile ScheduledFuture<?> heartBeat;
 	
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 		// TODO Auto-generated method stub
 		NettyMessage message = (NettyMessage)msg;
 		//握手成功，主动发送心跳消息
 		if (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_RESP.type) {
 			heartBeat = ctx.executor().scheduleAtFixedRate(new HeartBeatReqHandler.HeartBeatTask(ctx), 0, 5000, TimeUnit.MILLISECONDS);
 		}else if (message.getHeader() != null && message.getHeader().getType() == MessageType.HEART_RESP.type) {
 			System.out.println("Client receive server heart beat message : -->"+message);
 		}else 
 			ctx.fireChannelRead(msg);
 	}
 	//心跳任务
 	private class HeartBeatTask implements Runnable{
 		private final ChannelHandlerContext ctx;
 		
 		public HeartBeatTask(final ChannelHandlerContext ctx) {
 			this.ctx = ctx;
 		}
 
 		@Override
 		public void run() {
 			// TODO Auto-generated method stub
 			NettyMessage heartBeat = buildHeartBeat();
 			System.out.println("Client send heart beat message to server : -->"+heartBeat);
 			ctx.writeAndFlush(heartBeat);
 		}
 		
 		private NettyMessage buildHeartBeat(){
 			NettyMessage message = new NettyMessage();
 			Header header = new Header();
 			header.setType(MessageType.HEART_REQ.type);
 			message.setHeader(header);
 			return message;
 		}
 	}
 	
 	
 	@Override
 	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 		// TODO Auto-generated method stub
 		if (heartBeat != null) {
 			heartBeat.cancel(true);
 			heartBeat=null;
 		}
 		ctx.fireExceptionCaught(cause);
 		
 	}
 }

package nettyAgreement;
 
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
 
 /**
 * 服务端心跳消息
 * @author Gong.YiYang
 * @Date 2017年7月20日
 * @Version 
 * @Description
 */
 public class HeartBeatRespHandler extends ChannelHandlerAdapter {
 	
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 		// TODO Auto-generated method stub
 		NettyMessage message = (NettyMessage)msg;
 		if (message.getHeader() != null && message.getHeader().getType() == MessageType.HEART_REQ.type) {
 			System.out.println("Receive client beat message : -->"+message);
 			NettyMessage heartBeat = buildHeartBeat();
 			System.out.println("Send heart beat message to client ： -->"+heartBeat);
 			ctx.writeAndFlush(heartBeat);
 		}else 
 			
 		ctx.fireChannelRead(msg);
 	}
 	
 	private NettyMessage buildHeartBeat(){
 		NettyMessage message = new NettyMessage();
 		Header header = new Header();
 		header.setType(MessageType.HEART_RESP.type);
 		message.setHeader(header);
 		return message;
 	}
 	
 }

package nettyAgreement;
 
 
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
 
 /**
 * 客户端握手认证
 * 
 * @author Gong.YiYang
 * @Date 2017年7月20日
 * @Version
 * @Description
 */
 public class LoginAuthReqHandler extends ChannelHandlerAdapter {
 
 	@Override
 	public void channelActive(ChannelHandlerContext ctx) throws Exception {
 		// TODO Auto-generated method stub
 		ctx.writeAndFlush(buildLoginReq());
 	}
 
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 		// TODO Auto-generated method stub
 		NettyMessage message = (NettyMessage)msg;
 		//如果是握手应答消息，需要判断是否认证成功
 		if (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_RESP.type) {
 			byte loginResult = (byte)message.getBody();
 			if (loginResult != (byte)0) {
 				//握手失败，关闭连接;0表示认证通过，握手成功
 				ctx.close();
 			}else {
 				System.out.println("login is ok :"+message);
 				ctx.fireChannelRead(msg);
 			}
 		}else {
 			
 			ctx.fireChannelRead(msg);
 		}
 		
 	}
 
 	private NettyMessage buildLoginReq() {
 		// TODO Auto-generated method stub
 		NettyMessage message = new NettyMessage();
 		Header header = new Header();
 		header.setType(MessageType.LOGIN_REQ.type);
 		message.setHeader(header);
 		return message;
 	}
 	
 	@Override
 	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 		// TODO Auto-generated method stub
 		ctx.fireExceptionCaught(cause);
 	}
 }

package nettyAgreement;
 
 import java.net.InetSocketAddress;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 import io.netty.channel.ChannelHandlerAdapter;
 import io.netty.channel.ChannelHandlerContext;
 /**
 * 服务端握手请求，安全认证
 * @author Gong.YiYang
 * @Date 2017年7月20日
 * @Version 
 * @Description
 */
 public class LoginAuthRespHandler extends ChannelHandlerAdapter {
 
 	private Map<String, Boolean> nodeCheck = new ConcurrentHashMap<String,Boolean>();
 	private String[] whiteList = {"127.0.0.1","192.168.1.104"};//白名单
 	
 	@Override
 	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 		// TODO Auto-generated method stub
 		NettyMessage message = (NettyMessage)msg;
 		//如果是握手请求消息处理，其它消息透传
 		if (message.getHeader() != null && message.getHeader().getType() == MessageType.LOGIN_REQ.type) {
 			String nodeIndex = ctx.channel().remoteAddress().toString();//拿到远程地址
 			NettyMessage loginResp = null;
 			//重复登录，拒绝
 			if (nodeCheck.containsKey(nodeIndex)) {
 				loginResp = buildResponse((byte)-1);
 			}else {
 				InetSocketAddress address = (InetSocketAddress)ctx.channel().remoteAddress();
 				String ip = address.getAddress().getHostAddress();
 				boolean isOK = false;
 				for (String Wip : whiteList) {
 					if (Wip.equals(ip)) {
 						isOK = true;
 						break;
 					}
 				}
 				loginResp = isOK ? buildResponse((byte)0) : buildResponse((byte)-1);
 				if (isOK) 
 					nodeCheck.put(nodeIndex, true);
 					ConnectList.list.put(ip, ctx);
 				}
 				
 				System.out.println("The login response is:"+loginResp+"body["+loginResp.getBody()+"]");
 				ctx.writeAndFlush(loginResp);
 			}else {
 				ctx.fireChannelRead(msg);
 			}
 		}
 	
 
 	private NettyMessage buildResponse(byte result) {
 		// TODO Auto-generated method stub
 		NettyMessage message = new NettyMessage();
 		Header header = new Header();
 		header.setType(MessageType.LOGIN_RESP.type);
 		message.setHeader(header);
 		message.setBody(result);
 		return message;
 	}
 	
 	@Override
 	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 		// TODO Auto-generated method stub
 		nodeCheck.remove(ctx.channel().remoteAddress().toString());
 		ctx.fireExceptionCaught(cause);
 	}
 }

package nettyAgreement;
 /**
 * 消息类型
 * @author Gong.YiYang
 * @Date 2017年7月20日
 * @Version 
 * @Description
 */
 public enum MessageType {
 	
 	BUSINESS_REQ((byte)0,"业务请求消息"),
 	BUSINESS_RESP((byte)1,"业务应答消息"),
 	ONE_WAY((byte)2,"业务ONE WAY消息"),
 	LOGIN_REQ((byte)3,"握手请求消息"),
 	LOGIN_RESP((byte)4,"握手应答消息"),
 	HEART_REQ((byte)5,"心跳请求消息"),
 	HEART_RESP((byte)6,"心跳应答消息");
 	
 	public byte type;
 	String describe;
 	private MessageType(byte type, String describe) {
 		this.type = type;
 		this.describe = describe;
 	}
 	
 	
 }

package nettyAgreement;
    
    public class NettyConstant {
    	
    	public static final String REMOTEIP = "127.0.0.1";
    	public static final int PORT = 8080;
    	public static final int LOCAL_PORT = 12088;
    	public static final String LOCALIP = "127.0.0.1";
    }

package nettyAgreement;
 /**
 * 消息定义
 * @author Gong.YiYang
 * @Date 2017年7月18日
 * @Version 
 * @Description
 */
 public final class NettyMessage {
 	private Header header;//消息头
 	private Object body;//消息体
 	public final Header getHeader() {
 		return header;
 	}
 	public final void setHeader(Header header) {
 		this.header = header;
 	}
 	public final Object getBody() {
 		return body;
 	}
 	public final void setBody(Object body) {
 		this.body = body;
 	}
 	@Override
 	public String toString() {
 		return "NettyMessage [header=" + header + "]";
 	}
 	
 }

package nettyAgreement.client;
 /**
 * 客户端
 * @author Gong.YiYang
 * @Date 2017年7月21日
 * @Version 
 * @Description
 */
 
 import java.net.InetSocketAddress;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.channel.ChannelFuture;
 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.NioSocketChannel;
 import io.netty.handler.timeout.ReadTimeoutHandler;
 import nettyAgreement.HeartBeatReqHandler;
 import nettyAgreement.LoginAuthReqHandler;
 import nettyAgreement.NettyConstant;
 import nettyAgreement.decoder.NettyMessageDecoder;
 import nettyAgreement.encoder.NettyMessageEncoder;
 
 public class NettyClient {
 
 	private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
 	
 	EventLoopGroup group = new NioEventLoopGroup();
 	
 	public void connect(int port,String host) throws InterruptedException{
 		
 		try {
 			Bootstrap b = new Bootstrap();
 			b.group(group)
 			.channel(NioSocketChannel.class)
 			.option(ChannelOption.TCP_NODELAY, true)
 			.handler(new ChannelInitializer<SocketChannel>() {
 
 				@Override
 				protected void initChannel(SocketChannel ch) throws Exception {
 					// TODO Auto-generated method stub
 					ch.pipeline().addLast(new NettyMessageDecoder(1024 * 1024, 4, 4,-8,0));
 					ch.pipeline().addLast("MessageEncoder",new NettyMessageEncoder());
 					ch.pipeline().addLast("readTimeOutHandler",new ReadTimeoutHandler(50));
 					ch.pipeline().addLast("LoginAuthHandler",new LoginAuthReqHandler());
 					ch.pipeline().addLast("HeartBeatHandler",new HeartBeatReqHandler());
 				}
 			});
 			//发起异步连接操作
 			ChannelFuture future = b.connect(
 					new InetSocketAddress(host, port), 
 					new InetSocketAddress(NettyConstant.LOCALIP, NettyConstant.LOCAL_PORT)).sync();
 			future.channel().closeFuture().sync();
 		}
 		finally {
 			//所有资源释放完成之后，清空资源，再次发起重连操作
 			executor.execute(new Runnable() {
 				
 				@Override
 				public void run() {
 					// TODO Auto-generated method stub
 					try {
 						TimeUnit.SECONDS.sleep(5);
 						try {
 							connect(NettyConstant.PORT, NettyConstant.REMOTEIP);//发起重连操作
 						} catch (Exception e) {
 							// TODO: handle exception
 							e.printStackTrace();
 						}
 					} catch (InterruptedException e) {
 						// TODO: handle exception
 						e.printStackTrace();
 					}
 				}
 			});
 		} 
 	}
 	public static void main(String[] args) throws InterruptedException {
 		new NettyClient().connect(NettyConstant.PORT, NettyConstant.REMOTEIP);
 	}
 }

package nettyAgreement.server;
 
 
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.ChannelOption;
 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;
 import nettyAgreement.HeartBeatRespHandler;
 import nettyAgreement.LoginAuthRespHandler;
 import nettyAgreement.NettyConstant;
 import nettyAgreement.decoder.NettyMessageDecoder;
 import nettyAgreement.encoder.NettyMessageEncoder;
 
 /**
 * 服务端
 * @author Gong.YiYang
 * @Date 2017年7月21日
 * @Version 
 * @Description
 */
 public class NettyServer {
 	
 	public void build() throws InterruptedException{
 		NioEventLoopGroup bossGroup = new NioEventLoopGroup();
 		NioEventLoopGroup workGroup = new NioEventLoopGroup();
 		try {
 			ServerBootstrap b = new ServerBootstrap();
 			b.group(bossGroup, workGroup)
 			.channel(NioServerSocketChannel.class)
 			.option(ChannelOption.SO_BACKLOG, 100)
 			.handler(new LoggingHandler(LogLevel.INFO))
 			.childHandler(new ChannelInitializer<SocketChannel>() {
 				
 				@Override
 				protected void initChannel(SocketChannel ch) throws Exception {
 					// TODO Auto-generated method stub
 					ch.pipeline().addLast(new NettyMessageDecoder(1024 * 1024, 4, 4,-8,0));
 					ch.pipeline().addLast(new NettyMessageEncoder());
 					ch.pipeline().addLast("readTimeOutHandler",new ReadTimeoutHandler(50));
 					ch.pipeline().addLast(new LoginAuthRespHandler());
 					ch.pipeline().addLast("HeartBeatHandler",new HeartBeatRespHandler());
 				}
 			});
 			
 			//绑定端口，等待同步
 			 ChannelFuture future = b.bind(NettyConstant.PORT).sync();
 			System.out.println("Netty server start ok :"+(NettyConstant.REMOTEIP+":"+NettyConstant.PORT));
 			future.channel().closeFuture().sync();
 		} finally{
 			bossGroup.shutdownGracefully();
 			workGroup.shutdownGracefully();
 		}
 	}
 	
 	public static void main(String[] args) throws InterruptedException {
 		new NettyServer().build();
 	}
 }