Java 类com.google.common.util.concurrent.AbstractScheduledService 实例源码

@Override
protected void configure() {
  install(new PrivateModule() {
    @Override
    protected void configure() {
      bind(RowGarbageCollector.class).in(Singleton.class);
      bind(AbstractScheduledService.Scheduler.class).toInstance(
          AbstractScheduledService.Scheduler.newFixedRateSchedule(
              0L,
              DB_ROW_GC_INTERVAL.get().getValue(),
              DB_ROW_GC_INTERVAL.get().getUnit().getTimeUnit()));
      expose(RowGarbageCollector.class);
    }
  });
  SchedulerServicesModule.addSchedulerActiveServiceBinding(binder())
      .to(RowGarbageCollector.class);
}

/**
 * Set the bound instance to update from the provider at a certain delay.
 * 
 * @param time
 *            The amount of time to spend before the first registration and between registrations.
 * @param unit
 *            The time unit for time.
 * @param service
 *            An optional ScheduledExecutorService. By default this will be an exitingScheduledExecutorService with
 *            default parameters.
 */
@SuppressWarnings("unchecked")
public RegistrationServiceModuleBuilder<T> updating(@Nonnegative final long time, final TimeUnit unit,
        final ScheduledExecutorService service) {
    checkArgument(time > 0, "Time must be positive.");
    checkNotNull(unit, "TimeUnit must not be null.");
    checkNotNull(service, "Provided executor service must not be null.");

    scheduler = AbstractScheduledService.Scheduler.newFixedDelaySchedule(time, time, unit);

    executorService = service;

    bindType = UpdatingRegistrationService.class;

    return this;
}

@Test
public void testSelfStoppingService() {
    // If the delegate service stops itself then it gets restarted after reacquireDelay.
    int reacquireDelayMillis = 1500;
    ServiceTriggers triggers1 = new ServiceTriggers();
    ServiceTriggers triggers2 = new ServiceTriggers();
    LeaderService leader = newLeaderService(reacquireDelayMillis, TimeUnit.MILLISECONDS, supply(
            triggers1.listenTo(new AbstractScheduledService() {
                @Override
                protected void runOneIteration() throws Exception {
                    stopAsync();
                }

                @Override
                protected Scheduler scheduler() {
                    return Scheduler.newFixedDelaySchedule(10, 10, TimeUnit.MILLISECONDS);
                }
            }),
            triggers2.listenTo(new NopService())));
    leader.startAsync();

    assertTrue(triggers1.getRunning().firedWithin(1, TimeUnit.MINUTES));
    assertTrue(triggers1.getTerminated().firedWithin(1, TimeUnit.MINUTES));
    assertTrue(triggers2.getRunning().firedWithin(1, TimeUnit.MINUTES));
}

@Override
protected Scheduler scheduler() {

    final long guardLeasePeriodInMs = leasePeriodInMs / 4;

    return new AbstractScheduledService.CustomScheduler() {

        @Override
        protected Schedule getNextSchedule() throws Exception {
            if (!haveLease()) {
                // Get the current node version...
                Stat stat = zkClient.checkExists().forPath(leasePath);
                leaseNodeVersion = stat.getVersion();
                LOG.trace("{} will try to get lease (with Ver. {}) in {}ms", tsoHostAndPort, leaseNodeVersion,
                          leasePeriodInMs);
                // ...and wait the lease period
                return new Schedule(leasePeriodInMs, TimeUnit.MILLISECONDS);
            } else {
                long waitTimeInMs = getEndLeaseInMs() - System.currentTimeMillis() - guardLeasePeriodInMs;
                LOG.trace("{} will try to renew lease (with Ver. {}) in {}ms", tsoHostAndPort,
                          leaseNodeVersion, waitTimeInMs);
                return new Schedule(waitTimeInMs, TimeUnit.MILLISECONDS);
            }
        }
    };

}

public Builder addMotionEventConsumer(MotionEventConsumer motionEventConsumer) {
    if (motionEventConsumer instanceof Service) {
        AbstractScheduledService consumer = (AbstractScheduledService) motionEventConsumer;
        if (!consumer.isRunning()) {
            consumer.startAsync();
        }
    }
    this.motionEventConsumers.add(motionEventConsumer);
    return this;
}

@Override
protected void configure() {
  install(new PrivateModule() {
    @Override
    protected void configure() {
      if (enablePreemptor) {
        LOG.info("Preemptor Enabled.");
        bind(PreemptorMetrics.class).in(Singleton.class);
        bind(PreemptionVictimFilter.class)
            .to(PreemptionVictimFilter.PreemptionVictimFilterImpl.class);
        bind(PreemptionVictimFilter.PreemptionVictimFilterImpl.class).in(Singleton.class);
        bind(Preemptor.class).to(Preemptor.PreemptorImpl.class);
        bind(Preemptor.PreemptorImpl.class).in(Singleton.class);
        bind(new TypeLiteral<Amount<Long, Time>>() { })
            .annotatedWith(PendingTaskProcessor.PreemptionDelay.class)
            .toInstance(preemptionDelay);
        bind(BiCacheSettings.class).toInstance(
            new BiCacheSettings(PREEMPTION_SLOT_HOLD_TIME.get(), "preemption_slot_cache_size"));
        bind(new TypeLiteral<BiCache<PreemptionProposal, TaskGroupKey>>() { })
            .in(Singleton.class);
        bind(PendingTaskProcessor.class).in(Singleton.class);
        bind(ClusterState.class).to(ClusterStateImpl.class);
        bind(ClusterStateImpl.class).in(Singleton.class);
        expose(ClusterStateImpl.class);

        bind(PreemptorService.class).in(Singleton.class);
        bind(AbstractScheduledService.Scheduler.class).toInstance(
            AbstractScheduledService.Scheduler.newFixedRateSchedule(
                0L,
                slotSearchInterval.getValue(),
                slotSearchInterval.getUnit().getTimeUnit()));

        expose(PreemptorService.class);
        expose(PendingTaskProcessor.class);
      } else {
        bind(Preemptor.class).toInstance(NULL_PREEMPTOR);
        LOG.warn("Preemptor Disabled.");
      }
      expose(Preemptor.class);
    }
  });

  // We can't do this in the private module due to the known conflict between multibindings
  // and private modules due to multiple injectors.  We accept the added complexity here to keep
  // the other bindings private.
  PubsubEventModule.bindSubscriber(binder(), ClusterStateImpl.class);
  if (enablePreemptor) {
    SchedulerServicesModule.addSchedulerActiveServiceBinding(binder())
        .to(PreemptorService.class);
  }
}

PipelineRunner(EventBus eventBus,
               Supplier<ImmutableList<Source>> sourceSupplier,
               Supplier<ImmutableList<Step>> stepSupplier,
               Timer.Factory timerFactory) {
  this.sourceSupplier = sourceSupplier;
  this.stepSupplier = stepSupplier;
  Timer timer = timerFactory.create(this);
  this.pipelineService = new AutoRestartingService<>(
      () -> new AbstractScheduledService() {

        /**
         *
         * @throws InterruptedException This should never happen.
         */
        @Override
        protected void runOneIteration() throws InterruptedException {
          if (!super.isRunning()) {
            return;
          }

          pipelineFlag.acquire();
          eventBus.post(new RunStartedEvent());

          if (!super.isRunning()) {
            return;
          }
          timer.time(() -> runPipeline(super::isRunning));
          // This should not block access to the steps array
          eventBus.post(new RunStoppedEvent());
          if (super.isRunning()) {
            eventBus.post(new RenderEvent());
          }
        }

        @Override
        protected Scheduler scheduler() {
          return Scheduler.newFixedRateSchedule(0, 1, TimeUnit.MILLISECONDS);
        }

        @Override
        protected String serviceName() {
          return "Pipeline Runner Service";
        }
      }
  );
  this.pipelineService.addListener(new LoggingListener(logger, PipelineRunner.class),
      MoreExecutors.directExecutor());
}

public Module build() {
    checkState(targetHolder != null, "Target annotation has not been set.");
    checkState(discoveryHolder != null, "Discovery annotation has not been set.");
    checkState(binderAssistant != null, "No provider provided.");

    return new AbstractModule() {
        @SuppressWarnings("unchecked")
        @Override
        protected void configure() {
            final Key<RegistrationService<T>> registrationServiceKey = (Key<RegistrationService<T>>) targetHolder
                    .generateKey(Types.newParameterizedType(RegistrationService.class, clazz));

            install(new PrivateModule() {
                @SuppressWarnings("unchecked")
                @Override
                protected void configure() {

                    if (scheduler != null) {
                        bind(AbstractScheduledService.Scheduler.class).annotatedWith(Private.class).toInstance(
                                scheduler);
                        bind(ScheduledExecutorService.class).annotatedWith(Private.class).toInstance(
                                executorService);
                    }

                    bind(
                            (Key<ServiceDiscovery<T>>) Key.get(
                                    Types.newParameterizedType(ServiceDiscovery.class, clazz), Private.class)).to(
                            (Key<ServiceDiscovery<T>>) discoveryHolder.generateKey(Types.newParameterizedType(
                                    ServiceDiscovery.class, clazz)));

                    binderAssistant.bindToProvider(
                            binder(),
                            (Key<ServiceInstance<T>>) Key.get(
                                    Types.newParameterizedType(ServiceInstance.class, clazz), Private.class));

                    bind(registrationServiceKey).to(
                            (Key<? extends RegistrationService<T>>) Key.get(Types.newParameterizedType(bindType,
                                    clazz))).in(Scopes.SINGLETON);

                    expose(registrationServiceKey);
                }
            });

            Multibinder<RegistrationService<?>> services = Multibinder.newSetBinder(binder(),
                    new TypeLiteral<RegistrationService<?>>() {
                    });

            services.addBinding().to(registrationServiceKey);

        }
    };
}

public Schedule scheduleFixedRate(long initialDelay, long period, TimeUnit timeUnit) {
  return new Schedule(AbstractScheduledService.Scheduler.newFixedRateSchedule(initialDelay, period, timeUnit));
}

public Schedule scheduleFixedDelay(long initialDelay, long delay, TimeUnit timeUnit) {
  return new Schedule(AbstractScheduledService.Scheduler.newFixedDelaySchedule(initialDelay, delay, timeUnit));
}

@Override
protected void runDaemon(Injector injector) throws Exception {
    // Tracking of live data
    final TargetTracker targetTracker = new TargetTracker();

    // A cache manager where caches can be held and retrieved
    final CacheManager cacheManager = new CacheManager();

    // Setup the readers
    AisReaderGroup g = AisReaders.createGroup("AisView",
            sources == null ? Collections.<String> emptyList() : sources);
    AisReaders.manageGroup(g);

    // A job manager that takes care of tracking ongoing jobs
    final JobManager jobManager = new JobManager();

    // Setup the backup process
    // Files.createDirectories(backup);
    backup.mkdirs();
    if(!backup.isDirectory()) {
        throw new IOException("Unable to create directories for " + backup);
    }

    start(new TargetTrackerFileBackupService(targetTracker, backup.toPath()));

    // start tracking
    targetTracker.subscribeToPacketStream(g.stream());

    //target tracking cleanup service
    start(new AbstractScheduledService() {

        @Override
        protected Scheduler scheduler() {
            return Scheduler.newFixedDelaySchedule(1, 10, TimeUnit.MINUTES);
        }

        @Override
        protected void runOneIteration() throws Exception {
            final Date satellite = new Date(new Date().getTime()-(1000*60*60*48));
            final Date live = new Date(new Date().getTime()-(1000*60*60*12));

            targetTracker.removeAll((t, u) -> {
                switch(t.getSourceType()) {
                case SATELLITE:
                    return !u.hasPositionInfo() || new Date(u.getPositionTimestamp()).before(satellite);
                default:
                    return !u.hasPositionInfo() || new Date(u.getPositionTimestamp()).before(live);
                }
            });
        }
    });

    //target cleanup missing static data
    start(new AbstractScheduledService() {
        @Override
        protected Scheduler scheduler() {
            return Scheduler.newFixedDelaySchedule(1, 24, TimeUnit.HOURS);
        }

        @Override
        protected void runOneIteration() throws Exception {
            targetTracker.removeAll(u -> !u.hasStaticInfo());
        }
    });

    start(g.asService());

    // Start Ais Store Connection
    final CassandraConnection con = connect();

    WebServer ws = new WebServer(port);
    ws.getContext().setAttribute(
            AbstractResource.CONFIG,
            AbstractResource.create(g, con, targetTracker, cacheManager, jobManager));

    ws.start();
    LOG.info("AisView started");
    ws.join();
}

@Override
protected AbstractScheduledService.Scheduler scheduler() {
  return Scheduler.newFixedRateSchedule(0L, 1000L, TimeUnit.MILLISECONDS);
}

@Override
protected AbstractScheduledService.Scheduler scheduler() {
  return Scheduler.newFixedRateSchedule(1L, 1L, TimeUnit.SECONDS);
}

private Schedule(AbstractScheduledService.Scheduler scheduler) {this.scheduler = scheduler;}