Strom的trident小例子

上代码:

 1 public class TridentFunc {
 2     
 3     /**
 4      * 类似于普通的bolt
 5      */
 6     public static class MyFunction extends BaseFunction{
 7         @Override
 8         public void execute(TridentTuple tuple, TridentCollector collector) {
 9             Integer value = tuple.getIntegerByField("sentence");
10             System.out.println(value);
11         }
12     }
13     
14     public static void main(String[] args) {
15         @SuppressWarnings("unchecked")
16         FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
17         spout.setCycle(true);//让spout循环发送数据
18         
19         TridentTopology tridentTopology = new TridentTopology();
20         tridentTopology.newStream("spoutid",spout)
21             .each(new Fields("sentence"), new MyFunction(), new Fields(""));
22         
23         LocalCluster localCluster = new LocalCluster();
24         String simpleName = TridentFunc.class.getSimpleName();
25         localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
26         //运行结果就是  一直循环打印 1 2 1 2  
27     }
28 }

多数据源

 1 public class TridentMeger {
 2     
 3     /**
 4      * 类似于普通的bolt
 5      */
 6     public static class MyFunction extends BaseFunction{
 7         @Override
 8         public void execute(TridentTuple tuple, TridentCollector collector) {
 9             Integer value = tuple.getIntegerByField("sentence");
10             System.out.println(value);
11         }
12     }
13     
14     public static void main(String[] args) {
15         FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
16         //spout.setCycle(true);//让spout循环发送数据
17         
18         TridentTopology tridentTopology = new TridentTopology();
19         //指定多个数据源，流连接
20         Stream newStream1 = tridentTopology.newStream("spoutid1",spout);
21         Stream newStream2 = tridentTopology.newStream("spoutid2",spout);
22         
23         //tridentTopology.newStream("spoutid",spout) 之前是这种  但是只能有 一个数据源  
24         tridentTopology.merge(newStream1,newStream2)//使用这种就可以有多个数据源.
25             .each(new Fields("sentence"), new MyFunction(), new Fields(""));
26         
27         LocalCluster localCluster = new LocalCluster();
28         String simpleName = TridentMeger.class.getSimpleName();
29         localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
30     }
31 }

增加过滤器

 1 public class TridentFilter {
 2     
 3     /**
 4      * 类似于普通的bolt
 5      */
 6     public static class MyFunction extends BaseFunction{
 7         @Override
 8         public void execute(TridentTuple tuple, TridentCollector collector) {
 9             Integer value = tuple.getIntegerByField("sentence");
10             System.out.println(value);
11         }
12     }
13     
14     public static class MyFilter extends BaseFilter{//专门封装了一个Filter功能.
15         //对数据进行过滤  如果过滤出的数据不要了就false 保留就ture
16         @Override
17         public boolean isKeep(TridentTuple tuple) {
18             Integer value = tuple.getIntegerByField("sentence");
19             return value%2==0?true:false; //只要偶数不要奇数
20         }
21     }
22     
23     public static void main(String[] args) {
24         FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
25         spout.setCycle(true);//让spout循环发送数据
26         
27         TridentTopology tridentTopology = new TridentTopology();
28         tridentTopology.newStream("spoutid",spout)     //这个地方只能指定一个数据源,如果想指定多个数据源Spout 看TridentMeger.java
29             .each(new Fields("sentence"), new MyFilter())
30             .each(new Fields("sentence"), new MyFunction(), new Fields(""));
31         
32         LocalCluster localCluster = new LocalCluster();
33         String simpleName = TridentFilter.class.getSimpleName();
34         localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
35     }
36 }