1.Protobuf基本介绍和使用示意图
- Protobuf 是 Google 发布的开源项目，全称 Google Protocol Buffers，是一种轻便高效的结构化数据存储格式，可以用于结构化数据串行化，或者说序列化。它很适合做数据存储或 RPC[远程过程调用 remote procedure call ] 数据交换格式 。目前很多公司 http+json tcp+protobuf.
- 参考文档 : https://developers.google.com/protocol-buffers/docs/proto 语言指南.
- Protobuf 是以 message 的方式来管理数据的.
- 支持跨平台、跨语言，即[客户端和服务器端可以是不同的语言编写的（支持目前绝大多数语言，例如 C++、C#、Java、python 等）
2.Protobuf使用示意图
3.代码
1.编写.proto文件
//版本号
syntax = "proto3";
//加快解析
option optimize_for = SPEED; // 加快解析
//指定生成包下
//option java_package="com.hyy.netty.proto.moreObj";
//外部类名
option java_outer_classname = "MyData";
//内部类-1
message Student {
int32 id = 1; //属性,java的int类型
string name = 2; //java的String类型
}
//内部类-2
message Worker {
string name = 1; //java的String类型
int32 id = 2; //属性,java的int类型
}
//内部类-3,用来管理其他类型
message MyMessage {
//定义枚举类
enum DataType {
StudentType = 0; //在proto3中枚举从0开始
WorkerType = 1;
}
//用来标识传的是哪个枚举类型
DataType date_type = 1;
//每次只能出现一个枚举类型
oneof dataOneType {
Student student = 2;
Worker worker = 3;
}
}
2.使用protoc.exe编译后的java文件(部分)
public final class MyData {
private MyData() {}
public static void registerAllExtensions(
com.google.protobuf.ExtensionRegistryLite registry) {
}
.....代码过长,省略
}
3.服务端
public class MyServer { public static void main(String[] args) throws InterruptedException { NioEventLoopGroup boss = new NioEventLoopGroup(); NioEventLoopGroup worker = new NioEventLoopGroup(); try{ ServerBootstrap serverBootstrap=new ServerBootstrap(); serverBootstrap.group(boss,worker) .channel(NioServerSocketChannel.class) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); /** * 加入谷歌解码器ProtobufDecoder,需要指定对象 * ProtobufDecoder(MyDataInfo.Student.getDefaultInstance()) * 指定类的内部类对象 */ pipeline.addLast(new ProtobufDecoder(MyData.MyMessage.getDefaultInstance())); pipeline.addLast(new ServerHandler()); } }); ChannelFuture channelFuture = serverBootstrap.bind(7000).sync(); channelFuture.channel().closeFuture().sync(); }finally { boss.shutdownGracefully(); worker.shutdownGracefully(); } } }
4.服务端处理类
public class ServerHandler extends SimpleChannelInboundHandler<MyData.MyMessage> { /** * * 读取数据实际(这里我们可以读取客户端发送的消息) * @param ctx the {@link ChannelHandlerContext} which this {@link SimpleChannelInboundHandler} * belongs to * @param msg the message to handle * @throws Exception is thrown if an error occurred */ @Override protected void channelRead0(ChannelHandlerContext ctx, MyData.MyMessage msg) throws Exception { //根据客户端定义的DataType,显示不同的信息 MyData.MyMessage.DataType dataType=msg.getDateType(); if (dataType==MyData.MyMessage.DataType.StudentType){ MyData.Student student=msg.getStudent(); System.out.println("学生名字:"+student.getName()+",学生ID:"+student.getId()); }else if (dataType==MyData.MyMessage.DataType.WorkerType){ MyData.Worker worker=msg.getWorker(); System.out.println("工人名字:"+worker.getName()+",工人ID:"+worker.getId()); }else { System.out.println("类型不对"); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { //writeAndFlush 是 write + flush //将数据写入到缓存，并刷新 //一般讲，我们对这个发送的数据进行编码 //发送给客户端消息 ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端!!!!我是服务端.......", CharsetUtil.UTF_8)); } //处理异常, 一般是需要关闭通道 @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }
5.客户端
public class MyClient { public static void main(String[] args) throws InterruptedException { //创建一个事件循环组 NioEventLoopGroup eventExecutors = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(eventExecutors) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel ch) throws Exception { ChannelPipeline pipeline = ch.pipeline(); //加入谷歌编码器 pipeline.addLast(new ProtobufEncoder()); //加入自定义处理类 pipeline.addLast(new MyClientHandler()); } }); ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync(); //监视关闭 channelFuture.channel().closeFuture().sync(); } finally { eventExecutors.shutdownGracefully(); } } }
6.客户端处理类
public class MyClientHandler extends ChannelInboundHandlerAdapter {
//当通道就绪就会触发该方法
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
//随机发送Student/worker对象
int random = new Random().nextInt(3);
MyData.MyMessage myMessage = null;
if (0 == random) {
myMessage = MyData.MyMessage.newBuilder()
.setDateType(MyData.MyMessage.DataType.StudentType)
.setStudent(MyData.Student.newBuilder().setId(1).setName("张三"))
.build();
} else {
myMessage = MyData.MyMessage.newBuilder()
.setDateType(MyData.MyMessage.DataType.WorkerType)
.setWorker(MyData.Worker.newBuilder().setId(2).setName("李四"))
.build();
}
ctx.writeAndFlush(myMessage);
}
}