我正在尝试实现trident+drpc。我设计的拓扑结构不会无限期地运行。我有两个独立的类，一个用于喷口实现，另一个用于实现drpc和trident。我的spout类（扩展irichspout的spout）发出客户的id。即

public class TriSpout implements IRichSpout{
  //some logic here
    spoutOutputCollector.emit(new Values(id))
  }

现在我从另一个类中的输出收集器获得了值，该类使用drpc实现了trident。

public class TriDrpc{

    .....
    TriSpout spout=new TriSpout1();        
    TridentTopology topology = new TridentTopology();  
    TridentState wordCounts =
          topology.newStream("spout1",spout)
            .parallelismHint(1)
            .each(new Fields("id"), new Compute(), new Fields("value"))
            .persistentAggregate(new MemoryMapState.Factory(),
                                 new Count(), new Fields("count"))

drpc拓扑定义如下

topology.newDRPCStream("Calc", drpc)
         .each(new Fields("args"), new Split(), new Fields("word"))                
         .stateQuery(wordCounts, new Fields("word"), new MapGet(), new Fields("count"));

drpc请求如下

public static void main(String[] args) throws Exception {
    Config conf = new Config(); 

    if (args.length == 0) {
    LocalDRPC drpc = new LocalDRPC();
    LocalCluster cluster = new LocalCluster();
    cluster.submitTopology("Calculator", conf,   buildTopology(drpc));          
    System.out.println("DRPC RESULT: "
                + drpc.execute("Calc", "id"));
    Thread.sleep(1000);

    } else {
        conf.setNumWorkers(8);
        StormSubmitter.submitTopology(args[0], conf, buildTopology(null));
    }
}

现在在上面的代码中，在drpc请求中，即。

System.out.println("DRPC RESULT: " + drpc.execute("Calc", "id"));

这个 "id" 应该与喷口发出的id相同，即我想知道哪个客户有使用此id的活动帐户，因此我需要发送喷口发出的所有id的drpc请求。现在drpc是在main类中，我怎样才能在不手动指定id的情况下将喷口发出的值传递给drpc请求呢？
有人能帮忙吗
用新信息编辑