前言

比起上文的Protobuf协议，Protostuff的好处就是不用去编辑proto文件，然后生成对应的java文件，直接可以利用现有的实体类来通信，而且性能性能损失也是很少的。

github官方链接地址

依赖

 <dependency><groupId>io.protostuff</groupId><artifactId>protostuff-core</artifactId><version>1.7.2</version></dependency><dependency><groupId>io.protostuff</groupId><artifactId>protostuff-runtime</artifactId><version>1.7.2</version></dependency>

定义编解码器

Protobuf有netty提供已经封装好的编解码器，但是Protostuff需要自己定义。

工具类

public class ProtostuffUtils {/*** 缓存Schema*/private static Map<Class<?>, Schema<?>> schemaCache = new ConcurrentHashMap<>();/*** 序列化方法，把指定对象序列化成字节数组** @param obj* @param <T>* @return*/@SuppressWarnings("unchecked")public static <T> byte[] serialize(T obj) {Class<T> clazz = (Class<T>) obj.getClass();Schema<T> schema = getSchema(clazz);LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);byte[] data;try {data = ProtobufIOUtil.toByteArray(obj, schema, buffer);} finally {buffer.clear();}return data;}/*** 反序列化方法，将字节数组反序列化成指定Class类型** @param data* @param clazz* @param <T>* @return*/public static <T> T deserialize(byte[] data, Class<T> clazz) {Schema<T> schema = getSchema(clazz);T obj = schema.newMessage();ProtobufIOUtil.mergeFrom(data, obj, schema);return obj;}@SuppressWarnings("unchecked")private static <T> Schema<T> getSchema(Class<T> clazz) {Schema<T> schema = (Schema<T>) schemaCache.get(clazz);if (Objects.isNull(schema)) {//这个schema通过RuntimeSchema进行懒创建并缓存//所以可以一直调用RuntimeSchema.getSchema(),这个方法是线程安全的schema = RuntimeSchema.getSchema(clazz);if (Objects.nonNull(schema)) {schemaCache.put(clazz, schema);}}return schema;}
}

编码器

这里采用分隔符的方式来解决拆包粘包的问题，那么分隔符实验了多次，如果用换行符\n，会报错，错误信息如下：

Protocol message contained an invalid tag (zero)

好像是因为\的问题，所以只能采用不带\的分隔符了，这里选用@来作为分隔符。

/*** @author: zhouwenjie* @description:* @create: 2022-07-12 11:19**/
public class ProtostuffEncoder extends MessageToByteEncoder {//分隔符@，放在配置文件，方便修改配置private String delimiter;public ProtostuffEncoder(String delimiter) {this.delimiter = delimiter;}@Overrideprotected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception {if (msg instanceof ComputeResult) {byte[] bytes = ProtostuffUtils.serialize(msg);byte[] delimiterBytes = delimiter.getBytes();byte[] total = new byte[bytes.length + delimiterBytes.length];System.arraycopy(bytes, 0, total, 0, bytes.length);System.arraycopy(delimiterBytes, 0, total, bytes.length, delimiterBytes.length);out.writeBytes(total);}}
}

解码器

/*** @author: zhouwenjie* @description:* @create: 2022-07-12 11:17**/
public class ProtostuffDecoder extends MessageToMessageDecoder<ByteBuf>{// 写在这里，方便扩展private Class<?> clazz;public ProtostuffDecoder(Class<?> clazz) {this.clazz = clazz;}@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> list) {try {byte[] body = new byte[in.readableBytes()];  //传输正常in.readBytes(body);list.add(ProtostuffUtils.deserialize(body, clazz));} catch (Exception e) {e.printStackTrace();}}
}

整合到netty

这里以客户端为例，服务端同理。

启动类

/*** @author: zhouwenjie* @description: 客户端* @create: 2020-04-03 17:14**/
@Component
@Slf4j
public class NettyClient {@Value("${monitor.server.host}")private String host;@Value("${monitor.server.port}")private int port;@Value("${monitor.delimiter}")private String delimiter;@Autowiredprivate NettyClientHandler nettyClientHandler;private NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();private Bootstrap bootstrap;@PostConstructpublic void run() throws UnknownHostException {bootstrap = new Bootstrap();bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).remoteAddress(host, port).option(ChannelOption.TCP_NODELAY, true).option(ChannelOption.SO_KEEPALIVE, true).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel socketChannel) {//客户端初始化socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024*8, Unpooled.wrappedBuffer(delimiter.getBytes())));socketChannel.pipeline().addLast(new ProtostuffDecoder(ComputeResult.class));socketChannel.pipeline().addLast(new ProtostuffEncoder(delimiter));socketChannel.pipeline().addLast(nettyClientHandler);}});String hostAddress = InetAddress.getLocalHost().getHostAddress();// 指定本机ip端口，用来给服务端区分,指定端口，重启客户端会等两分钟才能连接上服务端bootstrap.localAddress(hostAddress,0);//连接netty服务器reconnect();}/*** 功能描述: 断线重连，客户端有断线重连机制，就更不能使用异步阻塞了* @param* @return void* @author zhouwenjie* @date 2021/3/19 14:53*/public void reconnect() {ChannelFuture channelFuture = bootstrap.connect();//使用最新的ChannelFuture -> 开启最新的监听器channelFuture.addListener((ChannelFutureListener) future -> {if (future.cause() != null) {log.error("连接失败。。。");future.channel().eventLoop().schedule(() -> reconnect(), 3, TimeUnit.SECONDS);} else {log.info("客户端连接成功。。。");}});}/*** 关闭 client*/@PreDestroypublic void shutdown() {// 优雅关闭 EventLoopGroup 对象eventLoopGroup.shutdownGracefully();log.info("[*Netty客户端关闭]");}
}

处理器

/*** @author: zhouwenjie* @description: 客户端处理类* @create: 2020-04-03 17:45**/
@Component
@Slf4j
@ChannelHandler.Sharable
public class NettyClientHandler extends SimpleChannelInboundHandler<ComputeResult> {/*** 注入NettyClient*/@Autowiredprivate NettyClient nettyClient;/*** 连接成功*/@Overridepublic void channelActive(ChannelHandlerContext ctx) {ComputeResult computeResult = new ComputeResult();computeResult.setComtype("aaaaaa");computeResult.setFromname("bbbbb");computeResult.setJunctionnum("ccccccc");ctx.writeAndFlush(computeResult);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {log.error("[*The netty server suspends service...]");super.channelInactive(ctx);ctx.fireChannelInactive();nettyClient.reconnect();}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {super.exceptionCaught(ctx, cause);log.error("[* Netty connection exception]:{}", cause.toString());cause.printStackTrace();}@Overridepublic void channelRead0(ChannelHandlerContext ctx, ComputeResult computeResult) {System.out.println(computeResult);}
}

注意事项

protostuff的序列化和反序列化特点是一个对象序列化时是按照可序列化字段顺序把值序列化到字节码中，反序列化时也是按照当前对象可序列化字段顺序赋值，如果序列化前的对象和反序列化接收的对象，对应顺序或者字段类型不一样时会出现反序列失败报错。为了避免以上问题，在使用protostuff序列化时，一定要保证客户端和服务端实体类字段顺序一致，且字段类型一致。

案例一：实体类字段顺序不一致

客户端

服务端

如果是这样，客户端传递comtype的值，将会被赋予给服务端的fromnum，因为是按顺序来序列化的。

案例二：实体类字段类型不一致

客户端

服务端

如果是这样，那么服务端反序列化将会报错，报错如下：

所以，一定要保证双方实体类字段的顺序和类型一致。

netty之Protostuff序列化协议相关推荐

微服务架构-高性能Netty服务器-064：Netty实战与反序列化与序列化协议
064:Netty实战与反序列化与序列化协议 1 回顾上节课TCP协议粘包拆分解决方案 2 Java序列化与反序列化的概念 3 Java实现对象持久化操作 4 构建String类型客户与服务器端通讯 ...
Netty序列化协议Protocol buff
序列化协议序列化和反序列化把对象转换为字节序列的过程称为对象的序列化,把字节序列恢复为对象的过程称为对象的反序列化.用途:文件的copy.网络数据的传输 Protocol buff(代替JSON) ...
Netty结合Protostuff传输对象案例，单机压测秒级接收35万个对象
点击上方蓝色"方志朋",选择"设为星标" 回复"666"获取独家整理的学习资料! 单纯netty结合protostuff进行rpc对象传输的 ...
用Netty解析Redis网络协议
用Netty解析Redis网络协议根据Redis官方文档的介绍,学习了一下Redis网络通信协议.然后偶然在GitHub上发现了个用Netty实现的Redis服务器,很有趣,于是就动手实现了一下! ...
JAVA RPC（二）序列化协议杂谈
序列化和反序列化作为Java里一个较为基础的知识点,大家心里也有那么几句要说的,但我相信很多小伙伴掌握的也就是那么几句而已,如果再深究问一下Java如何实现序列化和反序列化的,就可能不知所措了!遥记当 ...
一文彻底理解Redis序列化协议，你也可以编写Redis客户端
前提最近学习Netty的时候想做一个基于Redis服务协议的编码解码模块,过程中顺便阅读了Redis服务序列化协议RESP,结合自己的理解对文档进行了翻译并且简单实现了RESP基于Java语言的解析 ...
招银网络二面：什么是序列化？常见的序列化协议有哪些？
今天分享一道读者面试招银网络科技遇到的面试真题. 下面是正文. 序列化和反序列化相关概念什么是序列化?什么是反序列化? 如果我们需要持久化 Java 对象比如将 Java 对象保存在文件中,或者在网 ...
架构师之路 — 分布式系统 — Protocol Buffers 序列化协议
目录文章目录目录 Protocol Buffers 序列化协议定义数据结构序列化传输 Protocol Buffers 序列化协议 Protocol Buffers 是一种高性能的.语言无关的 ...
Dubbo 序列化协议 5 连问，这谁接得住啊？
作者:小哇说互联来源:www.toutiao.com/i6745361206137061895/ 1)dubbo 支持哪些通信协议? 2)支持哪些序列化协议? 3)说一下 Hessian 的数据结构 ...

netty之Protostuff序列化协议

Protostuff序列化详解

前言

依赖

定义编解码器

工具类

编码器

解码器

整合到netty

注意事项

netty之Protostuff序列化协议相关推荐

最新文章

热门文章