import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public Consul get() {

    String consulUrl = configuration.getConsulUrl();

    try {
        Consul.Builder builder = Consul.builder().withUrl(consulUrl);

        if (System.getenv(Constants.CONSUL_TOKEN_ENVIRONMENT) != null) {
            builder.withAclToken(Constants.CONSUL_TOKEN_ENVIRONMENT);
            log.info("connecting to consul with url '{}' using http token", consulUrl);
        } else {
            log.info("connecting to consul with url '{}'", consulUrl);
        }

        return builder.build();

    } catch (ConsulException e) {
        log.error("unable to connect to consul", e);
        throw new UnrecoverableException();
    }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
private void registerSelf() throws MalformedURLException {
  if (clientOnly) {
    LOG.debug("Client only. Not registering self with consul");
    return;
  }
  LOG.info("Registering self with Consul");
  Registration.RegCheck ttlCheck = ImmutableRegCheck.builder().ttl("60s").deregisterCriticalServiceAfter("1m").build();
  Registration registration =
    ImmutableRegistration.builder().address(consulServiceAddress).port(Integer.parseInt(consulServicePort))
      .name(consulServiceName).id(consulServiceId).check(ttlCheck).build();
  if (CONSUL_HOLDER != null && CONSUL_HOLDER.hasInstance()) {
    CONSUL_HOLDER.registerSelf(registration);
  } else {
    CONSUL_HOLDER = ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), this);
    CONSUL_HOLDER.registerSelf(registration);
  }
  Runtime.getRuntime().removeShutdownHook(shutdownHook);
  Runtime.getRuntime().addShutdownHook(shutdownHook);
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public Config initialize(final Config config) {
    try {
        visibleHost = getConfigPropertyString(config, "tsd.discovery.visble_host", "localhost");
        visiblePort = getConfigPropertyInt(config, "tsd.discovery.visble_port", 4242);
        serviceName = getConfigPropertyString(config, "tsd.discovery.service_name", "OpenTSDB");
        serviceId   = getConfigPropertyString(config, "tsd.discovery.service_id", "opentsdb");
        tsdMode     = getConfigPropertyString(config, "tsd.mode", "ro");

        String consulUrl = getConfigPropertyString(config, "tsd.discovery.consul_url", "http://localhost:8500");

        LOGGER.debug("Finished with config");

        consul = Consul.builder().withUrl(consulUrl).build();
        LOGGER.info("Consul ServiceDiscovery Plugin Initialized");

        updateZookeeperQuorum(config);

    } catch (Exception e) {
        LOGGER.error("Could not register this instance with Consul", e);
    }
    return config;
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public void start(StartContext startContext) throws StartException {

    Consul.Builder builder = Consul.builder();


    // pool because of multiple threads.
    ResteasyClientBuilder clientBuilder = new ResteasyClientBuilder();
    clientBuilder = clientBuilder.connectionPoolSize(20);

    builder.withClientBuilder(clientBuilder);
    builder.withUrl(this.url);

    try {
        this.consul = builder.build();
    } catch (Exception e) {
        throw new StartException("Failed to connect consul at " + url, e);
    }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public void initialize(Bootstrap<?> bootstrap) {
    // Replace variables with values from Consul KV. Please override
    // getConsulAgentHost() and getConsulAgentPort() if Consul is not
    // listening on the default localhost:8500.
    try {
        final Consul consul = Consul.builder().withHostAndPort(HostAndPort
                .fromParts(getConsulAgentHost(), getConsulAgentPort()))
                .build();
        bootstrap.setConfigurationSourceProvider(
                new SubstitutingSourceProvider(
                        bootstrap.getConfigurationSourceProvider(),
                        new ConsulSubstitutor(consul, strict,
                                substitutionInVariables)));
    } catch (ConsulException e) {
        LOGGER.warn(
                "Unable to query Consul running on {}:{},"
                        + " disabling configuration subsitution",
                getConsulAgentHost(), getConsulAgentPort(), e);
    }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
protected void setupEnvironment(ConsulFactory consulConfig, Environment environment) {
    final Consul consul = consulConfig.build();
    final String serviceId = getConsulServiceId();
    final ConsulAdvertiser advertiser = new ConsulAdvertiser(environment,
            consulConfig, consul, serviceId);

    // Register a Jetty listener to get the listening host and port
    environment.lifecycle().addServerLifecycleListener(
            new ConsulServiceListener(advertiser));

    // Register a ping healthcheck to the Consul agent
    environment.healthChecks().register("consul",
            new ConsulHealthCheck(consul));

    // Register a shutdown manager to deregister the service
    environment.lifecycle().manage(new ConsulAdvertiserManager(advertiser));

    // Add an administrative task to toggle maintenance mode
    environment.admin().addTask(new MaintenanceTask(consul, serviceId));
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public ConsulRoutePolicy(Consul consul) {
    this.consul = consul;
    this.sessionClient = consul.sessionClient();
    this.keyValueClient = consul.keyValueClient();
    this.suspendedRoutes =  new HashSet<>();
    this.leader = new AtomicBoolean(false);
    this.lock = new Object();
    this.index = new AtomicReference<>(BigInteger.valueOf(0));
    this.serviceName = null;
    this.servicePath = null;
    this.ttl = 60;
    this.lockDelay = 10;
    this.executorService = null;
    this.shouldStopConsumer = true;
    this.sessionId = null;
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
/**
 * @param consul            Consul-client
 * @param prefix            prefix for the key-value storage (Default /container)
 * @param useRemoteServices use services from remote consul nodes
 */
public ConsulMetadataStorage(Consul consul, String prefix, boolean useRemoteServices)
{
    this.consul = consul;
    if (null == prefix || 0 == prefix.length()) {
        prefix = "container";
    }
    this.prefix = prefix;
    this.useRemoteServices = useRemoteServices;
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public void start(StartContext startContext) throws StartException {

    Consul.Builder builder = Consul.builder();


    // pool because of multiple threads.
    ResteasyClientBuilder clientBuilder = new ResteasyClientBuilder();
    clientBuilder = clientBuilder.connectionPoolSize(20);

    builder.withClientBuilder(clientBuilder);
    builder.withUrl(this.url);

    try {
        this.consul = builder.build();
    } catch (Exception e) {
        throw new StartException("Failed to connect consul at "+url, e);
    }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
private Consul getConsul() {
    // test that the client is active - the purpose of this test is to make sure that the instance of consul in the
    // cluster is alive, is it's not we'll create a new one before giving up. The assumption is that the hostname of
    // consul is really a consul service that hides multiple instances of consul servers.
    RetryTemplate retryTemplate = new RetryTemplate();
    retryTemplate.setBackOffPolicy(new FixedBackOffPolicy());
    TimeoutRetryPolicy retryPolicy = new TimeoutRetryPolicy();
    retryPolicy.setTimeout(10000);
    retryTemplate.setRetryPolicy(retryPolicy);

    return retryTemplate.execute(context -> {
        if (consul == null || consul.statusClient() == null || isEmpty(consul.statusClient().getLeader())) {
            // if we can't find a leader this means that the consul client is not usable - we'll release it
            log.warn("couldn't verify connection to Consul");
            log.info("creating a new Consul client");
            consul = newClient(consulProperties.getHostname(), consulProperties.getHttpPort());

            // throwing this exception to retry according to the policies. The last retry will throw this exception
            throw new IllegalStateException("Consul connection could not be verified");
        } else {
            log.info("Consul connection verified");
            return consul;
        }
    });
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public void configure(final Binder binder) {
    loading();

    cfgs.forEach((id, cfg) -> {
        final Consul consul = build(cfg);
        binder.bind(Consul.class).annotatedWith(Names.named(CONSUL_PREFIX + id)).toInstance(consul);
        binder.bind(AgentClient.class).annotatedWith(Names.named(CONSUL_AGENT_CLIENT_PREFIX + id)).toInstance(consul.agentClient());
        binder.bind(HealthClient.class).annotatedWith(Names.named(CONSUL_HEALTH_CLIENT_PREFIX + id)).toInstance(consul.healthClient());
        binder.bind(KeyValueClient.class).annotatedWith(Names.named(CONSUL_KEY_VALUE_CLIENT_PREFIX + id)).toInstance(consul.keyValueClient());
        binder.bind(CatalogClient.class).annotatedWith(Names.named(CONSUL_CATALOG_CLIENT_PREFIX + id)).toInstance(consul.catalogClient());
        binder.bind(StatusClient.class).annotatedWith(Names.named(CONSUL_STATUS_CLIENT_PREFIX + id)).toInstance(consul.statusClient());
        binder.bind(SessionClient.class).annotatedWith(Names.named(CONSUL_SESSION_CLIENT_PREFIX + id)).toInstance(consul.sessionClient());
        binder.bind(EventClient.class).annotatedWith(Names.named(CONSUL_EVENT_CLIENT_PREFIX + id)).toInstance(consul.eventClient());
        binder.bind(PreparedQueryClient.class).annotatedWith(Names.named(CONSUL_PREPARED_QUERY_CLIENT_PREFIX + id))
                .toInstance(consul.preparedQueryClient());
        binder.bind(CoordinateClient.class).annotatedWith(Names.named(CONSUL_COORDINATE_CLIENT_PREFIX + id))
                .toInstance(consul.coordinateClient());
        binder.bind(OperatorClient.class).annotatedWith(Names.named(CONSUL_OPERATOR_CLIENT + id)).toInstance(consul.operatorClient());
    });

}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Inject
public ConsulDiscoveryCallback(final Consul consul,
                               final Configuration configuration,
                               final PluginExecutorService pluginExecutorService) {
    this.consul = consul;
    this.configuration = configuration;
    this.pluginExecutorService = pluginExecutorService;
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
@Override
public void start() throws Exception {
  LOG.info("Starting Consul Discovery Agent for " + basePath);
  taskRunner = new TaskRunnerFactory("Consul Discovery Agent");
  taskRunner.init();
  running.set(true);

  if (CONSUL_HOLDER == null || !CONSUL_HOLDER.hasInstance()) {
    CONSUL_HOLDER = ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), this);
  } else {
    CONSUL_HOLDER = ConsulHolder.instance();
  }
  registerSelf();

  executorService = Executors.newSingleThreadScheduledExecutor();
  executorService.scheduleWithFixedDelay(new Runnable() {
    @Override
    public void run() {
      if (running.get()) {
        try {
          if (CONSUL_HOLDER == null || !CONSUL_HOLDER.hasInstance()) {
            CONSUL_HOLDER =
              ConsulHolder.initialize(Consul.builder().withUrl(basePath.toURL()).build(), ConsulDiscoveryAgent.this);
          }
          checkTtl();

          List<ServiceHealth> services = CONSUL_HOLDER.getServices(consulServiceName);
          handleServices(services);
        } catch (Throwable t) {
          //We don't really want this thread to ever die as long as the agent is running, so we'll just log any
          // errors.
          LOG.warn("Failure in Consul service discovery", t);
        }
      } else {
        Thread.currentThread().interrupt();
      }
    }
  }, 0, pollingInterval, TimeUnit.MILLISECONDS);
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
synchronized static ConsulHolder initialize(Consul consul, ConsulDiscoveryAgent consulDiscoveryAgent) {
  synchronized (consulLock) {
    if (ConsulHolder.instance != null) {
      throw new ConsulDiscoveryException("Already intitialized");
    }
    instance = new ConsulHolder(consul, consulDiscoveryAgent);
    return instance;
  }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public void start() {
    consulContainer.start();
    consulContainer.followOutput(new Slf4jLogConsumer(logger).withPrefix("consul"));
    await("Consul is ready")
            .atMost(10, TimeUnit.SECONDS)
            .until(() -> {
                try {
                    consul.set(Consul.builder().withUrl(getConsulUrl()).build());
                    return true;
                } catch (Exception e) {
                    logger.info("Failed to create consul client, probably consul is not ready yet", e);
                    return false;
                }
            });
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public Consul getConsul() {
    if (consul.get() == null) {
        throw new IllegalStateException("Standalone consul must be started before calling getConsul");
    }

    return consul.get();
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
private static void registerToConsul(Config config, InetSocketAddress socketAddress) {
  try {
    final Consul consul = Consul.builder().withHostAndPort(HostAndPort.fromParts(config.consulHost, config.consulPort)).build();
    final String serviceId = consulTag + "-" + UUID.randomUUID();
    System.out.println("Registering consul service " + serviceId);
    if (!config.consulMode.equals("agent")) {
      Service service = ImmutableService.builder()
          .id(serviceId)
          .service(consulTag)
          .port(socketAddress.getPort())
          .address(InetAddress.getLocalHost().getHostName())
          .build();

      CatalogRegistration registration = ImmutableCatalogRegistration.builder()
          .datacenter("dc1")
          .node("node1")
          .service(service)
          .address(service.getAddress())
          .build();
      consul.catalogClient().register(registration);
    } else {
      URL url = new URL("http://" + socketAddress.getHostName() + ":" + socketAddress.getPort());
      consul.agentClient().register(config.port, url, 10, consulTag, serviceId);
    }

    Runtime.getRuntime().addShutdownHook(new Thread() {

      public void run() {
        System.out.println("Deregistering consul service: " + serviceId);
        if (serviceId != null) {
          consul.agentClient().deregister(serviceId);
        }
      }
    });
  } catch (Exception ex) {
    System.out.println("Consul is not available.");
    ex.printStackTrace();
  }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public void run(String... args) throws InterruptedException, IOException {
  Consul consul = consulWithRetry();
  if (consul != null) {
    KeyValueClient kvClient = consul.keyValueClient();
    if (isReset) {
      kvClient.deleteKeys(globalPrefix);
    } else if (!isConfigInitialized(kvClient)) {
      loadConfigFromPath(kvClient);
    }
  }
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public Consul consulWithRetry() throws InterruptedException {
  int fails = 0;
  LOGGER.info("Connecting to Consul at %s: %d", host, port);
  Consul consul;
  while (fails < failLimit) {
    consul = getConsul();
    if (consul != null)
      return consul;
    fails++;
    Thread.sleep(waitTimeBetweenFails * 1000L);
  }
  LOGGER.error("Failed to connect to Consul in allowed retries.");
  return null;
}
 

import com.orbitz.consul.Consul; //导入依赖的package包/类
public Consul getConsul() {
  LOGGER.info("Attempting to get consul connection.");
  RestTemplate restTemplate = new RestTemplate();
  try {
    ResponseEntity<String> response = restTemplate
        .getForEntity(new URL(protocol, host, port, CONSUL_STATUS_PATH).toString(), String.class);
    if (response.getStatusCode().is2xxSuccessful())
      return Consul.builder().withUrl(new URL(protocol, host, port, "")).build();
    return null;
  } catch (RestClientException | MalformedURLException e) {
    LOGGER.error(
        "Malformed URL or other issue when attempting to get to Consul: " + e.getMessage());
    return null;
  }
}