PipelineInterpreter.java 文件源码

@AssistedInject
public State(@Assisted ImmutableMap<String, Pipeline> currentPipelines,
             @Assisted ImmutableSetMultimap<String, Pipeline> streamPipelineConnections,
             MetricRegistry metricRegistry,
             @Named("processbuffer_processors") int processorCount,
             @Named("cached_stageiterators") boolean cachedIterators) {
    this.currentPipelines = currentPipelines;
    this.streamPipelineConnections = streamPipelineConnections;
    this.cachedIterators = cachedIterators;

    cache = CacheBuilder.newBuilder()
            .concurrencyLevel(processorCount)
            .recordStats()
            .build(new CacheLoader<Set<Pipeline>, StageIterator.Configuration>() {
                @Override
                public StageIterator.Configuration load(@Nonnull Set<Pipeline> pipelines) throws Exception {
                    return new StageIterator.Configuration(pipelines);
                }
            });

    // we have to remove the metrics, because otherwise we leak references to the cache (and the register call with throw)
    metricRegistry.removeMatching((name, metric) -> name.startsWith(name(PipelineInterpreter.class, "stage-cache")));
    MetricUtils.safelyRegisterAll(metricRegistry, new CacheStatsSet(name(PipelineInterpreter.class, "stage-cache"), cache));
}