Implementing a Custom Protocol with Netty: Design, Encoding/Decoding, and Heartbeat

The article begins by comparing the widely used HTTP protocol with custom protocols, highlighting three main drawbacks of HTTP for high‑frequency service interactions: low byte utilization due to headers and cookies, the overhead of TCP short‑connections requiring frequent three‑way handshakes, and the inability to satisfy specialized business requirements.

To address these issues, a custom binary protocol is introduced. The protocol consists of a fixed‑length header (magic number, version fields, session ID, message type, and attachment count) followed by a variable‑length body. An illustration of the protocol layout is provided, and Ping/Pong messages are defined for heartbeat detection.

Below are the core Java classes that implement the protocol using Netty. All code snippets are kept intact and wrapped in tags.

public class Message { private int magicNumber; private byte mainVersion; private byte subVersion; private byte modifyVersion; private String sessionId; private MessageTypeEnum messageType; private Map attachments = new HashMap<>(); private String body; // getters, setters, and attachment helpers omitted for brevity }

public enum MessageTypeEnum { REQUEST((byte)1), RESPONSE((byte)2), PING((byte)3), PONG((byte)4), EMPTY((byte)5); private byte type; MessageTypeEnum(byte type) { this.type = type; } public int getType() { return type; } public static MessageTypeEnum get(byte type) { for (MessageTypeEnum v : values()) { if (v.type == type) return v; } throw new RuntimeException("unsupported type: " + type); } }

public class MessageEncoder extends MessageToByteEncoder { @Override protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception { if (msg.getMessageType() != MessageTypeEnum.EMPTY) { out.writeInt(Constants.MAGIC_NUMBER); out.writeByte(Constants.MAIN_VERSION); out.writeByte(Constants.SUB_VERSION); out.writeByte(Constants.MODIFY_VERSION); if (!StringUtils.hasText(msg.getSessionId())) { String sessionId = SessionIdGenerator.generate(); msg.setSessionId(sessionId); out.writeCharSequence(sessionId, Charset.defaultCharset()); } out.writeByte(msg.getMessageType().getType()); out.writeShort(msg.getAttachments().size()); msg.getAttachments().forEach((k, v) -> { Charset cs = Charset.defaultCharset(); out.writeInt(k.length()); out.writeCharSequence(k, cs); out.writeInt(v.length()); out.writeCharSequence(v, cs); }); if (msg.getBody() == null) { out.writeInt(0); } else { out.writeInt(msg.getBody().length()); out.writeCharSequence(msg.getBody(), Charset.defaultCharset()); } } } }

public class MessageDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf in, List out) throws Exception { Message msg = new Message(); msg.setMagicNumber(in.readInt()); msg.setMainVersion(in.readByte()); msg.setSubVersion(in.readByte()); msg.setModifyVersion(in.readByte()); CharSequence sessionId = in.readCharSequence(Constants.SESSION_ID_LENGTH, Charset.defaultCharset()); msg.setSessionId(sessionId.toString()); msg.setMessageType(MessageTypeEnum.get(in.readByte())); short attachSize = in.readShort(); for (int i = 0; i < attachSize; i++) { int keyLen = in.readInt(); CharSequence key = in.readCharSequence(keyLen, Charset.defaultCharset()); int valLen = in.readInt(); CharSequence val = in.readCharSequence(valLen, Charset.defaultCharset()); msg.addAttachment(key.toString(), val.toString()); } int bodyLen = in.readInt(); CharSequence body = in.readCharSequence(bodyLen, Charset.defaultCharset()); msg.setBody(body.toString()); out.add(msg); } }

The server side pipeline adds a length‑field based frame decoder/prepender, the custom MessageEncoder and MessageDecoder , and a ServerMessageHandler that delegates processing to resolvers obtained from a singleton MessageResolverFactory .

public class ServerMessageHandler extends SimpleChannelInboundHandler { private MessageResolverFactory resolverFactory = MessageResolverFactory.getInstance(); @Override protected void channelRead0(ChannelHandlerContext ctx, Message msg) throws Exception { Resolver resolver = resolverFactory.getMessageResolver(msg); Message response = resolver.resolve(msg); ctx.writeAndFlush(response); } @Override public void channelRegistered(ChannelHandlerContext ctx) throws Exception { resolverFactory.registerResolver(new RequestMessageResolver()); resolverFactory.registerResolver(new ResponseMessageResolver()); resolverFactory.registerResolver(new PingMessageResolver()); resolverFactory.registerResolver(new PongMessageResolver()); } }

The client pipeline is similar but also adds an IdleStateHandler for heartbeat detection and uses a ClientMessageHandler that extends the server handler. When the channel is idle for reading, it sends a Ping message; when idle for writing, it closes the connection.

public class ClientMessageHandler extends ServerMessageHandler { private ExecutorService executor = Executors.newSingleThreadExecutor(); @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { executor.execute(new MessageSender(ctx)); } @Override public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { if (evt instanceof IdleStateEvent) { IdleStateEvent e = (IdleStateEvent) evt; if (e.state() == IdleState.READER_IDLE) { Message ping = new Message(); ping.setMessageType(MessageTypeEnum.PING); ctx.writeAndFlush(ping); } else if (e.state() == IdleState.WRITER_IDLE) { ctx.close(); } } } private static final class MessageSender implements Runnable { private static final AtomicLong counter = new AtomicLong(1); private volatile ChannelHandlerContext ctx; public MessageSender(ChannelHandlerContext ctx) { this.ctx = ctx; } @Override public void run() { try { while (true) { TimeUnit.SECONDS.sleep(new Random().nextInt(3)); Message msg = new Message(); msg.setMessageType(MessageTypeEnum.REQUEST); msg.setBody("this is my " + counter.getAndIncrement() + " message."); msg.addAttachment("name", "xufeng"); ctx.writeAndFlush(msg); } } catch (InterruptedException e) { e.printStackTrace(); } } } }

The MessageResolverFactory holds a thread‑safe list of Resolver implementations and returns the first one that supports the incoming message type.

public final class MessageResolverFactory { private static final MessageResolverFactory INSTANCE = new MessageResolverFactory(); private static final List resolvers = new CopyOnWriteArrayList<>(); private MessageResolverFactory() {} public static MessageResolverFactory getInstance() { return INSTANCE; } public void registerResolver(Resolver r) { resolvers.add(r); } public Resolver getMessageResolver(Message m) { for (Resolver r : resolvers) { if (r.support(m)) return r; } throw new RuntimeException("cannot find resolver, message type: " + m.getMessageType()); } }

Four concrete resolvers handle REQUEST, RESPONSE, PING, and PONG messages. The request resolver prints the received payload and attachments, then returns a RESPONSE message; the response resolver prints the reply and returns an EMPTY message; the ping resolver returns a PONG; the pong resolver returns an EMPTY message.

Finally, the article provides complete Server and Client bootstrap code that creates Netty event loops, configures the pipeline with the components described above, and starts listening on port 8585 (server) or connects to it (client). Sample console output shows the exchange of ping/pong, request/response, and random request messages.

In summary, the article first contrasts custom protocols with HTTP, defines a simple binary protocol, and then walks through a full Netty implementation that includes message serialization, deserialization, a resolver‑based processing model, and heartbeat functionality.