我想将kundera用作apache风暴的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（实现可序列化的接口）中
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);
}
我的螺栓代码
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类）
最佳答案
在螺栓的制备方法中实例化类型为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;
}