import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
private static String buildConnectionString(List<CatalogService> serviceList) {
    StringBuilder stringBuilder = new StringBuilder();

    Iterator<CatalogService> serviceListIterator = serviceList.iterator();
    while (serviceListIterator.hasNext()) {
        CatalogService serviceNode = serviceListIterator.next();

        stringBuilder
                .append(serviceNode.getServiceAddress())
                .append(":")
                .append(serviceNode.getServicePort());

        if (serviceListIterator.hasNext()) {
            stringBuilder.append(",");
        }
    }

    return stringBuilder.toString();
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
private static List<CatalogService> getServiceNodes(String serviceName) {
    try {
        CatalogClient catalogClient = consul.catalogClient();
        ConsulResponse<List<CatalogService>> serviceResponse = catalogClient.getService(serviceName);
        return serviceResponse.getResponse();
    } catch (NullPointerException e) {
        LOGGER.error("Could not retrieve Consul Catalog Client");
        return null;
    }
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
private static void updateZookeeperQuorum(final Config config) {
    String zkQuorum;
    List<CatalogService> zookeeperService = null;

    zookeeperService = getServiceNodes("zookeeper-2181");
    if (zookeeperService.size() > 0) {
        zkQuorum = buildConnectionString(zookeeperService);
        LOGGER.info("Updated Zookeeper Quorum to " + zkQuorum);
        config.overrideConfig("tsd.storage.hbase.zk_quorum", zkQuorum);
    } else {
        LOGGER.info("Unable to locate zookeeper-2181 in Consul");
    }
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public List<ServiceCallServer> getUpdatedListOfServers(String name) {
    List<CatalogService> services = getCatalogClient()
        .getService(name, getCatalogOptions())
        .getResponse();

    List<ServiceHealth> healths = getHealthClient()
        .getAllServiceInstances(name, getCatalogOptions())
        .getResponse();

    return services.stream()
        .filter(service -> !hasFailingChecks(service, healths))
        .map(this::newServer)
        .collect(Collectors.toList());
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
/**
 * Get a list of all members.
 * 
 * @return a collection with all members
 */
@Override
protected Collection<String> getMembers() {

    Collection<String> list = new LinkedList<>();
    String name = consulConfig.getName();
    try {

        ConsulResponse<List<CatalogService>> service = this.agentClient.getService(name);

        logger.info("Resolving service: " + name);

        for (CatalogService s : service.getResponse()) {
            if (logger.isFinestEnabled()) {
                logger.finest("Found service at: " + s.getAddress());
            }
            list.add(s.getAddress());
        }

        if (list.isEmpty()) {
            logger.info("No services found!");
        }
        return list;
    } catch (Exception e) {
        logger.warning(e);
        throw ExceptionUtil.rethrow(e);
    }
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public Collection<DiscoveredNode> discoverNodes() {

    Collection<DiscoveredNode> list = new LinkedList<>();
    Properties empty = new Properties();

    if (this.agentClient != null) {
        try {

            ConsulResponse<List<CatalogService>> service = this.agentClient.getService(serviceName);

            LOG.log(Level.WARNING, "Resolving service: {0}", serviceName);

            for (CatalogService s : service.getResponse()) {
                if (LOG.isLoggable(Level.FINEST)) {
                    LOG.log(Level.FINEST, "Found service at: {0}", s.getAddress());
                }
                Address address = new Address(s.getAddress(), s.getServicePort());
                list.add(new SimpleDiscoveredNode(address, empty));
            }

            if (list.isEmpty()) {
                LOG.warning("No consul instances found!");
            }
            return list;
        } catch (Exception e) {
            LOG.severe(e.getMessage());
            throw ExceptionUtil.rethrow(e);
        }
    }
    return list;
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
/**
 * Look in consul for nodes with a given name. Exceptions are not rethrown
 * to prevent HazelCast of shutting down. This method returns an empty
 * list in case of error. When for example, consul returns to a normal state
 * a list of nodes is returned. HazelCast will try to connect to this
 * nodes.
 * 
 * @return A Collection of Nodes.
 */
@Override
public Collection<DiscoveryNode> discoverNodes() {

    Collection<DiscoveryNode> list = new LinkedList<>();
    try {

        ConsulResponse<List<CatalogService>> service
                = this.getClient().getService(serviceName);

        for (CatalogService s : service.getResponse()) {
            if (LOG.isLoggable(Level.FINEST)) {
                LOG.log(Level.FINEST, "Found service at: {0}", s.getAddress());
            }
            Address address = new Address(s.getAddress(), s.getServicePort());
            list.add(new SimpleDiscoveryNode(address));
        }

        LOG.log(Level.INFO, "Hazelcast found ''{0}'' instance(s) of ''{1}'' ", new Object[]{list.size(), serviceName});

        return list;
        
    } catch (Exception e) {
        // The exception is no longer rethrown. When the exception
        // arrives in Hazelcast-core, it will stop working totally.
        // An empty list is delivered as default. This method will be
        // invoked again after a certain time. When consul is available,
        // it will answer, otherwise we just retry.
        LOG.severe(e.getMessage());
    }

    return list;
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public Set<CatalogService> resolveByName(String name) {
    CatalogClient catalogClient = getConsul().catalogClient();

    List<CatalogService> catalogServices = catalogClient.getService(name).getResponse();
    Set<CatalogService> result = new TreeSet<>((o1, o2) -> o1.getServiceName().compareToIgnoreCase(o2.getServiceName()));
    result.addAll(catalogServices);

    return result;
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
protected Collection<ServiceInfo> _discover(CatalogClient catalogClient, 
                                            String serviceName, 
                                            Collection<Integer> ports,
                                            Collection<String> mustMatchTags) throws Exception {
    
    List<ServiceInfo> discoveredServices = new ArrayList<ServiceInfo>();
    
    ConsulResponse<List<CatalogService>> resp = catalogClient.getService(serviceName);
    List<CatalogService> serviceList = resp.getResponse();
    
    logger.trace("_discover() catalogClient.getService("+serviceName+") returned " + serviceList.size() + " results..");
    
    for (CatalogService srv : serviceList) {
    	
    	logger.trace("_discover() evaluating consul service: name:" + srv.getServiceName() + 
    				" serviceId:" + srv.getServiceId() + 
    				" servicePort:" + srv.getServicePort() +
    				" tags: " + Arrays.toString(srv.getServiceTags().toArray()));
        
        if (matchesTags(srv.getServiceTags(),mustMatchTags)) {
            
            try {
                // we parse mapped port from serviceId format "xx:yy:port"
                // registrator sets the serviceId = to this format above for each
                // unique port
                int mappedPort = Integer.valueOf(srv.getServiceId().split(":")[2]);

                // if we care about this mapped port... capture the service 
                if (ports.contains(mappedPort)) {
                    
                	InetAddress exposedAddress = null;
                	if (srv.getServiceAddress() != null) {
                		exposedAddress = InetAddress.getByName(srv.getServiceAddress());
                	} else {
                		// https://www.consul.io/docs/agent/http/catalog.html#ServiceAddress
                		logger.trace("_discover() CatalogService.serviceAddress is null... "
                				+ "falling back to address["+srv.getAddress()+"]");
                		exposedAddress = InetAddress.getByName(srv.getAddress());
                	}
                	
                    ServiceInfo info = new ServiceInfo(srv.getServiceName(),
                                                        srv.getServiceId(),
                                                        exposedAddress,
                                                        srv.getServicePort(),
                                                        mappedPort,
                                                        srv.getServiceTags());
                    discoveredServices.add(info);
                    
                    logger.debug("_discover() Discovered ServiceInfo: " + info);
                    
                } else {
                	logger.trace("_discover() serviceNameToFind=" + serviceName + 
                			", skipping consul service: " + srv.getServiceName() + 
                			" as its mappedPort[" + mappedPort + "] is not in list of "
                					+ "ports we care about: " + Arrays.toString(ports.toArray()) );;
                }
                
            } catch(Exception e) {
                throw new Exception("discover() Unexpected error processing "
                		+ "service: " + srv.getServiceName() + " " + e.getMessage(),e);
            }
            
        } else {
        	logger.trace("_discover() serviceNameToFind=" + serviceName + 
        			" skipping consul service: " + srv.getServiceName() + 
        			" with tags: " + (srv.getServiceTags() != null ? Arrays.toString(srv.getServiceTags().toArray()) : "[no tags]") + 
        			" as they don't contain mustMatchTags: " + Arrays.toString(mustMatchTags.toArray()));
        }
        
    }
    
    return discoveredServices;
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
protected boolean isCheckOnService(ServiceHealth check, CatalogService service) {
    return check.getService().getService().equalsIgnoreCase(service.getServiceName());
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
protected boolean hasFailingChecks(CatalogService service, List<ServiceHealth> healths) {
    return healths.stream().anyMatch(health -> isCheckOnService(health, service) && isNotHealthy(health));
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
protected ServiceCallServer newServer(CatalogService service) {
    return new DefaultServiceCallServer(
        service.getServiceAddress(),
        service.getServicePort()
    );
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public Iterable<DiscoveryNode> discoverNodes() {
	
	List<DiscoveryNode> toReturn = new ArrayList<DiscoveryNode>();
	
	try {
		// discover healthy nodes only? (and its NOT the first invocation...)
		if (this.consulHealthyOnly && discoverNodesInvoked) {
			
			List<ServiceHealth> nodes = consulHealthClient.getHealthyServiceInstances(consulServiceName, ConsulUtility.getAclToken(this.consulAclToken)).getResponse();
			
			for (ServiceHealth node : nodes) {
				toReturn.add(new SimpleDiscoveryNode(
								new Address(node.getService().getAddress(),node.getService().getPort())));
				getLogger().info("Discovered healthy node: " + node.getService().getAddress()+":"+node.getService().getPort());
			}
			
		// discover all services, regardless of health or this is the first invocation
		} else {
			
			ConsulResponse<List<CatalogService>> response = this.consulCatalogClient.getService(consulServiceName, ConsulUtility.getAclToken(this.consulAclToken));
			
			for (CatalogService service : response.getResponse()) {
				
				String discoveredAddress = null;
				String rawServiceAddress = service.getServiceAddress();
				String rawAddress = service.getAddress();
				
				if (rawServiceAddress != null && !rawServiceAddress.trim().isEmpty()) {
				    discoveredAddress = rawServiceAddress;
					
				} else if (rawAddress != null && !rawAddress.trim().isEmpty()) {
				    getLogger().warning("discoverNodes() ServiceAddress was null/blank! " +
						     "for service: " + service.getServiceName() + 
						     " falling back to Address value");
				    discoveredAddress = rawAddress;
					
				} else {
				    getLogger().warning("discoverNodes() could not discover an address, " +
						     "both ServiceAddress and Address were null/blank! " +
						     "for service: " + service.getServiceName());
				}
				
				toReturn.add(new SimpleDiscoveryNode(
						new Address(discoveredAddress, service.getServicePort())));
				getLogger().info("Discovered healthy node: " + discoveredAddress+":"+service.getServicePort());
			}
		}
		
	} catch(Exception e) {
		getLogger().severe("discoverNodes() unexpected error: " + e.getMessage(),e);
	}

	// flag we were invoked
	discoverNodesInvoked = true;
	
	return toReturn;
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public String resolveByNameAsClusterDefinition(String name) {
    Set<CatalogService> services = resolveByName(name);
    return services.stream().map(cs -> cs.getServiceAddress() + ":" + cs.getServicePort()).collect(joining(","));
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
@Override
public Set<CatalogService> resolveByName(String name) {
    return emptySet();
}
 

import com.orbitz.consul.model.catalog.CatalogService; //导入依赖的package包/类
/**
 * resolves a service by its name
 *
 * @param name the service name to lookup
 * @return a set of unique {@link CatalogService} objects
 */
Set<CatalogService> resolveByName(String name);