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 static Map<String, Serializer> serializerMap = 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 = serializerMap.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());
+            }
+            serializerMap.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

@@ -82,7 +82,7 @@
  * The type Message codec factory.
  *
  */
-public class MessageCodecFactory {
+public abstract class MessageCodecFactory {
 
     /**
      * The constant UTF8.
@@ -95,7 +95,7 @@ public class MessageCodecFactory {
      * @param abstractMessage the abstract message
      * @return the message codec
      */
-    public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage) {
+    public MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage) {
         return getMessageCodec(abstractMessage.getTypeCode());
     }
 
@@ -105,7 +105,7 @@ public static MessageSeataCodec getMessageCodec(AbstractMessage abstractMessage)
      * @param typeCode the type code
      * @return the msg instance by code
      */
-    public static MessageSeataCodec getMessageCodec(short typeCode) {
+    public MessageSeataCodec getMessageCodec(short typeCode) {
         MessageSeataCodec msgCodec = null;
         switch (typeCode) {
             case MessageType.TYPE_SEATA_MERGE:
@@ -166,7 +166,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 MessageSeataCodec getMergeRequestMessageSeataCodec(int typeCode) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN:
                 return new GlobalBeginRequestCodec();
@@ -195,7 +195,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 MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCode) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN_RESULT:
                 return new GlobalBeginResponseCodec();
@@ -230,7 +230,7 @@ protected static MessageSeataCodec getMergeResponseMessageSeataCodec(int typeCod
      * @param typeCode the type code
      * @return the message
      */
-    public static AbstractMessage getMessage(short typeCode) {
+    public AbstractMessage getMessage(short typeCode) {
         AbstractMessage abstractMessage = null;
         switch (typeCode) {
             case MessageType.TYPE_SEATA_MERGE:
@@ -295,7 +295,7 @@ public static AbstractMessage getMessage(short typeCode) {
      * @param typeCode the type code
      * @return the merge request instance by code
      */
-    protected static AbstractMessage getMergeRequestInstanceByCode(int typeCode) {
+    protected AbstractMessage getMergeRequestInstanceByCode(int typeCode) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN:
                 return new GlobalBeginRequest();
@@ -324,7 +324,7 @@ protected static AbstractMessage getMergeRequestInstanceByCode(int typeCode) {
      * @param typeCode the type code
      * @return the merge response instance by code
      */
-    protected static AbstractMessage getMergeResponseInstanceByCode(int typeCode) {
+    protected AbstractMessage getMergeResponseInstanceByCode(int typeCode) {
         switch (typeCode) {
             case MessageType.TYPE_GLOBAL_BEGIN_RESULT:
                 return new GlobalBeginResponse();

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

@@ -20,38 +20,58 @@
 
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import io.seata.common.exception.NotSupportYetException;
 import io.seata.common.loader.LoadLevel;
+import io.seata.common.loader.Scope;
 import io.seata.common.util.BufferUtils;
 import io.seata.core.protocol.AbstractMessage;
+import io.seata.core.protocol.ProtocolConstants;
 import io.seata.core.serializer.Serializer;
+import io.seata.serializer.seata.protocol.v1.MessageCodecFactoryV1;
 
 /**
  * The Seata codec.
  *
  */
-@LoadLevel(name = "SEATA")
+@LoadLevel(name = "SEATA", scope = Scope.PROTOTYPE)
 public class SeataSerializer implements Serializer {
 
+    MessageCodecFactory factory;
+    byte protocolVersion ;
+
+    public SeataSerializer(Byte version){
+        if (version == ProtocolConstants.VERSION_1) {
+            factory =  new MessageCodecFactoryV1();
+        }else {
+            throw new NotSupportYetException("not support version" + version);
+        }
+        protocolVersion = version;
+    }
     @Override
     public <T> byte[] serialize(T t) {
         if (!(t instanceof AbstractMessage)) {
             throw new IllegalArgumentException("AbstractMessage isn't available.");
         }
         AbstractMessage abstractMessage = (AbstractMessage)t;
-        //typecode
+        //type code
         short typecode = abstractMessage.getTypeCode();
         //msg codec
-        MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typecode);
+        MessageSeataCodec messageCodec = factory.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);
 
-        //typecode + body
-        ByteBuffer byteBuffer = ByteBuffer.allocate(2 + body.length);
-        byteBuffer.putShort(typecode);
+        ByteBuffer byteBuffer;
+        if (protocolVersion == ProtocolConstants.VERSION_0) {
+            byteBuffer = ByteBuffer.allocate(body.length);
+        } else {
+            //typecode + body
+            byteBuffer = ByteBuffer.allocate(2 + body.length);
+            byteBuffer.putShort(typecode);
+        }
         byteBuffer.put(body);
 
         BufferUtils.flip(byteBuffer);
@@ -76,9 +96,9 @@ public <T> T deserialize(byte[] bytes) {
         byteBuffer.get(body);
         ByteBuffer in = ByteBuffer.wrap(body);
         //new Messgae
-        AbstractMessage abstractMessage = MessageCodecFactory.getMessage(typecode);
+        AbstractMessage abstractMessage = factory.getMessage(typecode);
         //get messageCodec
-        MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typecode);
+        MessageSeataCodec messageCodec = factory.getMessageCodec(typecode);
         //decode
         messageCodec.decode(abstractMessage, in);
         return (T)abstractMessage;

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

@@ -27,6 +27,7 @@
 import io.seata.core.protocol.BatchResultMessage;
 import io.seata.serializer.seata.MessageCodecFactory;
 import io.seata.serializer.seata.MessageSeataCodec;
+import io.seata.serializer.seata.protocol.v1.MessageCodecFactoryV1;
 
 /**
  * the type batch result message codec
@@ -35,6 +36,8 @@
  */
 public class BatchResultMessageCodec extends AbstractMessageCodec {
 
+    protected MessageCodecFactory factory = new MessageCodecFactoryV1();
+
     @Override
     public Class<?> getMessageClassType() {
         return BatchResultMessage.class;
@@ -53,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 = factory.getMessageCodec(typeCode);
             messageCodec.encode(msg, subBuffer);
             buffer.writeShort(msg.getTypeCode());
             buffer.writeBytes(subBuffer);
@@ -106,8 +109,8 @@ protected void decode(BatchResultMessage batchResultMessage, ByteBuffer byteBuff
         List<Integer> msgIds = new ArrayList<>();
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
-            AbstractMessage abstractResultMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            AbstractMessage abstractResultMessage = factory.getMessage(typeCode);
+            MessageSeataCodec messageCodec = factory.getMessageCodec(typeCode);
             messageCodec.decode(abstractResultMessage, byteBuffer);
             msgs.add((AbstractResultMessage) abstractResultMessage);
         }

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

@@ -24,13 +24,16 @@
 import io.seata.core.protocol.AbstractMessage;
 import io.seata.core.protocol.AbstractResultMessage;
 import io.seata.core.protocol.MergeResultMessage;
+import io.seata.serializer.seata.protocol.v1.MessageCodecFactoryV1;
 
 /**
  * The type Merge result message codec.
  *
  */
 public class MergeResultMessageCodec extends AbstractMessageCodec {
 
+    protected MessageCodecFactory factory = new MessageCodecFactoryV1();
+
     @Override
     public Class<?> getMessageClassType() {
         return MergeResultMessage.class;
@@ -48,7 +51,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 = factory.getMessageCodec(typeCode);
             messageCodec.encode(msg, out);
         }
 
@@ -90,8 +93,8 @@ protected void decode(MergeResultMessage mergeResultMessage, ByteBuffer byteBuff
         AbstractResultMessage[] msgs = new AbstractResultMessage[msgNum];
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
-            AbstractMessage abstractResultMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            AbstractMessage abstractResultMessage = factory.getMessage(typeCode);
+            MessageSeataCodec messageCodec = factory.getMessageCodec(typeCode);
             messageCodec.decode(abstractResultMessage, byteBuffer);
             msgs[idx] = (AbstractResultMessage)abstractResultMessage;
         }

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

@@ -26,13 +26,16 @@
 import io.seata.serializer.seata.MessageSeataCodec;
 import io.seata.core.protocol.AbstractMessage;
 import io.seata.core.protocol.MergedWarpMessage;
+import io.seata.serializer.seata.protocol.v1.MessageCodecFactoryV1;
 
 /**
  * The type Merged warp message codec.
  *
  */
 public class MergedWarpMessageCodec extends AbstractMessageCodec {
 
+    protected MessageCodecFactory factory = new MessageCodecFactoryV1();
+
     @Override
     public Class<?> getMessageClassType() {
         return MergedWarpMessage.class;
@@ -51,7 +54,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 = factory.getMessageCodec(typeCode);
             messageCodec.encode(msg, subBuffer);
             buffer.writeShort(msg.getTypeCode());
             buffer.writeBytes(subBuffer);
@@ -96,8 +99,8 @@ private void doDecode(MergedWarpMessage mergedWarpMessage, ByteBuffer byteBuffer
         List<AbstractMessage> msgs = new ArrayList<AbstractMessage>();
         for (int idx = 0; idx < msgNum; idx++) {
             short typeCode = byteBuffer.getShort();
-            AbstractMessage abstractMessage = MessageCodecFactory.getMessage(typeCode);
-            MessageSeataCodec messageCodec = MessageCodecFactory.getMessageCodec(typeCode);
+            AbstractMessage abstractMessage = factory.getMessage(typeCode);
+            MessageSeataCodec messageCodec = factory.getMessageCodec(typeCode);
             messageCodec.decode(abstractMessage, byteBuffer);
             msgs.add(abstractMessage);
         }