Netty 实现 protobuf

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");
    }
}

这样客户端就可以和服务器进行通讯了。