import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
public static void main(String[] args) {
    LOGGER.info("{} {} starting", NAME, VERSION);

    Configuration configuration = new Configuration();

    try {
        configuration = DataBindingUtils.readConfiguration(new File("config.yml"));
    } catch (IOException e) {
        LOGGER.error("Unable to read configuration, exiting.");
        LOGGER.error(e.getMessage());
        System.exit(1);
    }

    final ServiceFactory configurationAwareServiceFactory = new ServiceFactory(configuration);
    configurationAwareServiceFactory.initializePlugins();

    final ServiceManager serviceManager = new ServiceManager(configurationAwareServiceFactory.getServices());
    LOGGER.info("Starting services");
    serviceManager.startAsync();
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
public LocalJobLauncher(Properties jobProps) throws Exception {
  super(jobProps);

  TimingEvent jobLocalSetupTimer = this.eventSubmitter.getTimingEvent(TimingEventNames.RunJobTimings.JOB_LOCAL_SETUP);

  this.taskExecutor = new TaskExecutor(jobProps);
  this.taskStateTracker = new LocalTaskStateTracker(jobProps, this.taskExecutor);

  this.serviceManager = new ServiceManager(Lists.newArrayList(
      // The order matters due to dependencies between services
      this.taskExecutor, this.taskStateTracker));
  // Start all dependent services
  this.serviceManager.startAsync().awaitHealthy(5, TimeUnit.SECONDS);

  startCancellationExecutor();

  jobLocalSetupTimer.stop();
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
public LocalJobLauncher(Properties jobProps, SharedResourcesBroker<GobblinScopeTypes> instanceBroker) throws Exception {
  super(jobProps, ImmutableList.<Tag<?>> of(), instanceBroker);
  log.debug("Local job launched with properties: {}", jobProps);

  TimingEvent jobLocalSetupTimer = this.eventSubmitter.getTimingEvent(TimingEvent.RunJobTimings.JOB_LOCAL_SETUP);

  this.taskExecutor = new TaskExecutor(jobProps);
  this.taskStateTracker =
      new LocalTaskStateTracker(jobProps, this.jobContext.getJobState(), this.taskExecutor, this.eventBus);

  this.serviceManager = new ServiceManager(Lists.newArrayList(
      // The order matters due to dependencies between services
      this.taskExecutor, this.taskStateTracker));
  // Start all dependent services
  this.serviceManager.startAsync().awaitHealthy(5, TimeUnit.SECONDS);

  startCancellationExecutor();

  jobLocalSetupTimer.stop();
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
protected StandardGobblinInstanceDriver(String instanceName, Configurable sysConfig,
    JobCatalog jobCatalog,
    JobSpecScheduler jobScheduler, JobExecutionLauncher jobLauncher,
    Optional<MetricContext> instanceMetricContext,
    Optional<Logger> log,
    List<GobblinInstancePluginFactory> plugins,
    SharedResourcesBroker<GobblinScopeTypes> instanceBroker) {
  super(instanceName, sysConfig, jobCatalog, jobScheduler, jobLauncher, instanceMetricContext, log, instanceBroker);
  List<Service> componentServices = new ArrayList<>();

  checkComponentService(getJobCatalog(), componentServices);
  checkComponentService(getJobScheduler(), componentServices);
  checkComponentService(getJobLauncher(), componentServices);

  _plugins = createPlugins(plugins, componentServices);

  if (componentServices.size() > 0) {
    _subservices = new ServiceManager(componentServices);
  }
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@BeforeClass
public void setUp() throws Exception {
  this.jobConfigDir =
      Files.createTempDirectory(String.format("gobblin-test_%s_job-conf", this.getClass().getSimpleName()))
          .toString();

  FileUtils.forceDeleteOnExit(new File(this.jobConfigDir));
  FileUtils.copyDirectory(new File(JOB_CONFIG_FILE_DIR), new File(jobConfigDir));

  Properties properties = new Properties();
  try (Reader schedulerPropsReader = new FileReader("gobblin-test/resource/gobblin.test.properties")) {
    properties.load(schedulerPropsReader);
  }
  properties.setProperty(ConfigurationKeys.JOB_CONFIG_FILE_DIR_KEY, jobConfigDir);
  properties.setProperty(ConfigurationKeys.JOB_CONFIG_FILE_GENERAL_PATH_KEY, jobConfigDir);
  properties.setProperty(ConfigurationKeys.JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL_KEY, "1000");
  properties.setProperty(ConfigurationKeys.METRICS_ENABLED_KEY, "false");

  SchedulerService quartzService = new SchedulerService(new Properties());
  this.jobScheduler = new JobScheduler(properties, quartzService);
  this.serviceManager = new ServiceManager(Lists.newArrayList(quartzService, this.jobScheduler));
  this.serviceManager.startAsync().awaitHealthy(10, TimeUnit.SECONDS);;
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@SuppressWarnings({"unchecked", "rawtypes"})
private static void guavaServices(final Env env, final Binder binder,
    final Set<Class<Service>> serviceTypes) {
  Consumer<Class> guavaService = klass -> {
    binder.bind(klass).asEagerSingleton();
    serviceTypes.add(klass);
  };

  serviceTypes.forEach(guavaService);
  // lazy service manager
  AtomicReference<ServiceManager> sm = new AtomicReference<>();
  Provider<ServiceManager> smProvider = () -> sm.get();
  binder.bind(ServiceManager.class).toProvider(smProvider);
  // ask Guice for services, create ServiceManager and start services
  env.onStart(r -> {
    List<Service> services = serviceTypes.stream()
        .map(r::require)
        .collect(Collectors.toList());
    sm.set(new ServiceManager(services));
    sm.get().startAsync().awaitHealthy();
  });
  // stop services
  env.onStop(() -> {
    sm.get().stopAsync().awaitStopped();
  });
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Inject
private GroningenWorkhorse(final Provider<Pipeline> pipelineProvider,
    final PipelineIdGenerator pipelineIdGenerator,
    final ServiceManager backgroundServices,
    final SystemAdapter systemAdapter,
    final Settings settings,
    final PipelineManager pipelineManager,
    final ProtoBufConfigManagerFactory protoBufConfigManagerFactory,
    final Build build) {
  this.backgroundServices = backgroundServices;
  this.systemAdapter = systemAdapter;
  this.pipelineManager = pipelineManager;
  this.settings = settings;
  this.protoBufConfigManagerFactory = protoBufConfigManagerFactory;
  this.build = build;
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@VisibleForTesting
ProcessTracker(
    BuckEventBus buckEventBus,
    InvocationInfo invocationInfo,
    ProcessHelper processHelper,
    ProcessRegistry processRegistry,
    boolean isDaemon,
    boolean deepEnabled) {
  this.eventBus = buckEventBus;
  this.invocationInfo = invocationInfo;
  this.serviceManager = new ServiceManager(ImmutableList.of(this));
  this.processHelper = processHelper;
  this.processRegistry = processRegistry;
  this.isDaemon = isDaemon;
  this.deepEnabled = deepEnabled;
  serviceManager.startAsync();
  this.processRegistry.subscribe(processRegisterCallback);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Override
public ServiceManager get() {
    final ImmutableSet<Service> allServices = ImmutableSet.<Service>builder()
            .addAll(services)
            .addAll(configuration.getServices())
            .build();

    return new ServiceManager(allServices);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@BeforeEach
public void setup(
    @OracleSettings Map<String, String> settings
) throws Exception {
  this.config = new OracleSourceConnectorConfig(settings);
  this.offsetStorageReader = mock(OffsetStorageReader.class);
  this.changeWriter = mock(ChangeWriter.class);
  this.queryService = new QueryService(this.config, this.offsetStorageReader, this.changeWriter);
  this.serviceManager = new ServiceManager(Arrays.asList(this.queryService));
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
/**
 * Create a new {@link ServiceManagerIface} that wraps a {@link ServiceManager}.
 *
 * @param delegate Service manager to delegate to.
 * @return A wrapper.
 */
public static ServiceManagerIface serviceManager(final ServiceManager delegate) {
  return new ServiceManagerIface() {
    @Override
    public ServiceManagerIface startAsync() {
      delegate.startAsync();
      return this;
    }

    @Override
    public void awaitHealthy() {
      delegate.awaitHealthy();
    }

    @Override
    public ServiceManagerIface stopAsync() {
      delegate.stopAsync();
      return this;
    }

    @Override
    public void awaitStopped(long timeout, TimeUnit unit) throws TimeoutException {
      delegate.awaitStopped(timeout, unit);
    }

    @Override
    public ImmutableMultimap<State, Service> servicesByState() {
      return delegate.servicesByState();
    }
  };
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Provides
@Singleton
@AppStartup
ServiceManagerIface provideAppStartupServiceManager(
    @AppStartup Set<Service> services,
    LifecycleShutdownListener listener) {

  ServiceManager manager = new ServiceManager(services);
  manager.addListener(listener);
  return GuavaUtils.serviceManager(manager);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Provides
@Singleton
@SchedulerActive
ServiceManagerIface provideSchedulerActiveServiceManager(
    @SchedulerActive Set<Service> services,
    LifecycleShutdownListener listener) {

  ServiceManager manager = new ServiceManager(services);
  manager.addListener(listener);
  return GuavaUtils.serviceManager(manager);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Bean(initMethod = "startAsync", destroyMethod = "stopAsync")
public ServiceManager serviceManager() {
    List<PoolingThreadedService> services = new ArrayList<>();
    services.addAll(requestParsingServices());
    services.addAll(downloadingServices());
    services.addAll(metaparsingServices());
    return new ServiceManager(services);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
public SchedulerDaemon(Properties defaultProperties, Properties customProperties)
    throws Exception {
  Properties properties = new Properties();
  properties.putAll(defaultProperties);
  properties.putAll(customProperties);

  List<Service> services = Lists.<Service>newArrayList(new JobScheduler(properties));
  boolean jobExecInfoServerEnabled = Boolean
      .valueOf(properties.getProperty(ConfigurationKeys.JOB_EXECINFO_SERVER_ENABLED_KEY, Boolean.FALSE.toString()));
  if (jobExecInfoServerEnabled) {
    services.add(new JobExecutionInfoServer(properties));
  }
  this.serviceManager = new ServiceManager(services);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@BeforeClass
public void setUp()
    throws Exception {
  Properties properties = new Properties();
  properties.load(new FileReader("gobblin-test/resource/gobblin.test.properties"));
  properties.setProperty(ConfigurationKeys.JOB_CONFIG_FILE_DIR_KEY, JOB_CONFIG_FILE_DIR);
  properties.setProperty(ConfigurationKeys.JOB_CONFIG_FILE_MONITOR_POLLING_INTERVAL_KEY, "1000");
  properties.setProperty(ConfigurationKeys.METRICS_ENABLED_KEY, "false");

  this.jobScheduler = new JobScheduler(properties);
  this.serviceManager = new ServiceManager(Lists.newArrayList(this.jobScheduler));
  this.serviceManager.startAsync();
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Inject
DefaultCultivarStartStopManager(@Curator final CuratorManagementService curatorManagementService,
        @Cultivar final ServiceManager serviceManager) {
    this.curatorManagementService = curatorManagementService;
    this.serviceManager = serviceManager;

}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
@Before
public void setUp() {
    /*
     * Multiple things in AbstractIdleService and ServiceManager are final, necessitating indirect methods of seeing
     * if the service has been started up or shut down.
     */
    Service service = new AbstractIdleService() {
        @Override
        protected void startUp() throws Exception {
            logger.info("startUp");
        }

        @Override
        protected void shutDown() throws Exception {
            logger.info("shutDown");
        }

        @Override
        protected Executor executor() {
            return MoreExecutors.sameThreadExecutor();
        }
    };

    serviceManager = new ServiceManager(ImmutableSet.of(service));

    when(curatorManagementService.startAsync()).thenReturn(curatorManagementService);
    when(curatorManagementService.stopAsync()).thenReturn(curatorManagementService);

    startStopManager = new DefaultCultivarStartStopManager(curatorManagementService, serviceManager);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
private void getServices() {
  final Properties properties = ConfigUtils.configToProperties(this.clusterConfig);
  this.taskExecutor = new TaskExecutor(properties);
  this.taskStateTracker = new GobblinHelixTaskStateTracker(properties);

  final List<Service> services = Lists.newArrayList(this.taskExecutor, this.taskStateTracker);
  this.serviceManager = new ServiceManager(services);
}
 

import com.google.common.util.concurrent.ServiceManager; //导入依赖的package包/类
public GobblinTaskRunner(String applicationName, String helixInstanceName, String applicationId,
    String taskRunnerId, Config config, Optional<Path> appWorkDirOptional)
    throws Exception {
  this.helixInstanceName = helixInstanceName;
  this.taskRunnerId = taskRunnerId;
  this.applicationName = applicationName;
  this.applicationId = applicationId;

  Configuration conf = HadoopUtils.newConfiguration();
  this.fs = buildFileSystem(config, conf);

  this.appWorkPath = initAppWorkDir(config, appWorkDirOptional);

  this.config = saveConfigToFile(config);

  initHelixManager();

  this.containerMetrics = buildContainerMetrics();

  this.taskStateModelFactory = registerHelixTaskFactory();

  services.addAll(getServices());
  if (services.isEmpty()) {
    this.serviceManager = null;
  } else {
    this.serviceManager = new ServiceManager(services);
  }

  logger.debug("GobblinTaskRunner: applicationName {}, helixInstanceName {}, applicationId {}, taskRunnerId {}, config {}, appWorkDir {}",
      applicationName, helixInstanceName, applicationId, taskRunnerId, config, appWorkDirOptional);
}