public void registerSerialization(Class klass);
public void registerSerialization(Class klass, Class<? extends Serializer> serializerClass);

package cn.intsmaze.serializable.bean;
public class Person {
　　private int age;
　　private Studnet studnet;
　　private ArrayList arrayList=new ArrayList();
　　private LinkedList linkedList=new LinkedList();
　　public Person() {
　　}
　　public Person(int age,Studnet s) {
　　　　this.age = age;
　　　　this.studnet=s;
　　　　arrayList.add("ArrayList中的"+s.getName());
　　　　linkedList.add("linkedList中的"+s.getName());
　　}
　　@Override
　　public String toString() {
　　　　return "Person [age=" + age + ", studnet=" + studnet + ", arrayList="
　　　　+ arrayList + ", linkedList=" + linkedList + "]";
　　}
　　　　get(),set()......
}

package cn.intsmaze.serializable.bean;
public class Studnet {
　　private String name;
　　public Studnet() {
　　}
　　public Studnet(String name) {
　　　　this.name = name;
　　}
　　@Override
　　public String toString() {
　　　　return "Studnet [name=" + name + "]";
　　} 
　　get(),set()......
}

package cn.intsmaze.serializable;
import cn.intsmaze.serializable.bean.Person;
import cn.intsmaze.serializable.bean.Studnet;
public class SpoutBean extends BaseRichSpout {
　　SpoutOutputCollector collector;
　　public void open(Map conf, TopologyContext context,
　　　　SpoutOutputCollector collector) {
　　　　this.collector = collector;
　　}
　　public void nextTuple() {
　　　　Studnet s=new Studnet("xiaoxi");
　　　　collector.emit(new Values(new Person(100,s)));
　　　　Utils.sleep(500);
　　}
　　public void declareOutputFields(OutputFieldsDeclarer declarer) {
　　　　declarer.declare(new Fields("person"));
　　}
} 
 
package cn.intsmaze.serializable;
import cn.intsmaze.serializable.bean.Person;
public class BoltBean extends BaseBasicBolt {
　　public void prepare(Map stormConf, TopologyContext context) {
　　　　super.prepare(stormConf, context);
　　}
　　public void execute(Tuple input, BasicOutputCollector collector) {
　　　　Person person = (Person)input.getValueByField("person");
　　　　System.out.println("接收到spout节点传来的数据:"+person);
　　}
　　public void declareOutputFields(OutputFieldsDeclarer declarer) {
　　}
}

package cn.intsmaze.serializable;
import java.util.LinkedList;
import cn.intsmaze.serializable.bean.Person;
import cn.intsmaze.serializable.bean.Studnet;
public class TopologyBean {
public static void main(String[] args) throws Exception {
    TopologyBuilder builder = new TopologyBuilder();
    builder.setSpout("spout", new SpoutBean(), 1);
    builder.setBolt("bolt", new BoltBean(), 1).shuffleGrouping("spout");
    Config conf = new Config();
    conf.registerSerialization(Person.class);
    conf.registerSerialization(Studnet.class);
    //注释掉后，但Studnet没实现java序列化，则会报错。有两种方法，一种注册该类，一种实现java序列化。
    conf.registerSerialization(LinkedList.class);
    //这里如果注释掉，则会使用java序列化方式，如果我们取消掉禁止使用java序列化方法，则会提示注册LinkedList类报错。
    conf.setNumWorkers(2);
     // Config.setFallBackOnJavaSerialization(conf, false);//禁止使用java语言自己的序列化
    StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
   }
}

package cn.intsmaze.serializable.bean;
import java.util.ArrayList;
import java.util.LinkedList;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.Serializer;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
public class PersonSerializable extends Serializer<Person>{
　　@Override
　　public Person read(Kryo kryo, Input input, Class<Person> arg2) {
　　　　System.out.println("序列化");
　　　　Person person=new Person();
　　　　person.setAge(input.readInt());
　　　　person.setArrayList(kryo.readObject(input, ArrayList.class));
　　　　person.setLinkedList(kryo.readObject(input, LinkedList.class));//该类型，storm默认不支持，所以要在topology中注册该类型，如果不注册，则会使用java序列化。 person.setStudnet(kryo.readObject(input, Studnet.class));//该类型，storm默认不支持，所以要在topology中注册该类型，如果不注册，且java序列化没有实现，则会报错。
　　　　return person;
　　}
　　@Override
　　public void write(Kryo kryo, Output output, Person person) {
　　　　System.out.println("反序列化");
　　　　output.writeInt(person.getAge());
　　　　kryo.writeObject(output, person.getArrayList());
　　　　kryo.writeObject(output, person.getLinkedList());
　　　　kryo.writeObject(output, person.getStudnet());
　　}
}

package cn.intsmaze.serializable;
import java.util.LinkedList;
import cn.intsmaze.serializable.bean.Person;
import cn.intsmaze.serializable.bean.PersonSerializable;
import cn.intsmaze.serializable.bean.Studnet;
public class TopologyBean {
　　public static void main(String[] args) throws Exception {
    　　TopologyBuilder builder = new TopologyBuilder();
    　　builder.setSpout("spout", new SpoutBean(), 1);
   　　 builder.setBolt("bolt", new BoltBean(), 1).shuffleGrouping("spout");
    　　Config conf = new Config();
    　　conf.registerSerialization(Studnet.class);
   　　 conf.registerSerialization(LinkedList.class);
   　　 conf.registerSerialization(Person.class, PersonSerializable.class);
    　　conf.setNumWorkers(2);     
    　　StormSubmitter.submitTopologyWithProgressBar(args[0], conf,
    　　builder.createTopology());
    }
}