序

本文主要研究一下storm tuple的序列化

ExecutorTransfer.tryTransfer

storm-2.0.0/storm-client/src/jvm/org/apache/storm/executor/ExecutorTransfer.java

// Every executor has an instance of this classpublic class ExecutorTransfer {    private static final Logger LOG = LoggerFactory.getLogger(ExecutorTransfer.class);    private final WorkerState workerData;    private final KryoTupleSerializer serializer;    private final boolean isDebug;    private int indexingBase = 0;    private ArrayList
    
      localReceiveQueues; // [taskId-indexingBase] => queue : List of all recvQs local to this worker    private AtomicReferenceArray
     
       queuesToFlush;        // [taskId-indexingBase] => queue, some entries can be null. : outbound Qs for this executor instance    public ExecutorTransfer(WorkerState workerData, Map
      
        topoConf) {        this.workerData = workerData;        this.serializer = new KryoTupleSerializer(topoConf, workerData.getWorkerTopologyContext());        this.isDebug = ObjectReader.getBoolean(topoConf.get(Config.TOPOLOGY_DEBUG), false);    }    //......    // adds addressedTuple to destination Q if it is not full. else adds to pendingEmits (if its not null)    public boolean tryTransfer(AddressedTuple addressedTuple, Queue
       
         pendingEmits) {        if (isDebug) {            LOG.info("TRANSFERRING tuple {}", addressedTuple);        }        JCQueue localQueue = getLocalQueue(addressedTuple);        if (localQueue != null) {            return tryTransferLocal(addressedTuple, localQueue, pendingEmits);        }        return workerData.tryTransferRemote(addressedTuple, pendingEmits, serializer);    }    //......}
       
      
     
    

• ExecutorTransfer在构造器里头创建了KryoTupleSerializer
• 这里先判断目标地址是否是在localQueue中，如果是则进行local transfer，否则进行remote transfer
• remote transfer的时候调用了workerData.tryTransferRemote，并传递了serializer

WorkerState.tryTransferRemote

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java

/* Not a Blocking call. If cannot emit, will add 'tuple' to pendingEmits and return 'false'. 'pendingEmits' can be null */    public boolean tryTransferRemote(AddressedTuple tuple, Queue
    
      pendingEmits, ITupleSerializer serializer) {        return workerTransfer.tryTransferRemote(tuple, pendingEmits, serializer);    }
    

• WorkerState.tryTransferRemote实际上使用的是workerTransfer.tryTransferRemote

workerTransfer.tryTransferRemote

storm-2.0.0/storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerTransfer.java

/* Not a Blocking call. If cannot emit, will add 'tuple' to 'pendingEmits' and return 'false'. 'pendingEmits' can be null */    public boolean tryTransferRemote(AddressedTuple addressedTuple, Queue
    
      pendingEmits, ITupleSerializer serializer) {        if (pendingEmits != null && !pendingEmits.isEmpty()) {            pendingEmits.add(addressedTuple);            return false;        }        if (!remoteBackPressureStatus[addressedTuple.dest].get()) {            TaskMessage tm = new TaskMessage(addressedTuple.getDest(), serializer.serialize(addressedTuple.getTuple()));            if (transferQueue.tryPublish(tm)) {                return true;            }        } else {            LOG.debug("Noticed Back Pressure in remote task {}", addressedTuple.dest);        }        if (pendingEmits != null) {            pendingEmits.add(addressedTuple);        }        return false;    }
    

• 这里可以看到创建TaskMessage的时候，使用serializer.serialize(addressedTuple.getTuple())对tuple进行了序列化；该serializer为ITupleSerializer类型，它的实现类为KryoTupleSerializer

KryoTupleSerializer

storm-2.0.0/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java

public class KryoTupleSerializer implements ITupleSerializer {    KryoValuesSerializer _kryo;    SerializationFactory.IdDictionary _ids;    Output _kryoOut;    public KryoTupleSerializer(final Map
    
      conf, final GeneralTopologyContext context) {        _kryo = new KryoValuesSerializer(conf);        _kryoOut = new Output(2000, 2000000000);        _ids = new SerializationFactory.IdDictionary(context.getRawTopology());    }    public byte[] serialize(Tuple tuple) {        try {            _kryoOut.clear();            _kryoOut.writeInt(tuple.getSourceTask(), true);            _kryoOut.writeInt(_ids.getStreamId(tuple.getSourceComponent(), tuple.getSourceStreamId()), true);            tuple.getMessageId().serialize(_kryoOut);            _kryo.serializeInto(tuple.getValues(), _kryoOut);            return _kryoOut.toBytes();        } catch (IOException e) {            throw new RuntimeException(e);        }    }    //    public long crc32(Tuple tuple) {    //        try {    //            CRC32OutputStream hasher = new CRC32OutputStream();    //            _kryo.serializeInto(tuple.getValues(), hasher);    //            return hasher.getValue();    //        } catch (IOException e) {    //            throw new RuntimeException(e);    //        }    //    }}
    

• KryoTupleSerializer创建了KryoValuesSerializer，在serialize tuple的时候调用了_kryo.serializeInto(tuple.getValues(), _kryoOut)

KryoValuesSerializer

storm-2.0.0/storm-client/src/jvm/org/apache/storm/serialization/KryoValuesSerializer.java

public class KryoValuesSerializer {    Kryo _kryo;    ListDelegate _delegate;    Output _kryoOut;    public KryoValuesSerializer(Map
    
      conf) {        _kryo = SerializationFactory.getKryo(conf);        _delegate = new ListDelegate();        _kryoOut = new Output(2000, 2000000000);    }    public void serializeInto(List
     
    

• KryoValuesSerializer在构造器里头调用SerializationFactory.getKryo(conf)方法创建_kryo
• 这里的_delegate使用的是ListDelegate(即用它来包装一下List<Object> values)，_kryoOut为new Output(2000, 2000000000)
• serialize方法调用的是serializeInto方法，该方法最后调用的是原生的_kryo.writeObject方法进行序列化

SerializationFactory.getKryo

storm-2.0.0/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java

public static Kryo getKryo(Map
    
      conf) {        IKryoFactory kryoFactory = (IKryoFactory) ReflectionUtils.newInstance((String) conf.get(Config.TOPOLOGY_KRYO_FACTORY));        Kryo k = kryoFactory.getKryo(conf);        k.register(byte[].class);        /* tuple payload serializer is specified via configuration */        String payloadSerializerName = (String) conf.get(Config.TOPOLOGY_TUPLE_SERIALIZER);        try {            Class serializerClass = Class.forName(payloadSerializerName);            Serializer serializer = resolveSerializerInstance(k, ListDelegate.class, serializerClass, conf);            k.register(ListDelegate.class, serializer);        } catch (ClassNotFoundException ex) {            throw new RuntimeException(ex);        }        k.register(ArrayList.class, new ArrayListSerializer());        k.register(HashMap.class, new HashMapSerializer());        k.register(HashSet.class, new HashSetSerializer());        k.register(BigInteger.class, new BigIntegerSerializer());        k.register(TransactionAttempt.class);        k.register(Values.class);        k.register(org.apache.storm.metric.api.IMetricsConsumer.DataPoint.class);        k.register(org.apache.storm.metric.api.IMetricsConsumer.TaskInfo.class);        k.register(ConsList.class);        k.register(BackPressureStatus.class);        synchronized (loader) {            for (SerializationRegister sr : loader) {                try {                    sr.register(k);                } catch (Exception e) {                    throw new RuntimeException(e);                }            }        }        kryoFactory.preRegister(k, conf);        boolean skipMissing = (Boolean) conf.get(Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS);        register(k, conf.get(Config.TOPOLOGY_KRYO_REGISTER), conf, skipMissing);        kryoFactory.postRegister(k, conf);        if (conf.get(Config.TOPOLOGY_KRYO_DECORATORS) != null) {            for (String klassName : (List
     
      ) conf.get(Config.TOPOLOGY_KRYO_DECORATORS)) {                try {                    Class klass = Class.forName(klassName);                    IKryoDecorator decorator = (IKryoDecorator) klass.newInstance();                    decorator.decorate(k);                } catch (ClassNotFoundException e) {                    if (skipMissing) {                        LOG.info("Could not find kryo decorator named " + klassName + ". Skipping registration...");                    } else {                        throw new RuntimeException(e);                    }                } catch (InstantiationException e) {                    throw new RuntimeException(e);                } catch (IllegalAccessException e) {                    throw new RuntimeException(e);                }            }        }        kryoFactory.postDecorate(k, conf);        return k;    }    public static void register(Kryo k, Object kryoRegistrations, Map
      
        conf, boolean skipMissing) {        Map
       
         registrations = normalizeKryoRegister(kryoRegistrations);        for (Map.Entry
        
          entry : registrations.entrySet()) {            String serializerClassName = entry.getValue();            try {                Class klass = Class.forName(entry.getKey());                Class serializerClass = null;                if (serializerClassName != null) {                    serializerClass = Class.forName(serializerClassName);                }                if (serializerClass == null) {                    k.register(klass);                } else {                    k.register(klass, resolveSerializerInstance(k, klass, serializerClass, conf));                }            } catch (ClassNotFoundException e) {                if (skipMissing) {                    LOG.info("Could not find serialization or class for " + serializerClassName + ". Skipping registration...");                } else {                    throw new RuntimeException(e);                }            }        }    }
        
       
      
     
    

• SerializationFactory.getKryo静态方法首先根据Config.TOPOLOGY_KRYO_FACTORY创建IKryoFactory，默认是org.apache.storm.serialization.DefaultKryoFactory
• 之后通过IKryoFactory.getKryo创建Kryo，之后就是对Kryo进行一系列配置，这里注册了byte[].class、ListDelegate.class、ArrayList.class、HashMap.class、HashSet.class、BigInteger.class、TransactionAttempt.class、Values.class、org.apache.storm.metric.api.IMetricsConsumer.DataPoint.class、org.apache.storm.metric.api.IMetricsConsumer.TaskInfo.class、ConsList.class、BackPressureStatus.class
• ListDelegate.class为payload的容器，采用Config.TOPOLOGY_TUPLE_SERIALIZER(topology.tuple.serializer，默认是org.apache.storm.serialization.types.ListDelegateSerializer)配置的类进行序列化
• Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS(topology.skip.missing.kryo.registrations，默认为false)，当kryo找不到配置的要序列化的class对应serializers的时候，是抛出异常还是直接跳过注册；
• 最后通过Config.TOPOLOGY_KRYO_DECORATORS(topology.kryo.decorators)加载自定义的serialization

DefaultKryoFactory

storm-2.0.0/storm-client/src/jvm/org/apache/storm/serialization/DefaultKryoFactory.java

public class DefaultKryoFactory implements IKryoFactory {    @Override    public Kryo getKryo(Map
    
      conf) {        KryoSerializableDefault k = new KryoSerializableDefault();        k.setRegistrationRequired(!((Boolean) conf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION)));        k.setReferences(false);        return k;    }    @Override    public void preRegister(Kryo k, Map
     
       conf) {    }    public void postRegister(Kryo k, Map
      
        conf) {        ((KryoSerializableDefault) k).overrideDefault(true);    }    @Override    public void postDecorate(Kryo k, Map
       
         conf) {    }    public static class KryoSerializableDefault extends Kryo {        boolean _override = false;        public void overrideDefault(boolean value) {            _override = value;        }        @Override        public Serializer getDefaultSerializer(Class type) {            if (_override) {                return new SerializableSerializer();            } else {                return super.getDefaultSerializer(type);            }        }    }}
       
      
     
    

• 这里从配置读取Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION(topology.fall.back.on.java.serialization)，默认该值为true，则registrationRequired这里设置为false，即序列化的时候不要求该class必须在已注册的列表中

Kryo

kryo-4.0.2-sources.jar!/com/esotericsoftware/kryo/Kryo.java

/** If the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is false, it is automatically registered	 * using the {@link Kryo#addDefaultSerializer(Class, Class) default serializer}.	 * @throws IllegalArgumentException if the class is not registered and {@link Kryo#setRegistrationRequired(boolean)} is true.	 * @see ClassResolver#getRegistration(Class) */	public Registration getRegistration (Class type) {		if (type == null) throw new IllegalArgumentException("type cannot be null.");		Registration registration = classResolver.getRegistration(type);		if (registration == null) {			if (Proxy.isProxyClass(type)) {				// If a Proxy class, treat it like an InvocationHandler because the concrete class for a proxy is generated.				registration = getRegistration(InvocationHandler.class);			} else if (!type.isEnum() && Enum.class.isAssignableFrom(type) && !Enum.class.equals(type)) {				// This handles an enum value that is an inner class. Eg: enum A {b{}};				registration = getRegistration(type.getEnclosingClass());			} else if (EnumSet.class.isAssignableFrom(type)) {				registration = classResolver.getRegistration(EnumSet.class);			} else if (isClosure(type)) {				registration = classResolver.getRegistration(ClosureSerializer.Closure.class);			}			if (registration == null) {				if (registrationRequired) {					throw new IllegalArgumentException(unregisteredClassMessage(type));				}				if (warnUnregisteredClasses) {					warn(unregisteredClassMessage(type));				}				registration = classResolver.registerImplicit(type);			}		}		return registration;	}	/** Registers the class using the lowest, next available integer ID and the {@link Kryo#getDefaultSerializer(Class) default	 * serializer}. If the class is already registered, no change will be made and the existing registration will be returned.	 * Registering a primitive also affects the corresponding primitive wrapper.	 * 
	 * Because the ID assigned is affected by the IDs registered before it, the order classes are registered is important when	 * using this method. The order must be the same at deserialization as it was for serialization. */	public Registration register (Class type) {		Registration registration = classResolver.getRegistration(type);		if (registration != null) return registration;		return register(type, getDefaultSerializer(type));	}	/** Returns the best matching serializer for a class. This method can be overridden to implement custom logic to choose a	 * serializer. */	public Serializer getDefaultSerializer (Class type) {		if (type == null) throw new IllegalArgumentException("type cannot be null.");		final Serializer serializerForAnnotation = getDefaultSerializerForAnnotatedType(type);		if (serializerForAnnotation != null) return serializerForAnnotation;		for (int i = 0, n = defaultSerializers.size(); i < n; i++) {			DefaultSerializerEntry entry = defaultSerializers.get(i);			if (entry.type.isAssignableFrom(type)) {				Serializer defaultSerializer = entry.serializerFactory.makeSerializer(this, type);				return defaultSerializer;			}		}		return newDefaultSerializer(type);	}	/** Called by {@link #getDefaultSerializer(Class)} when no default serializers matched the type. Subclasses can override this	 * method to customize behavior. The default implementation calls {@link SerializerFactory#makeSerializer(Kryo, Class)} using	 * the {@link #setDefaultSerializer(Class) default serializer}. */	protected Serializer newDefaultSerializer (Class type) {		return defaultSerializer.makeSerializer(this, type);	}	/** Registers the class using the lowest, next available integer ID and the specified serializer. If the class is already	 * registered, the existing entry is updated with the new serializer. Registering a primitive also affects the corresponding	 * primitive wrapper.	 * 
	 * Because the ID assigned is affected by the IDs registered before it, the order classes are registered is important when	 * using this method. The order must be the same at deserialization as it was for serialization. */	public Registration register (Class type, Serializer serializer) {		Registration registration = classResolver.getRegistration(type);		if (registration != null) {			registration.setSerializer(serializer);			return registration;		}		return classResolver.register(new Registration(type, serializer, getNextRegistrationId()));	}	/** Returns the lowest, next available integer ID. */	public int getNextRegistrationId () {		while (nextRegisterID != -2) {			if (classResolver.getRegistration(nextRegisterID) == null) return nextRegisterID;			nextRegisterID++;		}		throw new KryoException("No registration IDs are available.");	}

• Kryo的getRegistration方法，当遇到class没有注册时会判断registrationRequired，如果为true，则抛出IllegalArgumentException；如果为false，则调用classResolver.registerImplicit进行隐式注册，同时如果warnUnregisteredClasses为true则会打印warning信息
• Kryo的register方法如果没有指定Serializer时，会通过getDefaultSerializer获取最匹配的Serializer，如果从已经注册的defaultSerializers没匹配到，则调用newDefaultSerializer创建一个，这里可能存在无法创建的异常，会抛出IllegalArgumentException
• register(Class type, Serializer serializer)方法最后是调用ClassResolver.register(Registration registration)方法，对于没有Registration的，这里new了一个，同时通过getNextRegistrationId，给Registration分配一个id

DefaultClassResolver.register

kryo-4.0.2-sources.jar!/com/esotericsoftware/kryo/util/DefaultClassResolver.java

static public final byte NAME = -1;	protected final IntMap
    
      idToRegistration = new IntMap();	protected final ObjectMap
     
       classToRegistration = new ObjectMap();	protected IdentityObjectIntMap
      
        classToNameId;	public Registration registerImplicit (Class type) {		return register(new Registration(type, kryo.getDefaultSerializer(type), NAME));	}	public Registration register (Registration registration) {		if (registration == null) throw new IllegalArgumentException("registration cannot be null.");		if (registration.getId() != NAME) {			if (TRACE) {				trace("kryo", "Register class ID " + registration.getId() + ": " + className(registration.getType()) + " ("					+ registration.getSerializer().getClass().getName() + ")");			}			idToRegistration.put(registration.getId(), registration);		} else if (TRACE) {			trace("kryo", "Register class name: " + className(registration.getType()) + " ("				+ registration.getSerializer().getClass().getName() + ")");		}		classToRegistration.put(registration.getType(), registration);		if (registration.getType().isPrimitive()) classToRegistration.put(getWrapperClass(registration.getType()), registration);		return registration;	}	public Registration writeClass (Output output, Class type) {		if (type == null) {			if (TRACE || (DEBUG && kryo.getDepth() == 1)) log("Write", null);			output.writeVarInt(Kryo.NULL, true);			return null;		}		Registration registration = kryo.getRegistration(type);		if (registration.getId() == NAME)			writeName(output, type, registration);		else {			if (TRACE) trace("kryo", "Write class " + registration.getId() + ": " + className(type));			output.writeVarInt(registration.getId() + 2, true);		}		return registration;	}	protected void writeName (Output output, Class type, Registration registration) {		output.writeVarInt(NAME + 2, true);		if (classToNameId != null) {			int nameId = classToNameId.get(type, -1);			if (nameId != -1) {				if (TRACE) trace("kryo", "Write class name reference " + nameId + ": " + className(type));				output.writeVarInt(nameId, true);				return;			}		}		// Only write the class name the first time encountered in object graph.		if (TRACE) trace("kryo", "Write class name: " + className(type));		int nameId = nextNameId++;		if (classToNameId == null) classToNameId = new IdentityObjectIntMap();		classToNameId.put(type, nameId);		output.writeVarInt(nameId, true);		output.writeString(type.getName());	}	public void reset () {		if (!kryo.isRegistrationRequired()) {			if (classToNameId != null) classToNameId.clear(2048);			if (nameIdToClass != null) nameIdToClass.clear();			nextNameId = 0;		}	}
      
     
    

• DefaultClassResolver.register(Registration registration)方法里头针对registration的id进行了判断，如果是NAME(这里用-1表示)则注册到ObjectMap<Class, Registration> classToRegistration，如果有id不是NAME的，则注册到IntMap<Registration> idToRegistration
• 前面提到如果registrationRequired是false，则调用classResolver.registerImplicit进行隐式注册，这里可以看到registerImplicit注册的registration的id是NAME
• registration的id是NAME与否具体在writeClass中有体现(如果要序列化的类的字段中不仅仅有基本类型，还有未注册的类，会调用这里的writeClass方法)，从代码可以看到如果是NAME，则使用的是writeName；不是NAME的则直接使用output.writeVarInt(registration.getId() + 2, true)，写入int；writeName方法第一次遇到NAME的class时会给它生成一个nameId，然后放入到IdentityObjectIntMap<Class> classToNameId中，然后写入int，再写入class.getName，第二次遇到该class的时候，由于classToNameId中已经存在nameId，因而直接写入int；但是DefaultClassResolver的reset方法在registrationRequired是false这种情况下会调用classToNameId.clear(2048)，进行清空或者resize，这个时候一旦这个方法被调用，那么下次可能无法利用classToNameId用id替代className来序列化。

Kryo.writeObject

kryo-4.0.2-sources.jar!/com/esotericsoftware/kryo/Kryo.java

/** Writes an object using the registered serializer. */	public void writeObject (Output output, Object object) {		if (output == null) throw new IllegalArgumentException("output cannot be null.");		if (object == null) throw new IllegalArgumentException("object cannot be null.");		beginObject();		try {			if (references && writeReferenceOrNull(output, object, false)) {				getRegistration(object.getClass()).getSerializer().setGenerics(this, null);				return;			}			if (TRACE || (DEBUG && depth == 1)) log("Write", object);			getRegistration(object.getClass()).getSerializer().write(this, output, object);		} finally {			if (--depth == 0 && autoReset) reset();		}	}	/** Resets unregistered class names, references to previously serialized or deserialized objects, and the	 * {@link #getGraphContext() graph context}. If {@link #setAutoReset(boolean) auto reset} is true, this method is called	 * automatically when an object graph has been completely serialized or deserialized. If overridden, the super method must be	 * called. */	public void reset () {		depth = 0;		if (graphContext != null) graphContext.clear();		classResolver.reset();		if (references) {			referenceResolver.reset();			readObject = null;		}		copyDepth = 0;		if (originalToCopy != null) originalToCopy.clear(2048);		if (TRACE) trace("kryo", "Object graph complete.");	}

• 这里要注意一下，writeObject方法在finally的时候判断如果depth为0且autoReset为true，会调用reset方法；而reset方法会调用classResolver.reset()，清空nameIdToClass以及classToNameId(classToNameId.clear(2048))

小结

• storm默认是用kryo来进行tuple的序列化，storm额外注册了byte[].class、ListDelegate.class、ArrayList.class、HashMap.class、HashSet.class、BigInteger.class、TransactionAttempt.class、Values.class、org.apache.storm.metric.api.IMetricsConsumer.DataPoint.class、org.apache.storm.metric.api.IMetricsConsumer.TaskInfo.class、ConsList.class、BackPressureStatus.class等类型
• Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION(topology.fall.back.on.java.serialization)如果为true，则kryo.setRegistrationRequired(false)，也就是如果一个class没有在kryo进行注册，不会抛异常；这个命名可能存在歧义(不是使用java自身的序列化机制来进行fallback)，它实际上要表达的是对于遇到没有注册的class要不要fallback，如果不fallback则直接抛异常，如果fallback，则会进行隐式注册，在classToNameId不会被reset的前提下，第一次使用className来序列化，同时分配一个id写入classToNameId，第二次则直接使用classToNameId中获取到的id，也就相当于手工注册的效果
• Config.TOPOLOGY_TUPLE_SERIALIZER(topology.tuple.serializer，默认是org.apache.storm.serialization.types.ListDelegateSerializer)用于配置tuple的payload的序列化类
• Config.TOPOLOGY_KRYO_DECORATORS(topology.kryo.decorators)用于加载自定义的serialization，可以直接通过Config.registerDecorator注册一个IKryoDecorator，在decorate方法中对Kyro注册要序列化的class
• Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS(topology.skip.missing.kryo.registrations，默认为false)这个属性容易跟Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION(topology.fall.back.on.java.serialization)混淆起来，前者是storm自身的属性而后者storm包装的kryo的属性(registrationRequired)；Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS配置的是在有自定义Config.TOPOLOGY_KRYO_DECORATORS的场景下，如果storm加载不到用户自定义的IKryoDecorator类时是skip还是抛异常
• Kryo的registrationRequired为false的话，则会自动对未注册的class进行隐式注册(注册到classToNameId)，只在第一次序列化的时候使用className，之后都用id替代，来节省空间；不过要注意的是如果Kryo的autoReset为true的话，那么classToNameId会被reset，因而隐式注册在非第一次遇到未注册的class的时候并不能一直走使用id代替className来序列化

doc