final Injector injector = GuiceInjectors.makeStartupInjector();
final Cli<Runnable> cli = builder.build();
final Runnable command = cli.parse(args);
injector.injectMembers(command);            //command已经是实例化好的线程类,直接注入
command.run();

JsonConfigProvider.bind(binder, "druid.realtime", RealtimeManagerConfig.class);
binder.bind(new TypeLiteral<List<FireDepartment>>(){}).toProvider(FireDepartmentsProvider.class).in(LazySingleton.class);   //①

JsonConfigProvider.bind(binder, "druid.realtime.cache", CacheConfig.class);
binder.install(new CacheModule());

binder.bind(QuerySegmentWalker.class).to(RealtimeManager.class).in(ManageLifecycle.class);
binder.bind(NodeTypeConfig.class).toInstance(new NodeTypeConfig("realtime"));

public class RealtimeManager implements QuerySegmentWalker {
  private final List<FireDepartment> fireDepartments;           //①
  private final QueryRunnerFactoryConglomerate conglomerate;    //②
  private final Map<String, List<FireChief>> chiefs;  //key=data source name,value=FireChiefs of all partition of that data source

  @Inject
  public RealtimeManager(List<FireDepartment> fireDepartments, QueryRunnerFactoryConglomerate conglomerate) {
    this.fireDepartments = fireDepartments;
    this.conglomerate = conglomerate;
    this.chiefs = Maps.newHashMap();
  }
}

public class FireDepartmentsProvider implements Provider<List<FireDepartment>>{
  private final List<FireDepartment> fireDepartments = Lists.newArrayList();
  public List<FireDepartment> get() { return fireDepartments; }
}

public class JacksonModule implements Module{
  public void configure(Binder binder){
    binder.bind(ObjectMapper.class).to(Key.get(ObjectMapper.class, Json.class));
  }

  @Provides @LazySingleton @Json
  public ObjectMapper jsonMapper() {
    return new DefaultObjectMapper();
  }
}

@JsonCreator
public FireDepartment(
    @JsonProperty("dataSchema") DataSchema dataSchema,
    @JsonProperty("ioConfig") RealtimeIOConfig ioConfig,
    @JsonProperty("tuningConfig") RealtimeTuningConfig tuningConfig
)

public class StorageNodeModule implements Module{
  public void configure(Binder binder) {
    JsonConfigProvider.bind(binder, "druid.server", DruidServerConfig.class);
    JsonConfigProvider.bind(binder, "druid.segmentCache", SegmentLoaderConfig.class);

    binder.bind(NodeTypeConfig.class).toProvider(Providers.<NodeTypeConfig>of(null));
    binder.bind(QueryableIndexFactory.class).to(MMappedQueryableIndexFactory.class).in(LazySingleton.class);
    binder.bind(ColumnConfig.class).to(DruidProcessingConfig.class);

    binder.bind(QueryRunnerFactoryConglomerate.class).to(DefaultQueryRunnerFactoryConglomerate.class).in(LazySingleton.class);
  }
}

public class QueryRunnerFactoryModule extends QueryToolChestModule {
  private static final Map<Class<? extends Query>, Class<? extends QueryRunnerFactory>> mappings =
      ImmutableMap.<Class<? extends Query>, Class<? extends QueryRunnerFactory>>builder()
                  .put(TimeseriesQuery.class, TimeseriesQueryRunnerFactory.class)
                  .put(TimeBoundaryQuery.class, TimeBoundaryQueryRunnerFactory.class)
                  .....build();

  public void configure(Binder binder) {
    super.configure(binder);
    binder.bind(QueryWatcher.class).to(QueryManager.class).in(LazySingleton.class);
    binder.bind(QueryManager.class).in(LazySingleton.class);

    final MapBinder<Class<? extends Query>, QueryRunnerFactory> queryFactoryBinder = DruidBinders.queryRunnerFactoryBinder(binder);
    for (Map.Entry<Class<? extends Query>, Class<? extends QueryRunnerFactory>> entry : mappings.entrySet()) {
      queryFactoryBinder.addBinding(entry.getKey()).to(entry.getValue());   //注入Map的key和value,对应factories
      binder.bind(entry.getValue()).in(LazySingleton.class);                //最好还要注入一下value
    }
    binder.bind(GroupByQueryEngine.class).in(LazySingleton.class);
  }
}

public void run() {
  plumber = initPlumber();    //fireDepartment.findPlumber() 先找到水管(水龙头)
  plumber.startJob();         //准备工作,接上管子
  firehose = initFirehose();  //fireDepartment.connect() (向消防局申请一条)消防带,给消防带接上水龙头
  runFirehose(firehose);      //开始消防工作,水会从源头数据源不断流出来
  plumber.finishJob();        //完成工作,卸掉消防带,关闭水龙头
}

private void runFirehose(Firehose firehose) {
  final Supplier<Committer> committerSupplier = Committers.supplierFromFirehose(firehose);
  while (firehose.hasMore()) {
    final InputRow inputRow = firehose.nextRow();
    lateEvent = plumber.add(inputRow, committerSupplier) == -1;
    if (indexLimitExceeded || lateEvent) {
      plumber.persist(committerSupplier.get());
    }
  }
}

public Firehose connect() throws IOException {
  return ioConfig.getFirehoseFactory().connect(dataSchema.getParser());
}

"ioConfig" : {
  "type" : "realtime",
  "firehose": {             //①
    "type": "kafka-0.8",    //对应KafkaEightFirehoseFactory
    "consumerProps": {
      "zookeeper.connect": "localhost:2181",
      ...
    },
    "feed": "wikipedia"
  },
  "plumber": {
    "type": "realtime"
  }
},

@JsonCreator
public DataSchema(
    @JsonProperty("dataSource") String dataSource,
    @JsonProperty("parser") Map<String, Object> parser,
    @JsonProperty("metricsSpec") AggregatorFactory[] aggregators,
    @JsonProperty("granularitySpec") GranularitySpec granularitySpec,
    @JacksonInject ObjectMapper jsonMapper
)

"dataSchema" : {
  "dataSource" : "wikipedia",
  "parser" : {              //②
    "type" : "string",      //对应StringInputRowParser
    "parseSpec" : {
      "format" : "json",
      "timestampSpec" : {
        "column" : "timestamp",
        "format" : "auto"
      },
      "dimensionsSpec" : {
        "dimensions": ["page","language","user","unpatrolled","newPage","robot","anonymous","namespace","continent","country","region","city"],
        "dimensionExclusions" : [],
        "spatialDimensions" : []
      }
    }
  },
  "metricsSpec" : [{
    "type" : "count",
    "name" : "count"
  }],
  "granularitySpec" : {
    "type" : "uniform",
    "segmentGranularity" : "DAY",
    "queryGranularity" : "NONE"
  }
}

public InputRowParser getParser(){
  final InputRowParser inputRowParser = jsonMapper.convertValue(this.parser, InputRowParser.class);
  final DimensionsSpec dimensionsSpec = inputRowParser.getParseSpec().getDimensionsSpec();
  final TimestampSpec timestampSpec = inputRowParser.getParseSpec().getTimestampSpec();
  return inputRowParser.withParseSpec(        //进入parseSpec
      inputRowParser.getParseSpec()           //获取parseSpec
            .withDimensionsSpec(              //进入dimensionsSpec
                dimensionsSpec                //获取dimensionsSpec
                    .withDimensionExclusions( //过滤dimensionExclusions
                        Sets.difference(dimensionExclusions, dimSet)
                    )
            )
  );
}

final ConsumerConnector connector = Consumer.createJavaConsumerConnector(new ConsumerConfig(consumerProps));
final Map<String, List<KafkaStream<byte[], byte[]>>> streams = connector.createMessageStreams(ImmutableMap.of(feed,1));
final List<KafkaStream<byte[], byte[]>> streamList = streams.get(feed);
final KafkaStream<byte[], byte[]> stream = streamList.get(0);
final ConsumerIterator<byte[], byte[]> iter = stream.iterator();

return new Firehose() {
  public boolean hasMore() {
    return iter.hasNext();
  }
  public InputRow nextRow(){
    final byte[] message = iter.next().message();
    return theParser.parse(ByteBuffer.wrap(message));
  }
}

public int add(InputRow row, Supplier<Committer> committerSupplier) throws IndexSizeExceededException {
  final Sink sink = getSink(row.getTimestampFromEpoch());
  final int numRows = sink.add(row);
  if (!sink.canAppendRow() || System.currentTimeMillis() > nextFlush) {
    persist(committerSupplier.get());
  }
  return numRows;
}
private Sink getSink(long timestamp) {
  final Granularity segmentGranularity = schema.getGranularitySpec().getSegmentGranularity();
  final VersioningPolicy versioningPolicy = config.getVersioningPolicy();
  final long truncatedTime = segmentGranularity.truncate(new DateTime(timestamp)).getMillis();
  Sink retVal = sinks.get(truncatedTime);
  if (retVal == null) {
    final Interval sinkInterval = new Interval(new DateTime(truncatedTime),segmentGranularity.increment(new DateTime(truncatedTime)));
    retVal = new Sink(sinkInterval, schema, config, versioningPolicy.getVersion(sinkInterval));
    segmentAnnouncer.announceSegment(retVal.getSegment());
    sinks.put(truncatedTime, retVal);
    sinkTimeline.add(retVal.getInterval(), retVal.getVersion(), new SingleElementPartitionChunk<Sink>(retVal));
  }
  return retVal;
}

public class OnheapIncrementalIndex extends IncrementalIndex<Aggregator> {
  private final ConcurrentHashMap<Integer, Aggregator[]> aggregators = new ConcurrentHashMap<>();
  private final ConcurrentNavigableMap<TimeAndDims, Integer> facts = new ConcurrentSkipListMap<>();

  protected Integer addToFacts(AggregatorFactory[] metrics, boolean deserializeComplexMetrics, InputRow row,
      AtomicInteger numEntries, TimeAndDims key, ThreadLocal<InputRow> rowContainer, Supplier<InputRow> rowSupplier){
      Aggregator[] aggs = new Aggregator[metrics.length];
      for (int i = 0; i < metrics.length; i++) {
        final AggregatorFactory agg = metrics[i];
        aggs[i] = agg.factorize(makeColumnSelectorFactory(agg, rowSupplier, deserializeComplexMetrics));
      }

      rowContainer.set(row);        //线程安全操作, 在开始聚合前设置为当前行
      for (Aggregator agg : aggs) {
        synchronized (agg) {
          agg.aggregate();          //这里调用聚合算子的aggregate会发生聚合操作
        }
      }
      rowContainer.set(null);       //结束当前行的聚合操作后, 设置为空
      return numEntries.get();      
  }
}

public void persist(final Committer committer) {
  final List<Pair<FireHydrant, Interval>> indexesToPersist = Lists.newArrayList();
  for (Sink sink : sinks.values()) {    //每个Sink都要进行切换, 旧的保存, 新的存储最新的实时数据
    if (sink.swappable()) {           
      indexesToPersist.add(Pair.of(sink.swap(), sink.getInterval()));
    }
  }

  for (Pair<FireHydrant, Interval> pair : indexesToPersist) {
    persistHydrant(pair.lhs, schema, pair.rhs, metadata));
  }
  committer.run();
  resetNextFlush();
}

protected int persistHydrant(FireHydrant indexToPersist, DataSchema schema, Interval interval, Map<String, Object> metaData){
  int numRows = indexToPersist.getIndex().size();     //增量索引中的行数
  final IndexSpec indexSpec = config.getIndexSpec();  //索引配置,在tuningConfig中,比如bitmap类型,列压缩格式
  final File persistedFile = indexMerger.persist(     //合并索引
      indexToPersist.getIndex(),
      new File(computePersistDir(schema, interval), String.valueOf(indexToPersist.getCount())),
      metaData,
      indexSpec
  );
  indexToPersist.swapSegment(                         //增量索引转换为可查询的索引片
      new QueryableIndexSegment(indexToPersist.getSegment().getIdentifier(), indexIO.loadIndex(persistedFile))
  );
  return numRows;    
}

return merge(
    Arrays.<IndexableAdapter>asList(new IncrementalIndexAdapter(dataInterval, index, indexSpec.getBitmapSerdeFactory().getBitmapFactory())),
    index.getMetricAggs(), outDir, segmentMetadata, indexSpec, progress
);