java - 如何使用 Java 在 Storm Bolts 中编写prepare() 方法？

Question

我想从 Bolt.prepare() 读取文件并将读取的文件对象传递给执行方法。任何人都可以给我发布示例代码，如何使用 Java 在 Storm Bolts 中执行此操作吗？我正在使用:

public void prepare(Map stormConf, TopologyContext context) {
    try {
        this.context = context;
        this.fileReader = new FileInputStream("/home/anji/all-users.csv");
    } catch (FileNotFoundException e) {
        /*throw new RuntimeException("Error reading file "
                + conf.get("inputFile"));*/
    }
    this.collector = collector;

}

我在 execute() 方法中调用 FileReader:User p = new User(fileReader);

错误消息:

java.io.IOException: Stream Closed
  at java.io.FileInputStream.readBytes(Native Method)
  at java.io.FileInputStream.read(FileInputStream.java:272)
  at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)
  at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)
  at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177)
  at java.io.InputStreamReader.read(InputStreamReader.java:184)
  at com.csvreader.CsvReader.checkDataLength(Unknown Source)
  at com.csvreader.CsvReader.readRecord(Unknown Source)
  at com.csvreader.CsvReader.readHeaders(Unknown Source)

Answer 1

只需在 Bolt 中使用包含所有User 的成员变量即可。在 prepare() 中仅读取一次文件:

public class MyClass implements IRichBolt {
    private final List<User> users = new ArrayList<User>();

    public void prepare(...) {
        File f = new File(...);
        // open file and create objects (or similar code)
       String line;
       while((line = reader.readLine()) != null) {
           users.add(new Users(line));
       }
    }

    // other methods including .execute(...) can use users 
}