遇到BlockingQueue java问题，实现Storm（分布式计算）？(having issues with BlockingQueue java , implementation of Storm (Distributed computing)?)

 这是我的输入spout的代码snippt，用于将元组发送到处理节点以进行集群上的流处理。 问题是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()));
 
 谢谢 

This is the code snippt of my input spout for emmiting tuple to a processing noded for stream processing over a cluster. The problem is The BlockingQueue is throwing 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"));
}
 
The exception description is as followes :---
 
java.lang.InterruptedException10930 [Thread-20] INFO backtype.storm.util - Async loop interrupted!
 
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)
 
And The nextTuple(InputStreamSpout.java:65 is ------>
 
        collector.emit(new Values("Single Temperature Reading", blockingQueue.take()));
 
Thanks 

原文：https://stackoverflow.com/questions/11343299