import backtype.storm.serialization.KryoValuesDeserializer; //导入依赖的package包/类
@Test
public void testSerialization() throws IOException {
    Map conf = Utils.readDefaultConfig();
    Config.registerSerialization(conf, Metadata.class);

    KryoValuesSerializer kvs = new KryoValuesSerializer(conf);
    Metadata md = new Metadata();
    byte[] content = kvs.serializeObject(md);

    KryoValuesDeserializer kvd = new KryoValuesDeserializer(conf);
    Metadata md2 = (Metadata) kvd.deserializeObject(content);

    // TODO compare md1 and md2
}
 

import backtype.storm.serialization.KryoValuesDeserializer; //导入依赖的package包/类
/**
 * Creates the kryo serializer for the general tuple.
 * @param conf the Storm configuration file
 */
@SuppressWarnings("rawtypes")
public KryoGeneralTupleSerializer(final Map conf) {
    this.valuesSer = new KryoValuesSerializer(conf);
    this.valuesDeser = new KryoValuesDeserializer(conf);
    this.kryoIn = new Input(1);
    this.kryoOut = new Output(2000, 2000000000);
}
 

import backtype.storm.serialization.KryoValuesDeserializer; //导入依赖的package包/类
/**
 * Creates the kryo serializer for the switch tuple.
 * @param conf the storm config
 */
@SuppressWarnings("rawtypes")
public KryoSwitchTupleSerializer(final Map conf) {
    this.valuesSer = new KryoValuesSerializer(conf);
    this.valuesDeser = new KryoValuesDeserializer(conf);
    this.kryoIn = new Input(1);
    this.kryoOut = new Output(2000, 2000000000);
}
 

import backtype.storm.serialization.KryoValuesDeserializer; //导入依赖的package包/类
public BinDecoder(IRunContext context) {
	
	_deser = new KryoValuesDeserializer(context.getStormConf());
}