Netty对Protocol Buffer的支持(七)

时间:2024-07-28 21:35:26

Netty对Protocol Buffer的支持(七)

一.简介

  在上一篇博文中笔者已经介绍了google的Protocol Buffer的使用,那么本文笔者就开始介绍netty对Protocol Buffer的支持。

二.编码的实现

2.1 服务端启动代码

public class ServerTest {
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup(); try{
ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ServerChannelInitilizer()); ChannelFuture channelFuture = serverBootstrap.bind(8989).sync();
channelFuture.channel().closeFuture().sync();
}finally{
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

2.2 服务端管道初始化代码

public class ServerChannelInitilizer extends ChannelInitializer<SocketChannel>{

    @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline(); /**
* 采用Base 128 Varints进行编码,在消息头上加上32个整数,来标注数据的长度。
*/
pipeline.addLast("protobufVarint32FrameDecoder", new ProtobufVarint32FrameDecoder());
pipeline.addLast("protobufDecoder", new ProtobufDecoder(PersonsBook.AddressBook.getDefaultInstance())); /**
* 对采用Base 128 Varints进行编码的数据解码
*/
pipeline.addLast("protobufVarint32LengthFieldPrepender", new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast("protobufEncoder", new ProtobufEncoder()); pipeline.addLast("serverHandler", new ServerHandler());
}
}

2.3 服务端Handler处理

public class ServerHandler extends SimpleChannelInboundHandler<AddressBook>{

    @Override
protected void channelRead0(ChannelHandlerContext ctx, AddressBook msg) throws Exception {
List<Person> list = msg.getPersonList();
list.forEach(p -> System.out.println(p.getName() + ";;" + p.getId() + ";;" + p.getEmail()));
}
}

2.4 客户端启动程序

public class ClientTest {
public static void main(String[] args) throws Exception {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ClientChannelInitializer()); ChannelFuture channelFuture = bootstrap.connect("localhost", 8989).sync();
channelFuture.channel().closeFuture().sync();
}finally{
eventLoopGroup.shutdownGracefully();
}
}
}

2.5客户端通道初始化

public class ClientChannelInitializer extends ChannelInitializer<SocketChannel>{

    @Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline(); pipeline.addLast("protobufVarint32FrameDecoder", new ProtobufVarint32FrameDecoder());
pipeline.addLast("protobufDecoder", new ProtobufDecoder(PersonsBook.AddressBook.getDefaultInstance()));
pipeline.addLast("protobufVarint32LengthFieldPrepender", new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast("protobufEncoder", new ProtobufEncoder()); pipeline.addLast("clientHandler", new ClientHandler());
}
}

2.6 客户端Handler处理代码

public class ClientHandler extends SimpleChannelInboundHandler<AddressBook>{

    @Override
protected void channelRead0(ChannelHandlerContext ctx, AddressBook msg) throws Exception { } @Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
AddressBook ab = AddressBook.newBuilder()
.addPerson(Person.newBuilder().setEmail("123@qq.com").setId(23).setName("zhangsan"))
.addPerson(Person.newBuilder().setEmail("789@163.com").setId(45).setName("lisi"))
.build(); ctx.writeAndFlush(ab);
}
}

三.程序运行

  先运行服务端启动代码,在运行客户端启动代码。