Storm框架：如何消费RabbitMq消息(代码案例)

1、定义拓扑topology

public class MessageTopology {

    public static void main(String[] args) throws Exception {
        //组装topology
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        topologyBuilder.setSpout("RabbitmqSpout", new RabbitmqSpout());
        topologyBuilder.setBolt("FilterBolt", new FilterBolt()).shuffleGrouping("RabbitmqSpout");

        Config conf = new Config ();
        try {
            if (args.length > 0) {
                StormSubmitter.submitTopology(args[0], conf, topologyBuilder.createTopology());
            } else {
                LocalCluster localCluster = new LocalCluster();
                localCluster.submitTopology("messageTopology", conf, topologyBuilder.createTopology());
            }
        } catch (AlreadyAliveException e) {
            e.printStackTrace();
        }
    }
}public class MessageTopology {