Netty集成ProtoBuf开发私有协议
私有协议
广义上区分，通信协议可以分为公有协议和私有协议。由于私有协议的灵活性，它往往会在某个公司或者组织内部使用，按需定制，也因为如此，升级起来会非常方便，灵活性好。绝大多数的私有协议传输层都基于TCP/IP，所以利用Netty的NIO TCP协议栈可以非常方便地进行私有协议的定制和开发。
通信模型
(1) Netty协议栈客户端发送握手请求消息，携带节点ID等有效身份认证信息;
(2) Netty 协议栈服务端对握手请求消息进行合法性校验，包括节点ID有效性校验、节点重复登录校验和IP地址合法性校验，校验通过后，返回登录成功的握手应答消息:
(3)链路建立成功之后，客户端发送业务消息;
(4)链路成功之后，服务端发送心跳消息;
(5)链路建立成功之后，客户端发送心跳消息;
(6)链路建立成功之后，服务端发送业务消息;
(7)服务端退出时，服务端关闭连接，客户端感知对方关闭连接后，被动关闭客户端连接。
ProtoBuf数据格式
syntax = "proto3";
option java_package = "com.fy.protobuf";
option java_outer_classname="CustomMessageData";
message MessageData{
int64 length = 1;
Content content = 2;
enum DataType {
REQ_LOGIN = 0; //上线登录验证环节 等基础信息上报
RSP_LOGIN = 1; //返回上线登录状态与基础信息
PING = 2; //心跳
PONG = 3; //心跳
REQ_ACT = 4; //动作请求
RSP_ACT = 5; //动作响应
REQ_CMD = 6; //指令请求
RSP_CMD = 7; //指令响应
REQ_LOG = 8 ;//日志请求
RSP_LOG = 9; //日志响应
}
DataType order = 3;
message Content{
int64 contentLength = 1;
string data = 2;
}
}开发步骤
tip????下列步骤有点吃力的小伙伴可以看看之前的文章：https://blog.csdn.net/kunfeisang5551/article/details/107957256
1、在D盘protobuf路径下执行命令：protoc.exe --java_out=D:\protobuf CustomMsg.proto
2、将生成的文件拷贝到项目中
开始Coding~
1、新建maven项目，引入依赖
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.51.Final</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.11.0</version>
</dependency>2、创建服务端启动代码
public class CustomServer {
public void bind(int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap
.group(bossGroup, workGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline()
//消息头定长
.addLast(new ProtobufVarint32FrameDecoder())
//解码指定的消息类型
.addLast(new ProtobufDecoder(CustomMessageData.MessageData.getDefaultInstance()))
//消息头设置长度
.addLast(new ProtobufVarint32LengthFieldPrepender())
//解码
.addLast(new ProtobufEncoder())
//心跳检测，超过设置的时间将会抛出异常ReadTimeoutException
.addLast(new ReadTimeoutHandler(8))
//消息处理
.addLast(new CustomServerHandler())
//心跳响应
.addLast(new CustomServerHeartBeatHandler());
}
});
// 绑定端口同步等待启动成功
ChannelFuture sync = bootstrap.bind(port).sync();
// 等待服务监听端口关闭
sync.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}3、创建服务端消息处理代码
public class CustomServerHandler extends ChannelInboundHandlerAdapter {
private String[] whiteIPv4List = {"127.0.0.1", "192.168.1.188"};
public static ConcurrentHashMap nodeCheck = new ConcurrentHashMap();
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
CustomMessageData.MessageData messageData = (CustomMessageData.MessageData) msg;
if (messageData.getOrder() == CustomMessageData.MessageData.DataType.UNRECOGNIZED) {
// 无法识别的消息类型
ctx.close();
}
if (messageData.getOrder() == CustomMessageData.MessageData.DataType.REQ_LOGIN) {
// 检查重复登录
String nodeIndex = ctx.channel().remoteAddress().toString();
if (nodeCheck.contains(nodeIndex)) {
// 重复登录
ctx.writeAndFlush(builderResp(false));
return;
} else {
InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
String ip = socketAddress.getAddress().getHostAddress();
boolean isOk = false;
// 检查白名单
for (String s : whiteIPv4List) {
if (s.equals(ip)) {
isOk = true;
break;
}
}
// 成功响应
CustomMessageData.MessageData responseData = isOk ? builderResp(true) : builderResp(false);
if (isOk) {
nodeCheck.put(nodeIndex, true);
}
ctx.writeAndFlush(responseData);
}
} else {
//心跳消息处理
ctx.fireChannelRead(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
nodeCheck.remove(ctx.channel().remoteAddress().toString());
if (ctx.channel().isActive()) {
ctx.close();
}
}
public CustomMessageData.MessageData builderResp(boolean isOk) {
String r = isOk ? "SUCCESS" : "FAILED";
CustomMessageData.MessageData.Content responseContent = CustomMessageData.MessageData.Content.newBuilder().setData(r).setContentLength(r.length()).build();
CustomMessageData.MessageData responseData = CustomMessageData.MessageData.newBuilder().setOrder(CustomMessageData.MessageData.DataType.RSP_LOGIN).setContent(responseContent).build();
return responseData;
}
}4、创建服务端心跳响应代码
public class CustomServerHeartBeatHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
CustomMessageData.MessageData messageData = (CustomMessageData.MessageData) msg;
if (messageData.getOrder() == CustomMessageData.MessageData.DataType.PING) {
CustomMessageData.MessageData req = CustomMessageData.MessageData.newBuilder()
.setOrder(CustomMessageData.MessageData.DataType.PONG).build();
System.out.println("Send-Client:PONG,time:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
ctx.writeAndFlush(req);
} else {
ctx.fireChannelRead(msg);
}
}
}5、创建客户端启动代码
public class CustomClient {
public void bind(int port) {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline()
.addLast(new ProtobufVarint32FrameDecoder())
.addLast(new ProtobufDecoder(CustomMessageData.MessageData.getDefaultInstance()))
.addLast(new ProtobufVarint32LengthFieldPrepender())
.addLast(new ProtobufEncoder())
// 消息处理
.addLast(new CustomClientHandler())
// 心跳响应
.addLast(new CustomClientHeartBeatHandler());
}
});
ChannelFuture f = b.connect("127.0.0.1", port).sync();
f.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
// 短线重连 定时5秒
group.execute(() -> {
try {
TimeUnit.MILLISECONDS.sleep(5);
bind(port);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
// group.shutdownGracefully();
}
}
}6、创建客户端消息处理代码
这里的逻辑主要是通道激活后马上发送业务消息，然后保持心跳
public class CustomClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
CustomMessageData.MessageData reqData = CustomMessageData
.MessageData
.newBuilder()
.setOrder(CustomMessageData.MessageData.DataType.REQ_LOGIN)
.build();
ctx.channel().writeAndFlush(reqData);
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
CustomMessageData.MessageData respData = (CustomMessageData.MessageData) msg;
if (respData.getOrder() == CustomMessageData.MessageData.DataType.RSP_LOGIN) {
// 响应登录请求处理逻辑
boolean equals = respData.getContent().getData().equals("SUCCESS");
if (equals) {
System.out.println("Receive-Server:LoginSuccess,time:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
System.out.println(respData.toString());
// 传递下一个handler
ctx.fireChannelRead(msg);
} else {
// 登录失败
if (ctx.channel().isActive()) {
ctx.close();
}
}
} else {
// 响应心跳处理逻辑
ctx.fireChannelRead(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
super.exceptionCaught(ctx, cause);
if (ctx.channel().isActive()) {
ctx.close();
}
}
}7、创建客户端心跳保持代码
public class CustomClientHeartBeatHandler extends ChannelInboundHandlerAdapter {
private static ScheduledFuture heartbeatFuture;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
CustomMessageData.MessageData messageData = (CustomMessageData.MessageData) msg;
if (messageData.getOrder() == CustomMessageData.MessageData.DataType.RSP_LOGIN) {
// 登录成功后保持心跳 间隔为5秒
heartbeatFuture = ctx.executor().scheduleAtFixedRate(() -> {
CustomMessageData.MessageData req = CustomMessageData.MessageData.newBuilder()
.setOrder(CustomMessageData.MessageData.DataType.PING).build();
System.out.println("Send-Server:PING,time:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
ctx.writeAndFlush(req);
}, 0, 5, TimeUnit.SECONDS);
} else if (messageData.getOrder() == CustomMessageData.MessageData.DataType.PONG) {
System.out.println("Receive-Server:PONG,time:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
System.out.println();
} else {
ctx.fireChannelRead(msg);
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
// 发生异常就取消心跳保持
if (heartbeatFuture != null) {
heartbeatFuture.cancel(true);
heartbeatFuture = null;
}
ctx.fireExceptionCaught(cause);
}
}8、启动服务端
public class Server {
public static void main(String[] args) throws Exception {
new CustomServer().bind(8080);
}
}9、启动客户端
public class Client {
public static void main(String[] args) {
new CustomClient().bind(8080);
}
}控制台打印
1、客户端
Receive-Server:LoginSuccess,time:2020-08-12 17:31:47
content {
contentLength: 7
data: "SUCCESS"
}
order: RSP_LOGIN
Send-Server:PING,time:2020-08-12 17:31:47
Receive-Server:PONG,time:2020-08-12 17:31:47
Send-Server:PING,time:2020-08-12 17:31:52
Receive-Server:PONG,time:2020-08-12 17:31:52
Send-Server:PING,time:2020-08-12 17:31:57
Receive-Server:PONG,time:2020-08-12 17:31:57
Send-Server:PING,time:2020-08-12 17:32:02
Receive-Server:PONG,time:2020-08-12 17:32:02我们可以看到，当客户端发送登录请求后，服务端响应登录成功消息，然后交替打印心跳保持信息，间隔为5秒。
2、服务端
Send-Client:PONG,time:2020-08-12 17:31:47
Send-Client:PONG,time:2020-08-12 17:31:52
Send-Client:PONG,time:2020-08-12 17:31:57
Send-Client:PONG,time:2020-08-12 17:32:02
Send-Client:PONG,time:2020-08-12 17:32:07服务端响应登录请求后交替打印心跳保持信息。
3、测试服务端异常
我们先停掉服务端，看看客户端有啥反应，客户端日志：
Connection refused: no further information客户端5秒打印一次异常信息，说明短线重连逻辑正常
我们接着再启动服务端，看看客户端有啥反应
io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: no further information: /127.0.0.1:8080
Caused by: java.net.ConnectException: Connection refused: no further information
at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717)
at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:330)
at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:702)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.lang.Thread.run(Thread.java:748)
Receive-Server:LoginSuccess,time:2020-08-12 17:44:15
content {
contentLength: 7
data: "SUCCESS"
}
order: RSP_LOGIN
Send-Server:PING,time:2020-08-12 17:44:15
Receive-Server:PONG,time:2020-08-12 17:44:15
Send-Server:PING,time:2020-08-12 17:44:20
Receive-Server:PONG,time:2020-08-12 17:44:20可以看到由异常转为正常啦~
通过测试可以验证是否符合私有协议的约定：
(1)客户端是否能够正常发起重连:
(2)重连成功之后，不再重连:
(3)断连期间，心跳定时器停止工作，不再发送心跳请求消息;
(4)服务端重启成功之后，允许客户端重新登录;
(5)服务端重启成功之后，客户端能够重连和握手成功:
(6)重连成功之后，双方的心跳能够正常互发。
(7)性能指标:重连期间，客户端资源得到了正常回收，不会导致句柄等资源泄漏。
GitHub服务端地址：https://github.com/GoodBoy2333/netty-server-maven.git
GitHub客户端地址：https://github.com/GoodBoy2333/netty-client-maven.git
