(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());
}

;; 請注意SYSTEM_TASK_ID和SYSTEM_TICK_STREAM_ID
(TupleImpl. context [tick-time-secs] Constants/SYSTEM_TASK_ID Constants/SYSTEM_TICK_STREAM_ID)

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就是我們發送話題tuple的源
builder.setSpout(spoutId, new TestWordSpout(), 5);
// RollingCountBolt的時間窗口為9秒鐘，每3秒發送一次統計結果到下游
builder.setBolt(counterId, new RollingCountBolt(9, 3), 4).fieldsGrouping(spoutId, new Fields("word"));
// IntermediateRankingsBolt，將完成部分聚合，統計出top-n的話題
builder.setBolt(intermediateRankerId, new IntermediateRankingsBolt(TOP_N), 4).fieldsGrouping(counterId, new Fields(
"obj"));
// TotalRankingsBolt， 將完成完整聚合，統計出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來了，將時間窗口內的統計結果發送，并讓窗口滾動
emitCurrentWindowCounts();
}
else {
// 常規tuple，對話題計數即可
countObjAndAck(tuple);
}
}

// obj即為話題，增加一個計數 count++
// 注意，這里的速度基本取決于流的速度，可能每秒百萬，也可能每秒幾十.
// 內存不足？ bolt可以scale-out.
private void countObjAndAck(Tuple tuple) {
Object obj = tuple.getValue(0);
counter.incrementCount(obj);
collector.ack(tuple);
}

// 將統計結果發送到下游
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");
// 將聚合并排序的結果發送到下游
emitRankings(collector);
}
else {
// 聚合并排序
updateRankingsWithTuple(tuple);
}
}

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

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

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