in core/src/main/java/org/apache/rocketmq/streams/core/running/AbstractProcessor.java [126:151]
public <V> V byte2Object(byte[] bytes) throws Throwable {
if (bytes == null || bytes.length == 0) {
return null;
}
ByteBuf byteBuf = Unpooled.wrappedBuffer(bytes);
int classNameLength = byteBuf.readInt();
ByteBuf classNameBuf = byteBuf.readBytes(classNameLength);
byte[] clazzNameBytes = new byte[classNameBuf.readableBytes()];
classNameBuf.readBytes(clazzNameBytes);
//实例化
String className = new String(clazzNameBytes, StandardCharsets.UTF_8);
Class<V> clazz = (Class<V>)Class.forName(className);
int objectLength = byteBuf.readInt();
ByteBuf objBuf = byteBuf.readBytes(objectLength);
byte[] objectBytes = new byte[objectLength];
objBuf.readBytes(objectBytes);
classNameBuf.release();
objBuf.release();
return Utils.byte2Object(objectBytes, clazz);
}