Netty使用Protobuf编解码

Protobuf是一个灵活、高效、结构化的数据序列化框架,相比于XML等传统的序列化工具,它更小、更快、更简单。Protobuf支持数据结构化一次可以到处使用,甚至跨语言使用,通过代码生成工具可以自动生成不同语言版本的源代码,甚至可以在使用不同版本的数据结构进程间进行数据传递,实现数据结构的前向兼容。

下面结合一个例子看看在Netty如何使用Protobuf对POJO对象进行编解码。

1.Protobuf环境搭建

https://developers.google.com/protocol-buffers/docs/downloads下载Protobuf,本例中是3.0.0版本。

下载后对压缩包进行解压,在/bin目录可以看到protoc.exe。protoc.exe根据.proto文件生成代码。下面根据图书订购程序定义SubcribeReq.proto和SubcribeResp.proto文件。
SubcribeReq.proto文件内容:

1
2
3
4
5
6
7
8
9
10
11
syntax = "proto3";
package netty;
option java_package = "com.tommy.netty.protobuf";
option java_outer_classname = "SubcribeReqProto";

message SubcribeReq {
int32 subReqID = 1;
string userName = 2;
string productName = 3;
repeated string address = 4;
}

SubcribeResp.proto文件内容:

1
2
3
4
5
6
7
8
9
10
syntax = "proto3";
package netty;
option java_package = "com.tommy.netty.protobuf";
option java_outer_classname = "SubcribeRespProto";

message SubcribeResp {
int32 subReqID = 1;
int32 respCode = 2;
string desc = 3;
}

通过protoc.exe生成代码。
进入到protoc.exe目录,分别执行:
protoc.exe --java_out=.\src .\netty\SubcribeReq.protoprotoc.exe --java_out=.\src .\netty\SubcribeResp.proto
在/bin目录的src目录中可以看到生成的文件。(注意:要提前创建好src目录)

将生成的SubcribeReqProto.java和SubcribeRespProto.java复制到工程中。

在工程中导入protobuf3.0.0的包:

1
2
3
4
5
6
<!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.0.0</version>
</dependency>

2.Protobuf编解码开发

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
/**
* @author j.tommy
* @version 1.0
* @date 2017/12/10
*/
public class TestSubcribeReqProto {
private static byte[] encode(SubcribeReqProto.SubcribeReq subcribeReq) {
return subcribeReq.toByteArray();
}
private static SubcribeReqProto.SubcribeReq decode(byte[] body) throws InvalidProtocolBufferException {
return SubcribeReqProto.SubcribeReq.parseFrom(body);
}
private static SubcribeReqProto.SubcribeReq createSubcribeReq() {
SubcribeReqProto.SubcribeReq.Builder builder = SubcribeReqProto.SubcribeReq.newBuilder();
builder.setSubReqID(1);
builder.setUserName("j.tommy");
builder.setProductName("Netty权威指南");
List<String> addressList = new ArrayList<String>();
addressList.add("北京市");
addressList.add("上海市");
addressList.add("西安市");
addressList.add("深圳市");
builder.addAllAddress(addressList);
return builder.build();
}
public static void main(String[] args) throws InvalidProtocolBufferException {
SubcribeReqProto.SubcribeReq req = createSubcribeReq();
System.out.println("Before encode:" + req);
SubcribeReqProto.SubcribeReq req2 = decode(encode(req));
System.out.println("After decode:" + req2);
System.out.println("Assert equals:" + req.equals(req2));
}
}

通过SubcribeReq的.newBuilder()创建Builder,通过Builder设置SubscribeReq的属性,最后通过builder.builder()方法生成对象。

编码时通过SubscribeReq的toByteArray()即可将SubcribeReq编码为直接数组。
解码时通过SubscribeReq的parseForm将二进制数组编码为SubscribeReq对象。

运行结果:

3.Netty的Protobuf订购程序开发

服务端代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
/**
* @author j.tommy
* @version 1.0
* @date 2017/12/10
*/
public class SubcribeServer {
public static void main(String[] args) {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ServerBootstrap s = new ServerBootstrap();
s.group(bossGroup,workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.option(ChannelOption.SO_BACKLOG, 100)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new ProtobufVarint32FrameDecoder());
sc.pipeline().addLast(new ProtobufDecoder(SubcribeReqProto.SubcribeReq.getDefaultInstance()));
sc.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
sc.pipeline().addLast(new ProtobufEncoder());
sc.pipeline().addLast(new SubcribeServerHandler());
}
});
try {
ChannelFuture cf = s.bind(9989).sync();
cf.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
class SubcribeServerHandler extends ChannelHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
ctx.close();
}
private SubcribeRespProto.SubcribeResp createSubcribeResp(int subReqID) {
SubcribeRespProto.SubcribeResp.Builder builder = SubcribeRespProto.SubcribeResp.newBuilder();
builder.setSubReqID(subReqID);
builder.setRespCode(0);
builder.setDesc("Order success.");
return builder.build();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
SubcribeReqProto.SubcribeReq req = (SubcribeReqProto.SubcribeReq) msg;
System.out.println("接收到客户端请求:" + req.getSubReqID() + ",userName:" + req.getUserName() + ",productName:" + req.getProductName());
SubcribeRespProto.SubcribeResp resp = createSubcribeResp(req.getSubReqID());
ctx.writeAndFlush(resp);
}
}

客户端代码:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
/**
* @author j.tommy
* @version 1.0
* @date 2017/12/10
*/
public class SubcribleClient {
public static void main(String[] args) {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel sc) throws Exception {
sc.pipeline().addLast(new ProtobufVarint32FrameDecoder());
sc.pipeline().addLast(new ProtobufDecoder(SubcribeRespProto.SubcribeResp.getDefaultInstance()));
sc.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
sc.pipeline().addLast(new ProtobufEncoder());
sc.pipeline().addLast(new SubcribleClientHandler());
}
});
try {
ChannelFuture f = b.connect("127.0.0.1", 9989).sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
}
class SubcribleClientHandler extends ChannelHandlerAdapter {
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
SubcribeReqProto.SubcribeReq req = null;
for (int i=0;i<10;i++) {
req = createSubcribeReq(i);
ctx.write(req);
}
ctx.flush();
}
private SubcribeReqProto.SubcribeReq createSubcribeReq(int subSeqID) {
SubcribeReqProto.SubcribeReq.Builder builder = SubcribeReqProto.SubcribeReq.newBuilder();
builder.setSubReqID(subSeqID);
builder.setUserName("j.tommy");
builder.setProductName("netty权威指南");
List<String> addressList = new ArrayList<String>();
addressList.add("北京市");
addressList.add("西安市");
addressList.add("深圳市");
return builder.build();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
SubcribeRespProto.SubcribeResp resp = (SubcribeRespProto.SubcribeResp) msg;
System.out.println("接收到服务端响应:" + resp.getSubReqID() + ",responseCode:" + resp.getRespCode() + ",desc:" + resp.getDesc());
}
}

ProtobufVarint32FrameDecoder,它主要用来处理半包;

ProtobufDecoder解压器,它的参数是com.google.protobuf.MessageLite,实际上是告诉ProtobufDecoder需要解码的目标类是什么,否则仅仅从字节数组是无法知道要解码的目标类型信息的。

服务端中ProtobufEncoder用于对响应的SubcribeResp进行编码。

运行结果:
服务端:

客户端:

4.Protobuf的使用注意事项

ProtobufDecoder仅仅负责解码,它不支持读半包。因此在ProtobufDecoder的前面,一定要有能够处理半包消息的解码器。
有3种方式可以选择:
1.使用Netty提供的ProtobufVarint32FrameDecoder,它可以处理半包消息;
2.继承Netty提供的通用半包解码器LengthFieldBasedFrameDecoder;
3.继承ByteToMessageDecoder类,自己处理半包消息。

如果只使用ProtobufDecoder解码器,而忽略对半包消息的处理,程序没法正常工作。

Donny wechat
欢迎关注我的个人公众号
打赏,是超越赞的一种表达。
Show comments from Gitment