(defn setup-ticks! [worker executor-data]
 (let [storm-conf (:storm-conf executor-data)
    tick-time-secs (storm-conf TOPOLOGY-TICK-TUPLE-FREQ-SECS)
    receive-queue (:receive-queue executor-data)
    context (:worker-context executor-data)]
  (when tick-time-secs
   (if (or (system-id? (:component-id executor-data))
       (and (= false (storm-conf TOPOLOGY-ENABLE-MESSAGE-TIMEOUTS))
          (= :spout (:type executor-data))))
    (log-message "Timeouts disabled for executor " (:component-id executor-data) ":" (:executor-id executor-data))
    (schedule-recurring
     (:user-timer worker)
     tick-time-secs
     tick-time-secs
     (fn []
      (disruptor/publish
       receive-queue
       [[nil (TupleImpl. context [tick-time-secs] Constants/SYSTEM_TASK_ID Constants/SYSTEM_TICK_STREAM_ID)]]
       )))))))

public static boolean isTick(Tuple tuple) {
  return tuple != null
      && Constants.SYSTEM_COMPONENT_ID .equals(tuple.getSourceComponent())
      && Constants.SYSTEM_TICK_STREAM_ID.equals(tuple.getSourceStreamId());
}

  public String getComponentId(int taskId) {
    if(taskId==Constants.SYSTEM_TASK_ID) {
      return Constants.SYSTEM_COMPONENT_ID;
    } else {
      return _taskToComponent.get(taskId);
    }
  }

String spoutId = "wordGenerator";
  String counterId = "counter";
  String intermediateRankerId = "intermediateRanker";
  String totalRankerId = "finalRanker";
  // 這里，假設TestWordSpout就是我們發(fā)送話題tuple的源
  builder.setSpout(spoutId, new TestWordSpout(), 5);
  // RollingCountBolt的時間窗口為9秒鐘，每3秒發(fā)送一次統(tǒng)計結果到下游
  builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));
  // IntermediateRankingsBolt，將完成部分聚合，統(tǒng)計出top-n的話題
  builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields(
    "obj"));
    // TotalRankingsBolt， 將完成完整聚合，統(tǒng)計出top-n的話題
  builder.setBolt(totalRankerId, new TotalRankingsBolt(TOP_N)).globalGrouping(intermediateRankerId);

 @Override
 public void execute(Tuple tuple) {
  if (TupleUtils.isTick(tuple)) {
   LOG.debug("Received tick tuple, triggering emit of current window counts");
   // tick來了，將時間窗口內(nèi)的統(tǒng)計結果發(fā)送，并讓窗口滾動
   emitCurrentWindowCounts();
  }
  else {
   // 常規(guī)tuple，對話題計數(shù)即可
   countObjAndAck(tuple);
  }
 }

// obj即為話題，增加一個計數(shù) count++
 // 注意，這里的速度基本取決于流的速度，可能每秒百萬，也可能每秒幾十.
 // 內(nèi)存不足？ bolt可以scale-out.
 private void countObjAndAck(Tuple tuple) {
  Object obj = tuple.getValue(0);
  counter.incrementCount(obj);
  collector.ack(tuple);
 }
 
 // 將統(tǒng)計結果發(fā)送到下游
 private void emitCurrentWindowCounts() {
  Map<Object, Long> counts = counter.getCountsThenAdvanceWindow();
  int actualWindowLengthInSeconds = lastModifiedTracker.secondsSinceOldestModification();
  lastModifiedTracker.markAsModified();
  if (actualWindowLengthInSeconds != windowLengthInSeconds) {
   LOG.warn(String.format(WINDOW_LENGTH_WARNING_TEMPLATE, actualWindowLengthInSeconds, windowLengthInSeconds));
  }
  emit(counts, actualWindowLengthInSeconds);
 }

 public final void execute(Tuple tuple, BasicOutputCollector collector) {
  if (TupleUtils.isTick(tuple)) {
   getLogger().debug("Received tick tuple, triggering emit of current rankings");
   // 將聚合并排序的結果發(fā)送到下游
   emitRankings(collector);
  }
  else {
   // 聚合并排序
   updateRankingsWithTuple(tuple);
  }
 }

// IntermediateRankingsBolt的聚合排序方法：
 @Override
 void updateRankingsWithTuple(Tuple tuple) {
  // 這一步，將話題、話題出現(xiàn)的次數(shù)提取出來
  Rankable rankable = RankableObjectWithFields.from(tuple);
  // 這一步，將話題出現(xiàn)的次數(shù)進行聚合，然后重排序所有話題
  super.getRankings().updateWith(rankable);
 }

// TotalRankingsBolt的聚合排序方法
 @Override
 void updateRankingsWithTuple(Tuple tuple) {
 // 提出來自IntermediateRankingsBolt的中間結果
  Rankings rankingsToBeMerged = (Rankings) tuple.getValue(0);
 // 聚合并排序
  super.getRankings().updateWith(rankingsToBeMerged);
 // 去0，節(jié)約內(nèi)存
  super.getRankings().pruneZeroCounts();
 }

 private void rerank() {
  Collections.sort(rankedItems);
  Collections.reverse(rankedItems);
 }