我的数据来自连接到风暴集群的websocket传感器,所以每当一个数据点到达我的websocket服务器时,我将其添加到ConcurrentLinkedQueue中。我没有关于数据点“生产”频率的先验信息。
我的spout从这个队列中获取数据点并发出相应的Tupple。一开始一切都很好(我会说大约1000个数据点),但随后我遇到了以下错误:
76245 [Thread-23-incDp] ERROR backtype.storm.util - Async loop died!
java.lang.NullPointerException: null
at streams.storm.spout.MySpout.nextTuple(MySpout.java:56) ~[bin/:na]
at backtype.storm.daemon.executor$fn__3373$fn__3388$fn__3417.invoke(executor.clj:565) ~[storm-core-0.9.3.jar:0.9.3]
at backtype.storm.util$async_loop$fn__464.invoke(util.clj:463) ~[storm-core-0.9.3.jar:0.9.3]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.7.0_67]
76246 [Thread-23-incDp] ERROR backtype.storm.daemon.executor -
java.lang.NullPointerException: null
at streams.storm.spout.MySpout.nextTuple(MySpout.java:56) ~[bin/:na]
at backtype.storm.daemon.executor$fn__3373$fn__3388$fn__3417.invoke(executor.clj:565) ~[storm-core-0.9.3.jar:0.9.3]
at backtype.storm.util$async_loop$fn__464.invoke(util.clj:463) ~[storm-core-0.9.3.jar:0.9.3]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.7.0_67]
76481 [Thread-23-incDp] ERROR backtype.storm.util - Halting process: ("Worker died")
java.lang.RuntimeException: ("Worker died")
at backtype.storm.util$exit_process_BANG_.doInvoke(util.clj:325) [storm-core-0.9.3.jar:0.9.3]
at clojure.lang.RestFn.invoke(RestFn.java:423) [clojure-1.5.1.jar:na]
at backtype.storm.daemon.worker$fn__3808$fn__3809.invoke(worker.clj:452) [storm-core-0.9.3.jar:0.9.3]
at backtype.storm.daemon.executor$mk_executor_data$fn__3274$fn__3275.invoke(executor.clj:240) [storm-core-0.9.3.jar:0.9.3]
at backtype.storm.util$async_loop$fn__464.invoke(util.clj:473) [storm-core-0.9.3.jar:0.9.3]
at clojure.lang.AFn.run(AFn.java:24) [clojure-1.5.1.jar:na]
at java.lang.Thread.run(Unknown Source) [na:1.7.0_67]
以下是我的Spout代码:
public class MySpout extends BaseRichSpout
{
private static final long serialVersionUID = 1L;
private AtomicLong messageIdCounter = new AtomicLong();
private static Queue<String> incData;
private SpoutOutputCollector collector;
public MySpout()
{
incData = new ConcurrentLinkedQueue<String>();
}
@Override
public void nextTuple()
{
if(incData.isEmpty())
{
Utils.sleep(500);
}
else
{
String[] splittedMsg = incData.poll().split(" ; ");
JSONParser jsonParser = new JSONParser();
try
{
int ts = Integer.parseInt(splittedMsg[0]);
JSONObject json = (JSONObject) jsonParser.parse(splittedMsg[1]);
collector.emit(new Values( ts, new DataPoint(ts, new ArrayList(json.values()))), messageIdCounter.incrementAndGet() );
}
catch (ParseException e)
{
System.err.println("Wrong input format: should be json");
e.printStackTrace();
}
}
}
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("time", "dps" ));
}
public static void addElmtToQueue(String json)
{
incData.add(json);
}
我猜问题出在队列上,可能是没有数据点或者两个线程之间的并发问题(顺便说一下,我知道将其设置为静态可能不好,但是我没有找到其他解决方案,因为我需要在服务器线程上访问它...)。
有人遇到过相同的问题吗?欢迎提供任何解决方案/评论 :)
提前感谢。
if(incData.isEmpty())
能够防止这种不良行为的发生!! - nbchn