<!-- https://mvnrepository.com/artifact/com.google.protobuf/protobuf-java -->
<dependency><groupId>com.google.protobuf</groupId><artifactId>protobuf-java</artifactId><version>3.6.1</version>
</dependency>

syntax = "proto3"; // 版本
option java_outer_classname = "StudentPOJO"; // Java的外部类名
// protobuf 使用message 管理数据
message Student { // 会在 StudentPOJO外部类中生成一个内部类 Student, 它是真正发送的POJO对象/**表示 在Student类中有一个属性名字为 id , 类型为int32(Protobuf类型) 1表示序号, 不是值*/int32 id = 1;string name = 2;
}

protoc.exe --java_out=. Student.proto

package com.dance.netty.netty.protobuf;import com.dance.netty.netty.protobuf.pojo.StudentPOJO;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;public class NettyServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup) //设置两个线程组.channel(NioServerSocketChannel.class) // 使用NioServerSocketChannel作为服务器的通道实现.option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列等待连接个数.childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();// 加入protobuf的解码器 指定解码默认实例pipeline.addLast(new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));pipeline.addLast(new NettyServerHandler());}});System.out.println("server is ready......");ChannelFuture channelFuture = serverBootstrap.bind(6668).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}}

package com.dance.netty.netty.protobuf;import com.dance.netty.netty.protobuf.pojo.StudentPOJO;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.nio.charset.StandardCharsets;
public class NettyServerHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {// 读取从客户端发送的额StudentPOJO.studentStudentPOJO.Student student = (StudentPOJO.Student) msg;System.out.println("客户端发送的数据:" + student.getId() + " " + student.getName());}@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("Hello 客户端", StandardCharsets.UTF_8));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();cause.printStackTrace();}
}

package com.dance.netty.netty.protobuf;import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.protobuf.ProtobufDecoder;
import io.netty.handler.codec.protobuf.ProtobufEncoder;public class NettyClient {public static void main(String[] args) throws InterruptedException {EventLoopGroup eventExecutors = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(eventExecutors).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {ChannelPipeline pipeline = socketChannel.pipeline();// 加入处理Protobuf的编码器pipeline.addLast(new ProtobufEncoder());pipeline.addLast(new NettyClientHandler());}});System.out.println("客户端 ok !");ChannelFuture sync = bootstrap.connect("127.0.0.1", 6668).sync();sync.channel().closeFuture().sync();} catch (InterruptedException e) {e.printStackTrace();} finally {eventExecutors.shutdownGracefully();}}}

package com.dance.netty.netty.protobuf;import com.dance.netty.netty.protobuf.pojo.StudentPOJO;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;import java.nio.charset.StandardCharsets;public class NettyClientHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {// 发送一个Student对象到服务器StudentPOJO.Student dance = StudentPOJO.Student.newBuilder().setId(4).setName("dance").build();ctx.writeAndFlush(dance);}@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;System.out.println("服务器回复的消息: " + byteBuf.toString(StandardCharsets.UTF_8));System.out.println("服务器地址: " + ctx.channel().remoteAddress());}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();cause.printStackTrace();}
}

server is ready......
客户端发送的数据:4 dance
客户端 ok !
服务器回复的消息: Hello 客户端
服务器地址: /127.0.0.1:6668

package com.dance.netty.netty.protobuf;import com.dance.netty.netty.protobuf.pojo.StudentPOJO;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;import java.nio.charset.StandardCharsets;
//public class NettyServerHandler extends ChannelInboundHandlerAdapter {public class NettyServerHandler extends SimpleChannelInboundHandler<StudentPOJO.Student> {@Overridepublic void channelReadComplete(ChannelHandlerContext ctx) throws Exception {ctx.writeAndFlush(Unpooled.copiedBuffer("Hello 客户端", StandardCharsets.UTF_8));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {ctx.close();cause.printStackTrace();}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, StudentPOJO.Student msg) throws Exception {System.out.println("客户端发送的数据:" + msg.getId() + " " + msg.getName());}
}

syntax = "proto3"; // 版本option optimize_for = SPEED; // 加快解析option java_package = "com.dance.netty.netty.protobuf2.pojo"; // 指定包option java_outer_classname = "MyDataInfo"; // Java的外部类名message MyMessage{// 定义一个枚举类型enum DataType {StudentType = 0;WorkerType = 1;}// 用DataType来标识传的是哪个枚举类型DataType dataType = 1;// 表示每次枚举类型最多只能出现其中的一个, 节省空间oneof dataBody{Student student = 2;Worker worker = 3;}
}message Student {int32 id = 1;string name = 2;
}message Worker {int32 id = 1;string name = 2;
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {// 随机发送Student或者Workerint i = new Random().nextInt(3);MyDataInfo.MyMessage myMessage = null;if(0 == i){myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType).setStudent(MyDataInfo.Student.newBuilder().setId(1).setName("flower").build()).build();}else{myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType).setWorker(MyDataInfo.Worker.newBuilder().setId(2).setName("dance").build()).build();}ctx.writeAndFlush(myMessage);}

pipeline.addLast(new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));

public class NettyServerHandler extends SimpleChannelInboundHandler<MyDataInfo.MyMessage> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, MyDataInfo.MyMessage msg) throws Exception {if(msg.getDataType() == MyDataInfo.MyMessage.DataType.StudentType){System.out.println("客户端发送的数据:" + msg.getStudent().getId() + " " + msg.getStudent().getName());}else if(msg.getDataType() == MyDataInfo.MyMessage.DataType.WorkerType){System.out.println("客户端发送的数据:" + msg.getWorker().getId() + " " + msg.getWorker().getName());}
}

客户端 ok !
服务器回复的消息: Hello 客户端
服务器地址: /127.0.0.1:6668

server is ready......
客户端发送的数据:2 dance
客户端发送的数据:2 dance
客户端发送的数据:1 flower