optimize: load SeataSerializer by version

core/src/main/java/io/seata/core/protocol/ProtocolConstants.java

@@ -31,10 +31,20 @@ public interface ProtocolConstants {
      */
     byte[] MAGIC_CODE_BYTES = {(byte) 0xda, (byte) 0xda};
 
+    /**
+     * Old protocol version
+     */
+    byte VERSION_0 = 0;
+
+    /**
+     * Protocol version
+     */
+    byte VERSION_1 = 1;
+
     /**
      * Protocol version
      */
-    byte VERSION = 1;
+    byte VERSION = VERSION_1;
 
     /**
      * Max frame length

core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Decoder.java

@@ -142,7 +142,7 @@ public Object decodeFrame(ByteBuf frame) {
                 frame.readBytes(bs);
                 Compressor compressor = CompressorFactory.getCompressor(compressorType);
                 bs = compressor.decompress(bs);
-                Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()));
+                Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()), ProtocolConstants.VERSION_1);
                 rpcMessage.setBody(serializer.deserialize(bs));
             }
         }

core/src/main/java/io/seata/core/rpc/netty/v1/ProtocolV1Encoder.java

@@ -92,7 +92,7 @@ public void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) {
                 if (messageType != ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST
                         && messageType != ProtocolConstants.MSGTYPE_HEARTBEAT_RESPONSE) {
                     // heartbeat has no body
-                    Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()));
+                    Serializer serializer = SerializerServiceLoader.load(SerializerType.getByCode(rpcMessage.getCodec()), ProtocolConstants.VERSION_1);
                     bodyBytes = serializer.serialize(rpcMessage.getBody());
                     Compressor compressor = CompressorFactory.getCompressor(rpcMessage.getCompressor());
                     bodyBytes = compressor.compress(bodyBytes);

core/src/main/java/io/seata/core/serializer/SerializerServiceLoader.java

@@ -20,15 +20,18 @@
 import io.seata.common.loader.EnhancedServiceNotFoundException;
 import io.seata.common.util.ReflectionUtil;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * The Service Loader for the interface {@link Serializer}
- *
  */
 public final class SerializerServiceLoader {
 
     private SerializerServiceLoader() {
     }
 
+    private final static Map<String, Serializer> SERIALIZER_MAP = new HashMap<>();
 
     private static final String PROTOBUF_SERIALIZER_CLASS_NAME = "io.seata.serializer.protobuf.ProtobufSerializer";
 
@@ -39,7 +42,7 @@ private SerializerServiceLoader() {
      * @return the service of {@link Serializer}
      * @throws EnhancedServiceNotFoundException the enhanced service not found exception
      */
-    public static Serializer load(SerializerType type) throws EnhancedServiceNotFoundException {
+    public static Serializer load(SerializerType type, byte version) throws EnhancedServiceNotFoundException {
         if (type == SerializerType.PROTOBUF) {
             try {
                 ReflectionUtil.getClassByName(PROTOBUF_SERIALIZER_CLASS_NAME);
@@ -48,6 +51,24 @@ public static Serializer load(SerializerType type) throws EnhancedServiceNotFoun
                         "Please manually reference 'io.seata:seata-serializer-protobuf' dependency ", e);
             }
         }
-        return EnhancedServiceLoader.load(Serializer.class, type.name());
+
+        String key = serialzerKey(type, version);
+        Serializer serializer = SERIALIZER_MAP.get(key);
+        if (serializer == null) {
+            if (type == SerializerType.SEATA) {
+                serializer = EnhancedServiceLoader.load(Serializer.class, type.name(), new Object[]{version});
+            } else {
+                serializer = EnhancedServiceLoader.load(Serializer.class, type.name());
+            }
+            SERIALIZER_MAP.put(key, serializer);
+        }
+        return serializer;
+    }
+
+    private static String serialzerKey(SerializerType type, byte version) {
+        if (type == SerializerType.SEATA) {
+            return type.name() + version;
+        }
+        return type.name();
     }
-}
+}

serializer/seata-serializer-seata/src/main/java/io/seata/serializer/seata/MessageCodecFactory.java

@@ -80,7 +80,6 @@
 
 /**
  * The type Message codec factory.
- *
  */
 public class MessageCodecFactory {
 
@@ -95,8 +94,8 @@ public class MessageCodecFactory {
      * @param abstractMessage the abstract message
      * @return the message codec
      */
-    public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage) {
-        return getMessageCodec(abstractMessage.getTypeCode());
+    public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage, byte version) {
+        return getMessageCodec(abstractMessage.getTypeCode(), version);
     }
 
     /**
@@ -105,14 +104,14 @@ public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage)
      * @param typeCode the type code
      * @return the msg instance by code
      */
-    public static MessageSeataCodec getMessageCodec(short typeCode) {
+    public static MessageSeataCodec getMessageCodec(short typeCode, byte version) {
         MessageSeataCodec msgCodec = null;
         switch (typeCode) {
             case MessageType.TYPE_SEATA_MERGE:
-                msgCodec = new MergedWarpMessageCodec();
+                msgCodec = new MergedWarpMessageCodec(version);
                 break;
             case MessageType.TYPE_SEATA_MERGE_RESULT:
-                msgCodec = new MergeResultMessageCodec();
+                msgCodec = new MergeResultMessageCodec(version);
                 break;
             case MessageType.TYPE_REG_CLT:
                 msgCodec = new RegisterTMRequestCodec();
@@ -136,7 +135,7 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
                 msgCodec = new GlobalReportRequestCodec();
                 break;
             case MessageType.TYPE_BATCH_RESULT_MSG:
-                msgCodec = new BatchResultMessageCodec();
+                msgCodec = new BatchResultMessageCodec(version);
                 break;
             default:
                 break;
@@ -147,15 +146,15 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
         }
 
         try {
-            msgCodec = getMergeRequestMessageSeataCodec(typeCode);
+            msgCodec = getMergeRequestMessageSeataCodec(typeCode, version);
         } catch (Exception exx) {
         }
 
         if (msgCodec != null) {
             return msgCodec;
         }
 
-        msgCodec = getMergeResponseMessageSeataCodec(typeCode);
+        msgCodec = getMergeResponseMessageSeataCodec(typeCode, version);
 
         return msgCodec;
     }
@@ -166,7 +165,7 @@ public static MessageSeataCodec getMessageCodec(short typeCode) {
      * @param typeCode the type code
      * @return the merge request instance by code
      */
-    protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode) {
+    protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode, byte version) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN:
                 return new GlobalBeginRequestCodec();
@@ -195,7 +194,7 @@ protected static MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode
      * @param typeCode the type code
      * @return the merge response instance by code
      */
-    protected static MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCode) {
+    protected static MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCode, byte version) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN_RESULT:
                 return new GlobalBeginResponseCodec();

serializer/seata-serializer-seata/src/main/java/io/seata/serializer/seata/SeataSerializer.java

@@ -16,72 +16,41 @@
  */
 package io.seata.serializer.seata;
 
-import java.nio.ByteBuffer;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.Unpooled;
 import io.seata.common.loader.LoadLevel;
-import io.seata.common.util.BufferUtils;
-import io.seata.core.protocol.AbstractMessage;
+import io.seata.common.loader.Scope;
+import io.seata.core.protocol.ProtocolConstants;
 import io.seata.core.serializer.Serializer;
+import io.seata.serializer.seata.protocol.v0.SeataSerializerV0;
+import io.seata.serializer.seata.protocol.v1.SeataSerializerV1;
 
 /**
  * The Seata codec.
- *
  */
-@LoadLevel(name = "SEATA")
+@LoadLevel(name = "SEATA", scope = Scope.PROTOTYPE)
 public class SeataSerializer implements Serializer {
 
-    @Override
-    public <T> byte[] serialize(T t) {
-        if (!(t instanceof AbstractMessage)) {
-            throw new IllegalArgumentException("AbstractMessage isn't available.");
-        }
-        AbstractMessage abstractMessage = (AbstractMessage)t;
-        //typecode
-        short typecode = abstractMessage.getTypeCode();
-        //msg codec
-        MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typecode);
-        //get empty ByteBuffer
-        ByteBuf out = Unpooled.buffer(1024);
-        //msg encode
-        messageCodec.encode(t, out);
-        byte[] body = new byte[out.readableBytes()];
-        out.readBytes(body);
+    Serializer versionSeataSerializer;
 
-        //typecode + body
-        ByteBuffer byteBuffer = ByteBuffer.allocate(2 + body.length);
-        byteBuffer.putShort(typecode);
-        byteBuffer.put(body);
 
-        BufferUtils.flip(byteBuffer);
-        byte[] content = new byte[byteBuffer.limit()];
-        byteBuffer.get(content);
-        return content;
+    public SeataSerializer(Byte version) {
+        if (version == ProtocolConstants.VERSION_0) {
+            versionSeataSerializer = SeataSerializerV0.getInstance();
+        } else if (version == ProtocolConstants.VERSION_1) {
+            versionSeataSerializer = SeataSerializerV1.getInstance();
+        }
+        if (versionSeataSerializer == null) {
+            throw new IllegalArgumentException("version is not supported");
+        }
+    }
+
+    @Override
+    public <T> byte[] serialize(T t) {
+        return versionSeataSerializer.serialize(t);
     }
 
     @Override
     public <T> T deserialize(byte[] bytes) {
-        if (bytes == null || bytes.length == 0) {
-            throw new IllegalArgumentException("Nothing to decode.");
-        }
-        if (bytes.length < 2) {
-            throw new IllegalArgumentException("The byte[] isn't available for decode.");
-        }
-        ByteBuffer byteBuffer = ByteBuffer.wrap(bytes);
-        //typecode
-        short typecode = byteBuffer.getShort();
-        //msg body
-        byte[] body = new byte[byteBuffer.remaining()];
-        byteBuffer.get(body);
-        ByteBuffer in = ByteBuffer.wrap(body);
-        //new Messgae
-        AbstractMessage abstractMessage = MessageCodecFactory.getMessage(typecode);
-        //get messageCodec
-        MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typecode);
-        //decode
-        messageCodec.decode(abstractMessage, in);
-        return (T)abstractMessage;
+        return versionSeataSerializer.deserialize(bytes);
     }
 
 }

serializer/seata-serializer-seata/src/main/java/io/seata/serializer/seata/protocol/BatchResultMessageCodec.java

@@ -35,6 +35,11 @@
  */
 public class BatchResultMessageCodec extends AbstractMessageCodec {
 
+    private byte version;
+
+    public BatchResultMessageCodec(byte version) {
+        this.version = version;
+    }
     @Override
     public Class<?> getMessageClassType() {
         return BatchResultMessage.class;
@@ -53,7 +58,7 @@ public <T> void encode(T t, ByteBuf out) {
         for (final AbstractMessage msg : msgs) {
             final ByteBuf subBuffer = Unpooled.buffer(1024);
             short typeCode = msg.getTypeCode();
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.encode(msg, subBuffer);
             buffer.writeShort(msg.getTypeCode());
             buffer.writeBytes(subBuffer);
@@ -107,7 +112,7 @@ protected void decode(BatchResultMessage batchResultMessage, ByteBuffer byteBuff
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
             AbstractMessage abstractResultMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.decode(abstractResultMessage, byteBuffer);
             msgs.add((AbstractResultMessage) abstractResultMessage);
         }

serializer/seata-serializer-seata/src/main/java/io/seata/serializer/seata/protocol/MergeResultMessageCodec.java

@@ -31,6 +31,11 @@
  */
 public class MergeResultMessageCodec extends AbstractMessageCodec {
 
+    private byte version;
+
+    public MergeResultMessageCodec(byte version) {
+        this.version = version;
+    }
     @Override
     public Class<?> getMessageClassType() {
         return MergeResultMessage.class;
@@ -48,7 +53,7 @@ public <T> void encode(T t, ByteBuf out) {
             short typeCode = msg.getTypeCode();
             //put typeCode
             out.writeShort(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.encode(msg, out);
         }
 
@@ -91,7 +96,7 @@ protected void decode(MergeResultMessage mergeResultMessage, ByteBuffer byteBuff
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
             AbstractMessage abstractResultMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.decode(abstractResultMessage, byteBuffer);
             msgs[idx] = (AbstractResultMessage)abstractResultMessage;
         }

serializer/seata-serializer-seata/src/main/java/io/seata/serializer/seata/protocol/MergedWarpMessageCodec.java

@@ -33,6 +33,11 @@
  */
 public class MergedWarpMessageCodec extends AbstractMessageCodec {
 
+    private byte version;
+
+    public MergedWarpMessageCodec(byte version) {
+        this.version = version;
+    }
     @Override
     public Class<?> getMessageClassType() {
         return MergedWarpMessage.class;
@@ -51,7 +56,7 @@ public <T> void encode(T t, ByteBuf out) {
         for (final AbstractMessage msg : msgs) {
             final ByteBuf subBuffer = Unpooled.buffer(1024);
             short typeCode = msg.getTypeCode();
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.encode(msg, subBuffer);
             buffer.writeShort(msg.getTypeCode());
             buffer.writeBytes(subBuffer);
@@ -97,7 +102,7 @@ private void doDecode(MergedWarpMessage mergedWarpMessage, ByteBuffer byteBuffer
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
             AbstractMessage abstractMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode, version);
             messageCodec.decode(abstractMessage, byteBuffer);
             msgs.add(abstractMessage);
         }