本文目录导读：
我来详细解析Java分布式数据发现API的服务机制和实现方式。
服务发现的核心架构
基础概念
// 服务注册与发现的核心接口
public interface ServiceDiscovery {
// 注册服务
void register(ServiceInstance instance) throws ServiceException;
// 注销服务
void deregister(String serviceId) throws ServiceException;
// 发现服务
List<ServiceInstance> discover(String serviceName) throws ServiceException;
// 监听服务变化
void watch(String serviceName, ServiceChangeListener listener);
}
主流实现方案
1 基于ZooKeeper的实现
@Service
public class ZookeeperServiceDiscovery implements ServiceDiscovery {
private final CuratorFramework client;
private final ServiceRegistry serviceRegistry;
private final Map<String, ServiceCache> serviceCaches = new ConcurrentHashMap<>();
public ZookeeperServiceDiscovery(CuratorFramework client) {
this.client = client;
this.serviceRegistry = new ServiceRegistryBuilder()
.client(client)
.basePath("/services")
.build();
// 初始化服务缓存
initServiceCaches();
}
@Override
public void register(ServiceInstance instance) {
try {
// 注册服务提供者
ServiceProvider provider = ServiceProvider.builder()
.id(instance.getId())
.serviceName(instance.getServiceName())
.address(instance.getHost() + ":" + instance.getPort())
.payload(instance.toJson())
.registrationTimeUTC(System.currentTimeMillis())
.build();
serviceRegistry.registerService(provider);
// 添加元数据
addServiceMetadata(instance);
log.info("Service registered: {} - {}",
instance.getServiceName(), instance.getId());
} catch (Exception e) {
throw new ServiceException("Failed to register service", e);
}
}
@Override
public List<ServiceInstance> discover(String serviceName) {
// 从缓存获取
ServiceCache cache = serviceCaches.get(serviceName);
if (cache != null) {
List<ServiceInstance> instances = cache.getInstances();
if (!instances.isEmpty()) {
return filterInstances(instances);
}
}
// 从ZooKeeper获取
return discoverFromZK(serviceName);
}
private List<ServiceInstance> discoverFromZK(String serviceName) {
try {
String path = "/services/" + serviceName;
List<String> children = client.getChildren().forPath(path);
return children.stream()
.map(child -> {
try {
byte[] data = client.getData().forPath(path + "/" + child);
return ServiceInstance.fromJson(new String(data));
} catch (Exception e) {
log.error("Failed to get service data", e);
return null;
}
})
.filter(Objects::nonNull)
.collect(Collectors.toList());
} catch (Exception e) {
throw new ServiceException("Failed to discover services", e);
}
}
@Override
public void watch(String serviceName, ServiceChangeListener listener) {
ServiceCache cache = serviceCaches.computeIfAbsent(serviceName,
k -> createServiceCache(serviceName));
cache.addListener(() -> {
List<ServiceInstance> instances = cache.getInstances();
listener.onChange(instances);
});
}
private ServiceCache createServiceCache(String serviceName) {
try {
ServiceCache cache = serviceRegistry.serviceCacheBuilder()
.name(serviceName)
.build();
cache.start();
return cache;
} catch (Exception e) {
throw new ServiceException("Failed to create service cache", e);
}
}
}
2 基于Consul的实现
@Component
public class ConsulServiceDiscovery implements ServiceDiscovery {
private final ConsulClient client;
private final ObjectMapper objectMapper;
@Autowired
public ConsulServiceDiscovery(ConsulProperties properties) {
this.client = new ConsulClient(properties.getHost(), properties.getPort());
this.objectMapper = new ObjectMapper();
}
@Override
public void register(ServiceInstance instance) {
try {
// 创建服务注册
NewService service = new NewService();
service.setId(instance.getId());
service.setName(instance.getServiceName());
service.setAddress(instance.getHost());
service.setPort(instance.getPort());
// 设置健康检查
String checkEndpoint = String.format("http://%s:%d/health",
instance.getHost(), instance.getPort());
NewService.Check check = new NewService.Check();
check.setHttp(checkEndpoint);
check.setInterval("10s");
check.setTimeout("3s");
check.setDeregisterCriticalServiceAfter("30s");
service.setCheck(check);
// 设置标签
service.setTags(instance.getTags());
// 注册服务
client.agentServiceRegister(service);
log.info("Service registered with Consul: {}", instance.getId());
} catch (Exception e) {
throw new ServiceException("Failed to register service with Consul", e);
}
}
@Override
public void deregister(String serviceId) {
try {
client.agentServiceDeregister(serviceId);
log.info("Service deregistered: {}", serviceId);
} catch (Exception e) {
throw new ServiceException("Failed to deregister service", e);
}
}
@Override
public List<ServiceInstance> discover(String serviceName) {
try {
// 查询健康的服务实例
Response<List<HealthService>> response =
client.getHealthServices(serviceName, true, QueryParams.DEFAULT);
return response.getValue().stream()
.map(this::convertToInstance)
.collect(Collectors.toList());
} catch (Exception e) {
throw new ServiceException("Failed to discover services", e);
}
}
private ServiceInstance convertToInstance(HealthService healthService) {
Service service = healthService.getService();
return ServiceInstance.builder()
.id(service.getId())
.serviceName(service.getService())
.host(service.getAddress())
.port(service.getPort())
.tags(new ArrayList<>(service.getTags()))
.metadata(extractMetadata(service))
.healthy(checkHealthStatus(healthService))
.build();
}
}
3 基于Eureka的实现
@Component
public class EurekaServiceDiscovery implements ServiceDiscovery {
private final EurekaClient eurekaClient;
private final ApplicationInfoManager applicationInfoManager;
@Autowired
public EurekaServiceDiscovery(EurekaClient eurekaClient,
ApplicationInfoManager applicationInfoManager) {
this.eurekaClient = eurekaClient;
this.applicationInfoManager = applicationInfoManager;
}
@Override
public void register(ServiceInstance instance) {
try {
// 创建实例信息
InstanceInfo instanceInfo = InstanceInfo.Builder.newBuilder()
.setInstanceId(instance.getId())
.setAppName(instance.getServiceName())
.setHostName(instance.getHost())
.setPort(instance.getPort())
.setIPAddr(instance.getIp())
.setStatus(InstanceStatus.UP)
.setLeaseInfo(LeaseInfo.Builder.newBuilder()
.setRenewalIntervalInSecs(30)
.setDurationInSecs(90)
.build())
.add("metadata", instance.getMetadata() != null ?
instance.getMetadata().toString() : "")
.build();
// 注册到Eureka
applicationInfoManager.registerAppInstance(instanceInfo);
} catch (Exception e) {
throw new ServiceException("Failed to register with Eureka", e);
}
}
@Override
public List<ServiceInstance> discover(String serviceName) {
try {
Application application = eurekaClient.getApplication(serviceName);
if (application == null) {
return Collections.emptyList();
}
return application.getInstances().stream()
.filter(instance -> InstanceStatus.UP.equals(instance.getStatus()))
.map(this::convertToInstance)
.collect(Collectors.toList());
} catch (Exception e) {
throw new ServiceException("Failed to discover services from Eureka", e);
}
}
private ServiceInstance convertToInstance(InstanceInfo instanceInfo) {
return ServiceInstance.builder()
.id(instanceInfo.getInstanceId())
.serviceName(instanceInfo.getAppName())
.host(instanceInfo.getHostName())
.ip(instanceInfo.getIPAddr())
.port(instanceInfo.getPort())
.securePort(instanceInfo.getSecurePort())
.metadata(extractMetadata(instanceInfo))
.healthy(InstanceStatus.UP.equals(instanceInfo.getStatus()))
.build();
}
}
智能路由与负载均衡
@Component
public class SmartRouter {
private final ServiceDiscovery discovery;
private final LoadBalancer loadBalancer;
private final CircuitBreaker circuitBreaker;
private final Map<String, List<ServiceInstance>> instanceCache = new ConcurrentHashMap<>();
@Autowired
public SmartRouter(ServiceDiscovery discovery) {
this.discovery = discovery;
this.loadBalancer = new WeightedRoundRobinLoadBalancer();
this.circuitBreaker = new Resilience4jCircuitBreaker();
// 定期刷新缓存
scheduleCacheRefresh();
}
public ServiceInstance route(String serviceName) {
// 获取可用实例
List<ServiceInstance> instances = getAvailableInstances(serviceName);
if (instances.isEmpty()) {
throw new NoAvailableInstanceException("No available instances for: " + serviceName);
}
// 负载均衡选择
ServiceInstance instance = loadBalancer.select(instances);
// 断路器检测
if (circuitBreaker.isOpen(instance)) {
// 排除熔断的实例
instances = instances.stream()
.filter(i -> !circuitBreaker.isOpen(i))
.collect(Collectors.toList());
if (instances.isEmpty()) {
throw new CircuitBreakerOpenException("All instances are circuit-breaked");
}
instance = loadBalancer.select(instances);
}
return instance;
}
private List<ServiceInstance> getAvailableInstances(String serviceName) {
// 先查询缓存
List<ServiceInstance> cached = instanceCache.get(serviceName);
if (cached != null && !cached.isEmpty()) {
// 过滤健康的实例
return cached.stream()
.filter(ServiceInstance::isHealthy)
.collect(Collectors.toList());
}
// 从服务发现获取
List<ServiceInstance> instances = discovery.discover(serviceName);
// 更新缓存
instanceCache.put(serviceName, instances);
return instances;
}
@Scheduled(fixedDelay = 30000)
private void scheduleCacheRefresh() {
instanceCache.forEach((serviceName, instances) -> {
try {
List<ServiceInstance> fresh = discovery.discover(serviceName);
instanceCache.put(serviceName, fresh);
} catch (Exception e) {
log.error("Failed to refresh cache for: {}", serviceName, e);
}
});
}
}
// 加权轮询负载均衡
@Component
public class WeightedRoundRobinLoadBalancer implements LoadBalancer {
private final Map<String, AtomicInteger> positionMap = new ConcurrentHashMap<>();
@Override
public ServiceInstance select(List<ServiceInstance> instances) {
if (instances.isEmpty()) {
return null;
}
// 计算总权重
int totalWeight = instances.stream()
.mapToInt(ServiceInstance::getWeight)
.sum();
// 获取当前位置
String key = instances.get(0).getServiceName();
AtomicInteger position = positionMap.computeIfAbsent(key,
k -> new AtomicInteger(0));
int current = position.getAndIncrement() % totalWeight;
// 根据权重选择
int weightSum = 0;
for (ServiceInstance instance : instances) {
weightSum += instance.getWeight();
if (current < weightSum) {
return instance;
}
}
return instances.get(0);
}
}
健康检查与故障转移
@Component
public class HealthChecker {
private final ServiceDiscovery discovery;
private final MetricsCollector metrics;
private final Map<String, HealthStatus> healthStatusMap = new ConcurrentHashMap<>();
@Autowired
public HealthChecker(ServiceDiscovery discovery, MetricsCollector metrics) {
this.discovery = discovery;
this.metrics = metrics;
}
@Scheduled(fixedDelay = 5000)
public void checkHealth() {
// 获取所有已注册的服务
List<ServiceInstance> allServices = discovery.getAllServices();
allServices.parallelStream().forEach(this::checkServiceHealth);
}
private void checkServiceHealth(ServiceInstance instance) {
String instanceId = instance.getId();
try {
// 发送健康检查请求
String healthUrl = String.format("http://%s:%d/health",
instance.getHost(), instance.getPort());
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(healthUrl))
.timeout(Duration.ofSeconds(5))
.GET()
.build();
HttpResponse<String> response = client.send(request,
HttpResponse.BodyHandlers.ofString());
if (response.statusCode() == 200) {
// 解析健康状态
HealthResponse healthResponse = parseHealthResponse(response.body());
HealthStatus status = HealthStatus.builder()
.instanceId(instanceId)
.healthy(true)
.lastCheckTime(Instant.now())
.responseTime(calculateResponseTime(response))
.build();
healthStatusMap.put(instanceId, status);
metrics.recordHealthCheck(instanceId, true);
} else {
markUnhealthy(instanceId);
}
} catch (Exception e) {
markUnhealthy(instanceId);
log.warn("Health check failed for {}: {}", instanceId, e.getMessage());
}
}
private void markUnhealthy(String instanceId) {
HealthStatus status = HealthStatus.builder()
.instanceId(instanceId)
.healthy(false)
.lastCheckTime(Instant.now())
.failureCount(getFailureCount(instanceId) + 1)
.build();
healthStatusMap.put(instanceId, status);
metrics.recordHealthCheck(instanceId, false);
// 如果连续失败超过阈值，通知服务发现
if (status.getFailureCount() >= 3) {
discovery.markUnhealthy(instanceId);
}
}
}
API网关集成
@RestController
@RequestMapping("/api/gateway")
public class ServiceGatewayController {
private final SmartRouter router;
private final ServiceDiscovery discovery;
private final RateLimiter rateLimiter;
@Autowired
public ServiceGatewayController(SmartRouter router,
ServiceDiscovery discovery,
RateLimiter rateLimiter) {
this.router = router;
this.discovery = discovery;
this.rateLimiter = rateLimiter;
}
@RequestMapping("/{serviceName}/**")
public ResponseEntity<?> proxyRequest(
@PathVariable String serviceName,
HttpServletRequest request) {
// 限流检查
if (!rateLimiter.tryAcquire(serviceName)) {
return ResponseEntity.status(429).body("Too many requests");
}
try {
// 智能路由
ServiceInstance instance = router.route(serviceName);
// 构建转发请求
String targetUrl = buildTargetUrl(instance, request);
// 发送请求并返回响应
ResponseEntity<?> response = sendRequest(targetUrl, request);
// 记录调用信息
recordInvocation(instance, request, response);
return response;
} catch (NoAvailableInstanceException e) {
return ResponseEntity.status(503).body("Service unavailable");
} catch (Exception e) {
log.error("Gateway error", e);
return ResponseEntity.status(500).body("Internal error");
}
}
@GetMapping("/services")
public ResponseEntity<List<ServiceInfo>> getAvailableServices() {
List<ServiceInfo> services = discovery.getAllServices().stream()
.map(this::convertToServiceInfo)
.collect(Collectors.toList());
return ResponseEntity.ok(services);
}
@GetMapping("/services/{serviceName}/instances")
public ResponseEntity<List<ServiceInstance>> getServiceInstances(
@PathVariable String serviceName) {
List<ServiceInstance> instances = discovery.discover(serviceName);
return ResponseEntity.ok(instances);
}
private ServiceInfo convertToServiceInfo(ServiceInstance instance) {
return ServiceInfo.builder()
.name(instance.getServiceName())
.instances(1) // 需要聚合
.healthy(instance.isHealthy())
.build();
}
}
配置与最佳实践
配置文件
# application.yml
service:
discovery:
type: consul # consul, zookeeper, eureka, kubernetes
consul:
host: localhost
port: 8500
health-check:
interval: 10s
timeout: 3s
deregister-after: 30s
cache:
enabled: true
refresh-interval: 30s
expiration: 5m
retry:
max-attempts: 3
backoff: 1s
circuit-breaker:
enabled: true
failure-threshold: 5
reset-timeout: 30s
使用示例
@Service
public class BusinessService {
@Autowired
private ServiceDiscovery discovery;
@Autowired
private ServiceRegistry registry;
@PostConstruct
public void init() {
// 注册当前服务实例
ServiceInstance instance = ServiceInstance.builder()
.id(UUID.randomUUID().toString())
.serviceName("user-service")
.host(InetAddress.getLocalHost().getHostAddress())
.port(8080)
.tags(Arrays.asList("v1", "production"))
.weight(100)
.metadata(Map.of("version", "1.0.0", "region", "us-east"))
.build();
registry.register(instance);
// 监听其他服务变化
discovery.watch("order-service", instances -> {
log.info("Order service instances updated: {}", instances.size());
updateOrderServiceClients(instances);
});
}
@PreDestroy
public void cleanup() {
registry.deregister();
}
public Order getOrder(String orderId) {
// 发现订单服务实例
List<ServiceInstance> instances = discovery.discover("order-service");
if (instances.isEmpty()) {
throw new ServiceUnavailableException("Order service not available");
}
// 负载均衡选择实例
ServiceInstance instance = selectInstance(instances);
// 调用服务
return callOrderService(instance, orderId);
}
}
Java分布式数据发现API的服务主要包括：
- 核心功能：服务注册、发现、健康检查
- 主流实现：ZooKeeper、Consul、Eureka、Kubernetes
- 高级特性：智能路由、负载均衡、熔断降级
- 运维集成：健康检查、自动故障恢复
- 性能优化：缓存机制、异步更新、批量操作
选择合适的服务发现方案需要考虑：
- 系统规模和复杂度
- 一致性需求（CP vs AP）
- 运维成本和团队经验
- 与现有技术栈的兼容性
