1

这是我的输入 spout 的代码片段，用于将元组发送到一个处理节点，以便在集群上进行流处理。问题是 BlockingQueue 正在抛出 InterruptedException 。

private SpoutOutputCollector collector;
public BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<String>();

public boolean isDistributed() {
    return true;
    }


public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context,
final SpoutOutputCollector collector) {

    this.collector=collector;

}

@Override
public void nextTuple() {


    try {
        //Utils.sleep(100);
        collector.emit(new Values("Single Temperature Reading", blockingQueue.take()));
    } catch (InterruptedException e) {
        e.printStackTrace();
    }



}

public void readInputfile() throws IOException, InterruptedException{
    FileInputStream file = new FileInputStream("/home/529076/Desktop/Temperature");
    DataInputStream readDate=new DataInputStream(file);
    BufferedReader readText=new BufferedReader(new InputStreamReader(readDate));

    String line;
    String singleReading = null;
    while((line=readText.readLine())!=null){
         singleReading=line;
         blockingQueue.add(singleReading);

    }

}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
    declarer.declare(new Fields("Single Temperature Reading"));
}

异常描述如下：---

java.lang.InterruptedException10930 [Thread-20] INFO backtype.storm.util - 异步循环中断！

at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.reportInterruptAfterWait(AbstractQueuedSynchronizer.java:1961)
at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:1996)
at java.util.concurrent.LinkedBlockingQueue.take(LinkedBlockingQueue.java:399)
at com.tcs.storm.test.InputStreamSpout.nextTuple(InputStreamSpout.java:65)
at backtype.storm.daemon.task$fn__3349$fn__3404.invoke(task.clj:413)

而 nextTuple(InputStreamSpout.java:65 是 ------>

        collector.emit(new Values("Single Temperature Reading", blockingQueue.take()));

谢谢

4

2 回答 2

1

错误是由于以下事实 - BlockingQueue 未在输出收集器中初始化；

于 2012-07-30T11:30:40.090 回答
-1

在 Storm 中，建议在 prepare()/open() 方法中初始化 Bolts/Spouts 的字段。

原因是 Bolts/Spouts 是在您上传拓扑的节点上创建的：您的 BlockingQueue 将在那里构建。Bolts/Spout 然后被序列化并分发到工作节点，在那里它们被反序列化。这个序列化/反序列化过程可能无法保留在构造 Bolt/Spout 时设置的字段的所有属性。在 prepare() 或 open() 中初始化的字段没有这个问题。

于 2013-01-28T18:47:05.937 回答