java.io.NotSerializedException : com. impetus.kundera.persistence.EntityManagerImpl

Question

我想使用kundera作为apachestorm的dao层，将数据存储在cassandra数据库中。我的拓扑创建没有任何问题，我可以接收消息，但是当我尝试将数据存储在数据库中时，我收到消息:

26240 [main-EventThread] INFO  o.a.s.s.o.a.c.f.s.ConnectionStateManager - State change: CONNECTED
26275 [main] INFO  o.a.s.d.supervisor - Starting supervisor with id 1db18608-a2df-47e0-8ae6-cc634c90f81d at host 10.0.0.4
26308 [main] ERROR o.a.s.s.o.a.z.s.NIOServerCnxnFactory - Thread Thread[main,5,main] died
java.lang.IllegalStateException: Bolt 'bolt1' contains a non-serializable field of type com.impetus.kundera.persistence.EntityManagerImpl, which was instantiated prior to topology creation. com.impetus.kundera.persistence.EntityManagerImpl should be instantiated within the prepare method of 'bolt1 at the earliest.
        at org.apache.storm.topology.TopologyBuilder.createTopology(TopologyBuilder.java:127) ~[storm-core-1.0.0.jar:1.0.0]
        at topology.ConnectorTopology.main(ConnectorTopology.java:52) ~[Zorro-0.0.1-SNAPSHOT-jar-with-dependencies.jar:?]
Caused by: java.lang.RuntimeException: java.io.NotSerializableException: com.impetus.kundera.persistence.EntityManagerImpl
        at org.apache.storm.utils.Utils.javaSerialize(Utils.java:167) ~[storm-core-1.0.0.jar:1.0.0]
        at org.apache.storm.topology.TopologyBuilder.createTopology(TopologyBuilder.java:122) ~[storm-core-1.0.0.jar:1.0.0]
        ... 1 more
Caused by: java.io.NotSerializableException: com.impetus.kundera.persistence.EntityManagerImpl
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1183) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1547) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1508) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1431) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1177) ~[?:1.7.0_99]
        at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:347) ~[?:1.7.0_99]
        at org.apache.storm.utils.Utils.javaSerialize(Utils.java:163) ~[storm-core-1.0.0.jar:1.0.0]
        at org.apache.storm.topology.TopologyBuilder.createTopology(TopologyBuilder.java:122) ~[storm-core-1.0.0.jar:1.0.0]
        ... 1 more

在topologybuilder类中我有:

brokerSpout = new BrokerSpout(rabbitMQAMQP);
            builder.setSpout("spout1", brokerSpout);
printerbolt = new PrinterBolt();
            builder.setBolt("bolt1", printerbolt).shuffleGrouping("spout1");

在我的类 dao 中(它实现了可序列化接口(interface))

private EntityManager em;
    @Transient
    private EntityManagerFactory emf;
    public SensorDAOImpl() {
        // TODO Auto-generated constructor stub

         emf = Persistence.createEntityManagerFactory("cassandra_pu");
         em = emf.createEntityManager();
    }




    @Override
    public void insert(Object entity)
    {

        em.persist(entity);

    }

我的 bolt 代码

private OutputCollector collector;
    public PrinterBolt() {
        this.index=0;
    EntityManagerFactory emf = Persistence.createEntityManagerFactory("cassandra_pu");
    EntityManager em= emf.createEntityManager();
        database= new DatabaseController(em);
        // TODO Auto-generated constructor stub
    }

    /* (non-Javadoc)
     * @see org.apache.storm.task.IBolt#execute(org.apache.storm.tuple.Tuple)
     */
    public void prepare(@SuppressWarnings("rawtypes") Map config, TopologyContext context,
            OutputCollector collector) {
        this.collector = collector;
    }
    public void execute(Tuple tuple) {
        //tuple.getBinaryByField(arg0)
        String message = tuple.getStringByField("message");
        System.out.println("Receive ["+index+"]"+message);
        database.saveEntitie(tuple.getBinaryByField("message"));
        index++;

    }

那么，我该如何解决这个问题，(我无法触及 com.impetus.kundera.persistence.EntityManagerImpl 类)

Answer 1

在 Bolt 的准备方法中实例化 com.impetus.kundera.persistence.EntityManagerImpl 类型的字段。

public void prepare(@SuppressWarnings("rawtypes") Map config, TopologyContext context,
            OutputCollector collector) {
     emf = Persistence.createEntityManagerFactory("cassandra_pu");
     em = emf.createEntityManager();
     database = new DatabaseController(em);
     this.collector = collector;
    }