java接收小程序发送的protobuf消息

1、定义一个proto类型的消息

消息的核心内容如下：

message ChargingCmd {string cmd = 1;
}

2、使用前面博客介绍的方法将消息转为Java类ChargingCmdProtobuf

3、创建一个消息类型转换处理器

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import lombok.extern.slf4j.Slf4j;import java.util.List;/*** @author: * @Desc:消息类型转换处理器* @create: 2025-10-03 18:04**/
@Slf4j
public class ProtobufMsgToMsgHandler extends MessageToMessageDecoder<WebSocketFrame> {@Overrideprotected void decode(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame, List<Object> list) throws Exception {// 判断WebSocketFrame类型，如果是TextWebSocketFrame，则进行处理if (webSocketFrame instanceof TextWebSocketFrame) {//转为字符串类型String text = ((TextWebSocketFrame) webSocketFrame).text();log.info("WebSocket收到消息：{}", text);//处理二进制消息,protobuf消息正是以二进制形式传输的}else if(webSocketFrame instanceof BinaryWebSocketFrame){ByteBuf bytes = ((BinaryWebSocketFrame) webSocketFrame).content();//添加到list中，交给下一个handler处理list.add(bytes);//retain引用计数，防止释放bytes.retain();}}
}

核心代码：

//处理二进制消息,protobuf消息以二进制形式传输的
if(webSocketFrame instanceof BinaryWebSocketFrame){ByteBuf bytes = ((BinaryWebSocketFrame) webSocketFrame).content();//添加到list中，交给下一个handler处理list.add(bytes);//retain引用计数，防止释放bytes.retain();
}

4、创建一个反序列化处理器

@Slf4j
public class ProtobufDeserializeHandler extends SimpleChannelInboundHandler<ChargingCmdProtobuf.ChargingCmd> {@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, ChargingCmdProtobuf.ChargingCmd chargingCmd) throws Exception {log.info("接收到小程序发送过来的指令为:{}", chargingCmd.getCmd());}
}

5、Websocket入站处理器增加protobuf消息的处理

    /*** 接收到消息时触发，处理WebSocket消息* @param channelHandlerContext* @param webSocketFrame* @throws Exception*/@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, WebSocketFrame webSocketFrame) throws Exception {// 判断WebSocketFrame类型，如果是TextWebSocketFrame，则进行处理if (webSocketFrame instanceof TextWebSocketFrame) {//转为字符串类型String text = ((TextWebSocketFrame) webSocketFrame).text();log.info("WebSocket收到消息：{}", text);}else if(webSocketFrame instanceof BinaryWebSocketFrame){//处理二进制消息,protobuf消息正是以二进制形式传输的byte[] bytes = ((BinaryWebSocketFrame) webSocketFrame).content().array();log.info("WebSocket收到二进制消息：{}", bytes);}}

6、自定义通道处理器添加上以上添加的处理器

/*** @Desc:自定义通道初始化器* @create: 2025-10-01 10:04* **/
public class MyChannelInit extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel socketChannel) throws Exception {//取出channelPipelineChannelPipeline pipeline = socketChannel.pipeline();/* ** Netty内置了很多编码解码器* 1、HttpServerCodec：HTTP解码器，用于处理HTTP请求* Netty中以Codec为后缀的类，一般都是编码解码器,既可以编码也可以解码* HttpServerCodec只能处理HTTP get请求，不能处理post请求* 2、HttpObjectAggregator：可处理Http POST请求，用于将多个Http消息聚合为完整的HTTP请求* HttpObjectAggregator能够将HttpMessage和HttpContent聚合为FullHttpRequest或FullHttpResponse***/pipeline//添加Netty内置的读写超时的处理器.addLast(new IdleStateHandler(30,30,30, TimeUnit.SECONDS))//添加自定义心跳检测处理器.addLast(new MyServerHeartBeatHandler())//处理HTTP GET请求.addLast(new HttpServerCodec())//处理HTTP POST请求，聚合Http消息，防止粘包.addLast(new HttpObjectAggregator(1048576))//处理WebSocket请求.addLast(new WebSocketServerProtocolHandler("/ws"))//添加消息转换处理器，将WebSocketFrame消息转换为ProtoBuf二进制类型的消息.addLast(new ProtobufMsgToMsgHandler())//接收数据，对ProtoBu数据进行反序列化,返回的是ChargingCmd对象，ChargingCmdProtobuf.ChargingCmd是我们定义的消息类型，.addLast(new ProtobufDecoder(ChargingCmdProtobuf.ChargingCmd.getDefaultInstance()))//发送数据，对ProtoBu数据进行序列化.addLast(new ProtobufDeserializeHandler())//发送数据，ProtoBu序列化.addLast(new ProtobufEncoder())//自定义处理器.addLast(new WebSocketInboundHandler());}
}

核心代码：

//添加消息转换处理器，将WebSocketFrame消息转换为ProtoBuf二进制类型的消息
.addLast(new ProtobufMsgToMsgHandler())
//接收数据，对ProtoBu数据进行反序列化,返回的是ChargingCmd对象，ChargingCmdProtobuf.ChargingCmd是我们定义的消息类型，
.addLast(new ProtobufDecoder(ChargingCmdProtobuf.ChargingCmd.getDefaultInstance()))
//发送数据，对ProtoBu数据进行序列化
.addLast(new ProtobufDeserializeHandler())
//发送数据，ProtoBu序列化
.addLast(new ProtobufEncoder())
//自定义处理器
.addLast(new WebSocketInboundHandler());