简介
本文是整理《netty权威指南》第十二章：私有协议栈开发，方便后续自己实现网络通信协议的参考。
几个基本功能描述
消息协议定义，消息编解码，连接认证，链路有效性校验机制，链路的断开重连机制。
主要设计
消息协议设计：消息头+消息体。消息头定义了基本协议字段定义，添加attachment方便扩展。Object编解码使用Marshaller。
链路交互设计：链路什么时候建立，什么时候关闭，握手请求与相应。
可靠性设计：心跳检查机制，重连机制，重复登录保护，消息缓存重发。
安全性设计：本章使用基于ip地址的白名单安全认证机制。
可扩展性设计：添加attachment方便扩展。
具体实现
编解码
编码
编码类继承MessageToByteEncoder，然后编码类重写了encode方法，把具体的消息存放到ByteBuf中就可以了。
解码
解码类继承LengthFieldBasedFrameDecoder：可以根据消息的长度进行拆包，自动的TCP粘包、半包处理。然后解码类重写decode方法进行解析即可。
如上编解码中使用Marshaller对Object对象编解码。
握手与安全认证
TCP建立连接之后，客户端发起握手，服务端进行验证，
客户端LoginAuthReqHandler
重写了channelActive方法，发送登录请求。
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ctx.writeAndFlush(buildLoginReq());
}
private NettyMessage buildLoginReq() {
NettyMessage message = new NettyMessage();
Header header = new Header();
header.setType(MessageType.LOGIN_REQ.value());
message.setHeader(header);
return message;
}
重写了channelRead方法
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
NettyMessage message = (NettyMessage) msg;
// 如果是握手应答消息，需要判断是否认证成功
if (message.getHeader() != null
&& message.getHeader().getType() == MessageType.LOGIN_RESP
.value()) {
byte loginResult = (byte) message.getBody();
if (loginResult != (byte) 0) {
// 握手失败，关闭连接
ctx.close();
} else {
System.out.println("Login is ok : " + message);
ctx.fireChannelRead(msg);
}
} else
ctx.fireChannelRead(msg);
}
服务端LoginAuthRespHandler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
NettyMessage message = (NettyMessage) msg;
// 如果是握手请求消息，处理，其它消息透传
if (message.getHeader() != null
&& message.getHeader().getType() == MessageType.LOGIN_REQ
.value()) {
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 : whitekList) {
if (WIP.equals(ip)) {
isOK = true;
break;
}
}
loginResp = isOK ? buildResponse((byte) 0)
: buildResponse((byte) -1);
if (isOK)
nodeCheck.put(nodeIndex, true);
}
System.out.println("The login response is : " + loginResp
+ " body [" + loginResp.getBody() + "]");
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.value());
message.setHeader(header);
message.setBody(result);
return message;
}
心跳检查机制
客户端发起心跳，服务端检查心跳
客户端检查到握手应答消息，启动心跳检查task，每5秒发送一个heartbeat请求。
客户端HeartBeatReqHandler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
NettyMessage message = (NettyMessage) msg;
// 握手成功，主动发送心跳消息
if (message.getHeader() != null
&& message.getHeader().getType() == MessageType.LOGIN_RESP
.value()) {
heartBeat = ctx.executor().scheduleAtFixedRate(
new HeartBeatTask(ctx), 0, 5000,
TimeUnit.MILLISECONDS);
} else if (message.getHeader() != null
&& message.getHeader().getType() == MessageType.HEARTBEAT_RESP
.value()) {
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() {
NettyMessage heatBeat = buildHeatBeat();
System.out
.println("Client send heart beat messsage to server : ---> "
+ heatBeat);
ctx.writeAndFlush(heatBeat);
}
}
服务端：HeartBeatRespHandler
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
NettyMessage message = (NettyMessage) msg;
// 返回心跳应答消息
if (message.getHeader() != null
&& message.getHeader().getType() == MessageType.HEARTBEAT_REQ
.value()) {
System.out.println("Receive client heart beat message : ---> "
+ 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.value());
message.setHeader(header);
return message;
}
断开重连
客户端断开之后，重写启动连接。
public void connect(int port, String host) throws Exception {
// 配置客户端NIO线程组
try {
Bootstrap b = new Bootstrap();
b.group(group).channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch)
throws Exception {
ch.pipeline().addLast(
new NettyMessageDecoder(1024 * 1024, 4, 4));
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() {
try {
TimeUnit.SECONDS.sleep(1);
try {
connect(NettyConstant.PORT, NettyConstant.REMOTEIP);// 发起重连操作
} catch (Exception e) {
e.printStackTrace();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
链路有效性检测
实现了一个ReadTimeoutHandler，记录每次有数据读的时间lastReadTime，启动一个超时task定期检测，如果在制定时间内没有读事件，则连接抛出读超时异常。
主要成员变量
private final long timeoutMillis; //判定超时时间
private volatile ScheduledFuture<?> timeout; //task
private volatile long lastReadTime;//最近读操作的时间
private volatile int state; // 0 - none, 1 - Initialized, 2 - Destroyed; 连接状态
private boolean closed;//是否已关闭
在handlerAdded，channelRegistered，channelActive初始化
private void initialize(ChannelHandlerContext ctx) {
// Avoid the case where destroy() is called before scheduling timeouts.
// See: https://github.com/netty/netty/issues/143
switch (state) {
case 1:
case 2:
return;
}
state = 1;
lastReadTime = System.currentTimeMillis();
if (timeoutMillis > 0) {
timeout = ctx.executor().schedule(
new ReadTimeoutTask(ctx),
timeoutMillis, TimeUnit.MILLISECONDS);
}
}
ReadTimeoutTask
private final class ReadTimeoutTask implements Runnable {
private final ChannelHandlerContext ctx;
ReadTimeoutTask(ChannelHandlerContext ctx) {
this.ctx = ctx;
}
@Override
public void run() {
if (!ctx.channel().isOpen()) {
return;
}
long currentTime = System.currentTimeMillis();
long nextDelay = timeoutMillis - (currentTime - lastReadTime);
if (nextDelay <= 0) {
// Read timed out - set a new timeout and notify the callback.
timeout = ctx.executor().schedule(this, timeoutMillis, TimeUnit.MILLISECONDS);
try {
readTimedOut(ctx);
} catch (Throwable t) {
ctx.fireExceptionCaught(t);
}
} else {
// Read occurred before the timeout - set a new timeout with shorter delay.
timeout = ctx.executor().schedule(this, nextDelay, TimeUnit.MILLISECONDS);
}
}
}
readTimedOut
protected void readTimedOut(ChannelHandlerContext ctx) throws Exception {
if (!closed) {
ctx.fireExceptionCaught(ReadTimeoutException.INSTANCE);
ctx.close();
closed = true;
}
}
网友评论