package stormNew;

import java.util.HashMap;
import java.util.Map;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class LocalTopology {

public static void main(String[] args) {

//×é×°Topology
TopologyBuilder build = new TopologyBuilder();
//¶¨ÒåspoutµÄid
build.setSpout("spout", new Spout());
//¶¨ÒåboltµÄid,ʹÓÃnew Fields("field")×ֶνøÐзÖ×é
build.setBolt("bolt", new Bolt()).fieldsGrouping("spout", new Fields("field"));

try {
Config conf = new Config();
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("LocalTopology", conf, build.createTopology());
// Config stormConf = new Config();
// stormConf.setNumWorkers(2);
// StormSubmitter.submitTopology("luluTology", stormConf,build.createTopology());
} catch (Exception e) {
e.printStackTrace();
}
}

public static class Spout extends BaseRichSpout{

private Map conf;
private TopologyContext context;
private SpoutOutputCollector collector;

private int i;
private HashMap<Integer,Integer> map = new HashMap<Integer,Integer>();

//spoutÖÐÑ­»·Õâ¸ö·½·¨£¬½øÐÐÏûÏ¢»ñÈ¡Óë·¢ËÍ
public void nextTuple() {
// System.err.println("Spout:"+i);
//×÷ΪÿһÌõÏûÏ¢µÄΨһ±êʶ£¬ÓÃÓÚackµÄÏûϢȷÈÏ»úÖÆ
int mgsid = i;
//·¢ËÍtupleÏûÏ¢µ½boltÖд¦Àí
this.collector.emit(new Values(i++,i%3),mgsid);
//spout×ÔÉíά»¤×ÅÏûÏ¢Óë±êʶ֮¼äµÄ¹Øϵ
map.put(mgsid, i);
try {
//ÐÝÃßһϣ¬ÇåÎú¿´³öЧ¹û
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

public void open(Map conf, TopologyContext context,
SpoutOutputCollector collector) {
this.conf = conf;
this.context = context;
this.collector = collector;
}


public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
//¶¨Òå·¢ËÍ×ֶεÄÃû³Æ
outputFieldsDeclarer.declare(new Fields("num","field"));
}

@Override
//µ±boltµ÷ÓõÄack·½Ãæʱ£¬»Øµ÷spoutÖеÄack·½·¨
public void ack(Object msgId) {
System.out.println("È·ÈÏÐÅÏ¢-----------");
}

@Override
//µ±boltµ÷ÓõÄfail·½Ãæʱ£¬»Øµ÷spoutÖеÄfail·½·¨
public void fail(Object msgId) {
System.out.println("ÏûϢʧ°Ü-----------"+map.get(msgId));
}



}


public static class Bolt extends BaseRichBolt{


private Map conf;
private TopologyContext context;
private OutputCollector collector;

private int sum = 0;

//bolt½ÓÊܶ¯spout·¢Ë͹ýÀ´µÄÏûÏ¢¾Íµ÷execute
public void execute(Tuple input) {
// TODO Auto-generated method stub
//ͨ¹ý×Ö¶ÎÃû³ÆÀ´»ñÈ¡Êý¾Ý
int num = input.getIntegerByField("num");
System.err.println("--------------------num:"+(num));

/**
* ΪÁ˸üºÃµÄ¿´ÊéackÏûϢȷÈÏ»úÖƵÄЧ¹û£¬ËùÒÔÖ±½Óµ÷ÓÃackÓëfail·½·¨
*
* ĬÈϸ÷½·¨ÊÇÕâÑù½øÐÐackµÄµ÷ÓÃ
* try{
* ÒµÎñÂß¼­
* this.collector.ack(input);
*}catch(){
* this.collector.fail(input);
*}
*/
if(num >=10 && num <=20){
//System.err.println("sum:"+(sum+=num));
this.collector.ack(input);
}else{
this.collector.fail(input);
}
}

public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
// TODO Auto-generated method stub
this.conf = conf;
this.context = context;
this.collector = collector;
}

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
// TODO Auto-generated method stub

}

}

}

ͨ¹ý´úÂëµ÷Õû
topologyBuilder.setBolt("green-bolt", new GreenBolt(),2)
.setNumTasks(4).shuffleGrouping("blue-spout);
ͨ¹ýshellµ÷Õû
# 10ÃëÖ®ºó¿ªÊ¼µ÷Õû
# Reconfigure the topology "mytopology" to use 5 worker processes,
# the spout "blue-spout" to use 3 executors and
# the bolt "yellow-bolt" to use 10 executors.
storm rebalance mytopology -w 10 -n 5 -e blue-spout=3 -e yellow-bolt=10