Protobuf是Google protocol buffer的简称，是一种语言中立、平台无关、易于扩展的结构化数据序列化技术，可用于数据传输、存储等领域。
与Protoful类似的序列化技术还有XML、JSON、Thrift等，但Protoful更快、更小、更简单，且具备良好的兼容性。
protobuf 环境准备
首先需要编写 proto 文件。 我们使用的 proto 文件如下。
请求类 SubscribeReq 的内容如下：
syntax = "proto3";
package netty;
option java_package = "com.wyk.netty.codec.protobuf";
option java_outer_classname = "SubscribeReqProto";
message SubscribeReq
{
int32 subReqID = 1;
string userName = 2;
string productName = 3;
repeated string address = 4;
}
返回类 SubscribeResp 的内容如下：
syntax = "proto3";
package netty;
option java_package = "com.wyk.netty.codec.protobuf";
option java_outer_classname = "SubscribeRespProto";
message SubscribeResp {
int32 subReqID = 1;
int32 respCode = 2;
string desc = 3;
}
接下来使用 protoc 工具（可在 protobuf 官网下载）用来将 proto 文件生成 java 类。
protoc --java_out=. SubscribeReq.proto
protoc --java_out=. SubscribeResp.proto
我们需要的 java 类文件就会生成。下面创建 Maven 项目，并在Maven中添加 Netty 和 protobuf 依赖，其中 protobuf 的依赖版本要和 protoc 工具一致。
<dependencies>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.6.Final</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.11.0</version>
</dependency>
</dependencies>
最后将我们生成的2个文件复制到对应的包下。
Netty 服务端开发
首先看下 SubReqServerHandler 的实现。由于ProtobufDecoder 会对消息自动解码，所以这里我们可以直接写业务逻辑。
@ChannelHandler.Sharable
public class SubReqServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
SubscribeReqProto.SubscribeReq req = (SubscribeReqProto.SubscribeReq)msg;
if("wyk".equalsIgnoreCase(req.getUserName())) {
System.out.println("Service accepts client subscribe req : [" +
req.toString() + "]");
ctx.writeAndFlush(resp(req.getSubReqID()));
}
}
// 生成返回报文
private SubscribeRespProto.SubscribeResp resp(int subReqId) {
SubscribeRespProto.SubscribeResp.Builder builder =
SubscribeRespProto.SubscribeResp.newBuilder();
builder.setSubReqID(subReqId);
builder.setRespCode(0);
builder.setDesc("Books order succeed, send to the addresses 3 days later.");
return builder.build();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.close();
}
}
下面看下 SubReqServer 的代码。SubReqServer 首先向 pipeline 中添加 ProtobufVarint32FrameDecoder， 它主要用于半包处理，然后添加 ProtobufDecoder 解码器，它的参数是需要解码的类型。
public class SubReqServer {
public void bind(int port) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup 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
public void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
ch.pipeline().addLast(new ProtobufDecoder(
SubscribeReqProto.SubscribeReq.getDefaultInstance()
));
ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
ch.pipeline().addLast(new ProtobufEncoder());
ch.pipeline().addLast(new SubReqServerHandler());
}
});
ChannelFuture f = b.bind(port).sync();
f.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workGroup.shutdownGracefully();
}
}
}
最后编写启动类。
public class App
{
public static void main( String[] args ) throws Exception {
int port = 8080;
new SubReqServer().bind(port);
}
}
Netty 客户端开发
客户端的环境准备和服务端相同，下面直接讲如何开发。
SubReqClientHandler 的编写如下，其中也是主要包含一些简单的业务逻辑。
public class SubReqClientHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
for(int i = 0; i < 3; i++) {
ctx.write(subReq(i));
}
ctx.flush();
}
private SubscribeReqProto.SubscribeReq subReq(int i) {
SubscribeReqProto.SubscribeReq.Builder builder =
SubscribeReqProto.SubscribeReq.newBuilder();
builder.setSubReqID(i);
builder.setUserName("wyk");
builder.setProductName("book");
builder.addAddress("Beijing");
builder.addAddress("Hebei");
builder.addAddress("Shanghai");
return builder.build();
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg)
throws Exception {
SubscribeRespProto.SubscribeResp resp = (SubscribeRespProto.SubscribeResp)msg;
System.out.println("subReqId: " + resp.getSubReqID());
System.out.println("respCode: " + resp.getRespCode());
System.out.println("desc: " + resp.getDesc());
System.out.println();
@Override
public void channelReadComplete(ChannelHandlerContext ctx)
throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
cause.printStackTrace();
ctx.flush();
}
}
SubReqClient代码如下。
public class SubReqClient {
public void connect(int port, String host) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
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 ProtobufVarint32FrameDecoder());
ch.pipeline().addLast(new ProtobufDecoder(
SubscribeRespProto.SubscribeResp.getDefaultInstance()
));
ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
ch.pipeline().addLast(new ProtobufEncoder());
ch.pipeline().addLast(new SubReqClientHandler());
}
});
ChannelFuture future = b.connect(host, port).sync();
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
最后编写启动类。
public class App
{
public static void main( String[] args ) throws Exception{
int port = 8080;
new SubReqClient().connect(8080, "localhost");
}
}
这样客户端就可以和服务器进行通讯了。
