Integrating Spring Boot with Netty Using Protobuf for Data Transmission

Introduction

This article introduces the integration of Spring Boot with Netty and the use of Google Protobuf for efficient binary data exchange between a server and a client.

Protobuf

Overview

Protobuf is a language‑independent binary serialization format that is faster than XML and suitable for distributed systems, configuration files, and data storage.

https://github.com/google/protobuf

Usage

Define a UserMsg message in a .proto file (proto3 syntax) and generate Java classes with protoc.exe :

syntax = "proto3";
// package name
option java_package = "com.pancm.protobuf";
// outer class name
option java_outer_classname = "UserInfo";

message UserMsg {
  // ID
  int32 id = 1;
  // Name
  string name = 2;
  // Age
  int32 age = 3;
  // State
  int32 state = 4;
}

Compile the proto file:

protoc.exe --java_out=E:\protobuf User.proto

Server Implementation

The server side consists of a Spring‑managed Netty bootstrap, a filter that sets up the pipeline (idle handler, Protobuf decoders/encoders), and a handler that processes incoming UserMsg objects, sends heartbeats, and handles idle timeouts.

@Service("nettyServer")
public class NettyServer {
    private static final int port = 9876;
    private static EventLoopGroup boss = new NioEventLoopGroup();
    private static EventLoopGroup work = new NioEventLoopGroup();
    private static ServerBootstrap b = new ServerBootstrap();
    @Autowired
    private NettyServerFilter nettyServerFilter;
    public void run() {
        try {
            b.group(boss, work);
            b.channel(NioServerSocketChannel.class);
            b.childHandler(nettyServerFilter);
            ChannelFuture f = b.bind(port).sync();
            System.out.println("服务端启动成功,端口是:" + port);
            f.channel().closeFuture().sync();
        } finally {
            work.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }
}

@Component
public class NettyServerFilter extends ChannelInitializer
{
    @Autowired
    private NettyServerHandler nettyServerHandler;
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline ph = ch.pipeline();
        ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
        ph.addLast(new ProtobufVarint32FrameDecoder());
        ph.addLast(new ProtobufDecoder(UserMsg.getDefaultInstance()));
        ph.addLast(new ProtobufVarint32LengthFieldPrepender());
        ph.addLast(new ProtobufEncoder());
        ph.addLast("nettyServerHandler", nettyServerHandler);
    }
}

@Service("nettyServerHandler")
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    private int idle_count = 1;
    private int count = 1;
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("连接的客户端地址:" + ctx.channel().remoteAddress());
        UserInfo.UserMsg userMsg = UserInfo.UserMsg.newBuilder()
            .setId(1).setAge(18).setName("xuwujing").setState(0).build();
        ctx.writeAndFlush(userMsg);
        super.channelActive(ctx);
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            if (IdleState.READER_IDLE.equals(event.state())) {
                System.out.println("已经5秒没有接收到客户端的信息了");
                if (idle_count > 1) {
                    System.out.println("关闭这个不活跃的channel");
                    ctx.channel().close();
                }
                idle_count++;
            }
        } else {
            super.userEventTriggered(ctx, obj);
        }
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println("第" + count + "次,服务端接受的消息:" + msg);
        try {
            if (msg instanceof UserMsg) {
                UserInfo.UserMsg userState = (UserInfo.UserMsg) msg;
                if (userState.getState() == 1) {
                    System.out.println("客户端业务处理成功!");
                } else if (userState.getState() == 2) {
                    System.out.println("接受到客户端发送的心跳!");
                } else {
                    System.out.println("未知命令!");
                }
            } else {
                System.out.println("未知数据!" + msg);
                return;
            }
        } finally {
            ReferenceCountUtil.release(msg);
        }
        count++;
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

Client Implementation

The client mirrors the server pipeline, sends heartbeat messages every 4 seconds, and attempts reconnection after a disconnection.

@Service("nettyClientHandler")
@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    @Autowired
    private NettyClient nettyClient;
    private int fcount = 1;
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("建立连接时：" + new Date());
        ctx.fireChannelActive();
    }
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("关闭连接时：" + new Date());
        final EventLoop eventLoop = ctx.channel().eventLoop();
        nettyClient.doConnect(new Bootstrap(), eventLoop);
        super.channelInactive(ctx);
    }
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            if (IdleState.WRITER_IDLE.equals(event.state())) {
                UserMsg.Builder userState = UserMsg.newBuilder().setState(2);
                ctx.channel().writeAndFlush(userState);
                fcount++;
            }
        }
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (!(msg instanceof UserMsg)) {
            System.out.println("未知数据!" + msg);
            return;
        }
        try {
            UserInfo.UserMsg userMsg = (UserInfo.UserMsg) msg;
            System.out.println("客户端接受到的用户信息。编号:" + userMsg.getId() + ",姓名:" + userMsg.getName() + ",年龄:" + userMsg.getAge());
            UserMsg.Builder userState = UserMsg.newBuilder().setState(1);
            ctx.writeAndFlush(userState);
            System.out.println("成功发送给服务端!");
        } finally {
            ReferenceCountUtil.release(msg);
        }
    }
}

Testing

Start the server first, then the client. The console output shows successful connection, message exchange, heartbeat handling, and automatic reconnection when the client starts before the server.

服务端启动成功,端口是:9876
连接的客户端地址:/127.0.0.1:53319
第1次,服务端接受的消息:state: 1
客户端业务处理成功!
第2次,服务端接受的消息:state: 2
接受到客户端发送的心跳!
...

Other Resources

Project source code:

https://github.com/xuwujing/springBoot-study/tree/master/springboot-netty-protobuf

Non‑Spring‑Boot Netty‑Protobuf example:

https://github.com/xuwujing/Netty-study/tree/master/Netty-protobuf