Strom的trident小例子

上代码:

public class TridentFunc {
    
    /**
     * 类似于普通的bolt
     */
    public static class MyFunction extends BaseFunction{
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            Integer value = tuple.getIntegerByField("sentence");
            System.out.println(value);
        }
    }
    
    public static void main(String[] args) {
        @SuppressWarnings("unchecked")
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
        spout.setCycle(true);//让spout循环发送数据
        
        TridentTopology tridentTopology = new TridentTopology();
        tridentTopology.newStream("spoutid",spout)
            .each(new Fields("sentence"), new MyFunction(), new Fields(""));
        
        LocalCluster localCluster = new LocalCluster();
        String simpleName = TridentFunc.class.getSimpleName();
        localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
        //运行结果就是  一直循环打印 1 2 1 2  
    }
}

 多数据源

public class TridentMeger {
    
    /**
     * 类似于普通的bolt
     */
    public static class MyFunction extends BaseFunction{
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            Integer value = tuple.getIntegerByField("sentence");
            System.out.println(value);
        }
    }
    
    public static void main(String[] args) {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
        //spout.setCycle(true);//让spout循环发送数据
        
        TridentTopology tridentTopology = new TridentTopology();
        //指定多个数据源，流连接
        Stream newStream1 = tridentTopology.newStream("spoutid1",spout);
        Stream newStream2 = tridentTopology.newStream("spoutid2",spout);
        
        //tridentTopology.newStream("spoutid",spout) 之前是这种  但是只能有 一个数据源  
        tridentTopology.merge(newStream1,newStream2)//使用这种就可以有多个数据源.
            .each(new Fields("sentence"), new MyFunction(), new Fields(""));
        
        LocalCluster localCluster = new LocalCluster();
        String simpleName = TridentMeger.class.getSimpleName();
        localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
    }
}

 增加过滤器

public class TridentFilter {
    
    /**
     * 类似于普通的bolt
     */
    public static class MyFunction extends BaseFunction{
        @Override
        public void execute(TridentTuple tuple, TridentCollector collector) {
            Integer value = tuple.getIntegerByField("sentence");
            System.out.println(value);
        }
    }
    
    public static class MyFilter extends BaseFilter{//专门封装了一个Filter功能.
        //对数据进行过滤  如果过滤出的数据不要了就false 保留就ture
        @Override
        public boolean isKeep(TridentTuple tuple) {
            Integer value = tuple.getIntegerByField("sentence");
            return value%2==0?true:false; //只要偶数不要奇数
        }
    }
    
    public static void main(String[] args) {
        FixedBatchSpout spout = new FixedBatchSpout(new Fields("sentence"), 1, new Values(1),new Values(2));
        spout.setCycle(true);//让spout循环发送数据
        
        TridentTopology tridentTopology = new TridentTopology();
        tridentTopology.newStream("spoutid",spout)     //这个地方只能指定一个数据源,如果想指定多个数据源Spout 看TridentMeger.java
            .each(new Fields("sentence"), new MyFilter())
            .each(new Fields("sentence"), new MyFunction(), new Fields(""));
        
        LocalCluster localCluster = new LocalCluster();
        String simpleName = TridentFilter.class.getSimpleName();
        localCluster.submitTopology(simpleName, new Config(), tridentTopology.build());
    }
}