服务注册、发现与心跳机制详解

老马啸西风2025/9/1大约 10 分钟

在微服务架构中，服务注册与发现是实现服务间通信的基础。当服务实例启动时，它们需要向注册中心注册自己的信息；当其他服务需要调用它们时，需要从注册中心获取可用的服务实例列表。同时，为了确保注册信息的准确性，还需要通过心跳机制来检测服务实例的健康状态。本章将深入探讨服务注册、发现和心跳机制的实现原理。

服务注册机制

服务注册是微服务架构中的第一步，当服务实例启动时，它需要向注册中心注册自己的元数据信息，包括IP地址、端口号、服务名称、协议类型等。

注册信息的组成

一个完整的服务注册信息通常包含以下内容：

// 服务实例信息
public class ServiceInstance {
    private String serviceId;        // 服务ID
    private String host;             // 主机地址
    private int port;                // 端口号
    private String scheme;           // 协议类型 (HTTP/HTTPS)
    private Map<String, String> metadata; // 元数据信息
    private long registerTime;       // 注册时间
    private boolean enabled;         // 是否启用
    private boolean healthy;         // 健康状态
    
    // 构造函数、getter和setter方法
    public ServiceInstance(String serviceId, String host, int port) {
        this.serviceId = serviceId;
        this.host = host;
        this.port = port;
        this.scheme = "http";
        this.metadata = new HashMap<>();
        this.registerTime = System.currentTimeMillis();
        this.enabled = true;
        this.healthy = true;
    }
    
    // ... 其他方法
}

注册流程实现

服务注册的实现通常包括以下几个步骤：

// 服务注册接口
public interface ServiceRegistry {
    /**
     * 注册服务实例
     * @param serviceInstance 服务实例信息
     */
    void register(ServiceInstance serviceInstance);
    
    /**
     * 注销服务实例
     * @param serviceInstance 服务实例信息
     */
    void deregister(ServiceInstance serviceInstance);
    
    /**
     * 获取所有服务名称
     * @return 服务名称列表
     */
    List<String> getServices();
    
    /**
     * 根据服务名称获取实例列表
     * @param serviceId 服务名称
     * @return 服务实例列表
     */
    List<ServiceInstance> getInstances(String serviceId);
}

// 服务注册实现类
public class DefaultServiceRegistry implements ServiceRegistry {
    // 使用ConcurrentHashMap存储服务实例信息
    private final Map<String, List<ServiceInstance>> registry = new ConcurrentHashMap<>();
    private final Map<String, Long> lastHeartbeatTime = new ConcurrentHashMap<>();
    
    @Override
    public void register(ServiceInstance serviceInstance) {
        String serviceId = serviceInstance.getServiceId();
        
        // 将服务实例添加到注册表中
        registry.computeIfAbsent(serviceId, k -> new CopyOnWriteArrayList<>())
                .add(serviceInstance);
        
        // 记录心跳时间
        String instanceKey = getInstanceKey(serviceInstance);
        lastHeartbeatTime.put(instanceKey, System.currentTimeMillis());
        
        System.out.println("服务注册成功: " + serviceInstance.getServiceId() + 
                          " -> " + serviceInstance.getHost() + ":" + serviceInstance.getPort());
    }
    
    @Override
    public void deregister(ServiceInstance serviceInstance) {
        String serviceId = serviceInstance.getServiceId();
        String instanceKey = getInstanceKey(serviceInstance);
        
        // 从注册表中移除服务实例
        List<ServiceInstance> instances = registry.get(serviceId);
        if (instances != null) {
            instances.removeIf(instance -> 
                instance.getHost().equals(serviceInstance.getHost()) && 
                instance.getPort() == serviceInstance.getPort());
        }
        
        // 移除心跳记录
        lastHeartbeatTime.remove(instanceKey);
        
        System.out.println("服务注销成功: " + serviceInstance.getServiceId());
    }
    
    @Override
    public List<String> getServices() {
        return new ArrayList<>(registry.keySet());
    }
    
    @Override
    public List<ServiceInstance> getInstances(String serviceId) {
        List<ServiceInstance> instances = registry.get(serviceId);
        return instances != null ? new ArrayList<>(instances) : Collections.emptyList();
    }
    
    // 生成实例唯一标识
    private String getInstanceKey(ServiceInstance instance) {
        return instance.getServiceId() + ":" + instance.getHost() + ":" + instance.getPort();
    }
}

服务发现机制

服务发现是客户端从注册中心获取可用服务实例列表的过程。服务消费者通过服务发现机制获取服务提供者的地址信息，然后进行远程调用。

服务发现实现

// 服务发现接口
public interface ServiceDiscovery {
    /**
     * 发现服务实例
     * @param serviceId 服务ID
     * @return 服务实例列表
     */
    List<ServiceInstance> discover(String serviceId);
    
    /**
     * 获取所有服务名称
     * @return 服务名称列表
     */
    List<String> getAllServices();
}

// 服务发现实现类
public class DefaultServiceDiscovery implements ServiceDiscovery {
    private final ServiceRegistry serviceRegistry;
    
    public DefaultServiceDiscovery(ServiceRegistry serviceRegistry) {
        this.serviceRegistry = serviceRegistry;
    }
    
    @Override
    public List<ServiceInstance> discover(String serviceId) {
        // 从注册中心获取服务实例列表
        List<ServiceInstance> instances = serviceRegistry.getInstances(serviceId);
        
        // 过滤掉不健康或未启用的实例
        return instances.stream()
                .filter(instance -> instance.isEnabled() && instance.isHealthy())
                .collect(Collectors.toList());
    }
    
    @Override
    public List<String> getAllServices() {
        return serviceRegistry.getServices();
    }
}

负载均衡服务发现

在实际应用中，通常会结合负载均衡策略来选择合适的服务实例：

// 负载均衡策略接口
public interface LoadBalancer {
    /**
     * 从服务实例列表中选择一个实例
     * @param instances 服务实例列表
     * @return 选中的服务实例
     */
    ServiceInstance choose(List<ServiceInstance> instances);
}

// 随机负载均衡策略
public class RandomLoadBalancer implements LoadBalancer {
    private final Random random = new Random();
    
    @Override
    public ServiceInstance choose(List<ServiceInstance> instances) {
        if (instances == null || instances.isEmpty()) {
            return null;
        }
        
        int index = random.nextInt(instances.size());
        return instances.get(index);
    }
}

// 轮询负载均衡策略
public class RoundRobinLoadBalancer implements LoadBalancer {
    private final AtomicInteger counter = new AtomicInteger(0);
    
    @Override
    public ServiceInstance choose(List<ServiceInstance> instances) {
        if (instances == null || instances.isEmpty()) {
            return null;
        }
        
        int index = counter.getAndIncrement() % instances.size();
        // 避免整数溢出
        if (counter.get() < 0) {
            counter.set(0);
        }
        
        return instances.get(index);
    }
}

// 带负载均衡的服务发现
public class LoadBalancedServiceDiscovery {
    private final ServiceDiscovery serviceDiscovery;
    private final LoadBalancer loadBalancer;
    
    public LoadBalancedServiceDiscovery(ServiceDiscovery serviceDiscovery, LoadBalancer loadBalancer) {
        this.serviceDiscovery = serviceDiscovery;
        this.loadBalancer = loadBalancer;
    }
    
    public ServiceInstance chooseInstance(String serviceId) {
        List<ServiceInstance> instances = serviceDiscovery.discover(serviceId);
        return loadBalancer.choose(instances);
    }
}

心跳机制

心跳机制是确保注册信息准确性的关键，通过定期发送心跳包来检测服务实例的健康状态。如果服务实例在一定时间内没有发送心跳，注册中心会将其标记为不健康或从注册表中移除。

心跳检测实现

// 心跳检测器
public class HeartbeatDetector {
    private final ServiceRegistry serviceRegistry;
    private final Map<String, Long> lastHeartbeatTime;
    private final long heartbeatTimeout; // 心跳超时时间（毫秒）
    private final ScheduledExecutorService scheduler;
    
    public HeartbeatDetector(ServiceRegistry serviceRegistry, long heartbeatTimeout) {
        this.serviceRegistry = serviceRegistry;
        this.lastHeartbeatTime = new ConcurrentHashMap<>();
        this.heartbeatTimeout = heartbeatTimeout;
        this.scheduler = Executors.newScheduledThreadPool(1);
    }
    
    // 启动心跳检测
    public void start() {
        scheduler.scheduleAtFixedRate(this::checkHeartbeat, 0, 
                                    heartbeatTimeout / 2, TimeUnit.MILLISECONDS);
    }
    
    // 停止心跳检测
    public void stop() {
        scheduler.shutdown();
    }
    
    // 检查心跳状态
    private void checkHeartbeat() {
        long currentTime = System.currentTimeMillis();
        List<String> services = serviceRegistry.getServices();
        
        for (String serviceId : services) {
            List<ServiceInstance> instances = serviceRegistry.getInstances(serviceId);
            
            for (ServiceInstance instance : instances) {
                String instanceKey = getInstanceKey(instance);
                Long lastHeartbeat = lastHeartbeatTime.get(instanceKey);
                
                // 如果超过心跳超时时间未收到心跳，则标记为不健康
                if (lastHeartbeat != null && 
                    currentTime - lastHeartbeat > heartbeatTimeout) {
                    instance.setHealthy(false);
                    System.out.println("服务实例不健康: " + instanceKey);
                }
            }
        }
    }
    
    // 处理心跳包
    public void handleHeartbeat(ServiceInstance serviceInstance) {
        String instanceKey = getInstanceKey(serviceInstance);
        lastHeartbeatTime.put(instanceKey, System.currentTimeMillis());
        
        // 更新服务实例的健康状态
        List<ServiceInstance> instances = serviceRegistry.getInstances(
            serviceInstance.getServiceId());
        
        for (ServiceInstance instance : instances) {
            if (instance.getHost().equals(serviceInstance.getHost()) && 
                instance.getPort() == serviceInstance.getPort()) {
                instance.setHealthy(true);
                break;
            }
        }
    }
    
    private String getInstanceKey(ServiceInstance instance) {
        return instance.getServiceId() + ":" + instance.getHost() + ":" + instance.getPort();
    }
}

客户端心跳发送

服务实例需要定期向注册中心发送心跳包：

// 客户端心跳发送器
public class HeartbeatSender {
    private final ServiceInstance serviceInstance;
    private final String registryUrl;
    private final long heartbeatInterval;
    private final ScheduledExecutorService scheduler;
    
    public HeartbeatSender(ServiceInstance serviceInstance, String registryUrl, 
                          long heartbeatInterval) {
        this.serviceInstance = serviceInstance;
        this.registryUrl = registryUrl;
        this.heartbeatInterval = heartbeatInterval;
        this.scheduler = Executors.newScheduledThreadPool(1);
    }
    
    // 启动心跳发送
    public void start() {
        scheduler.scheduleAtFixedRate(this::sendHeartbeat, 0, 
                                    heartbeatInterval, TimeUnit.MILLISECONDS);
    }
    
    // 停止心跳发送
    public void stop() {
        scheduler.shutdown();
    }
    
    // 发送心跳包
    private void sendHeartbeat() {
        try {
            // 构造心跳请求
            String url = registryUrl + "/heartbeat";
            OkHttpClient client = new OkHttpClient();
            
            String json = new ObjectMapper().writeValueAsString(serviceInstance);
            RequestBody body = RequestBody.create(
                MediaType.parse("application/json"), json);
            
            Request request = new Request.Builder()
                .url(url)
                .post(body)
                .build();
            
            Response response = client.newCall(request).execute();
            if (response.isSuccessful()) {
                System.out.println("心跳发送成功: " + serviceInstance.getServiceId());
            } else {
                System.out.println("心跳发送失败: " + response.code());
            }
        } catch (Exception e) {
            System.err.println("发送心跳异常: " + e.getMessage());
        }
    }
}

临时节点与永久节点

在服务注册中心中，通常会区分临时节点和永久节点：

临时节点特性

临时节点具有以下特点：

当客户端与注册中心的连接断开时，临时节点会自动被删除
适用于服务实例，因为服务实例可能会意外宕机
通过心跳机制来维持节点的存在

永久节点特性

永久节点具有以下特点：

即使客户端断开连接，节点仍然存在
适用于配置信息、元数据等不会频繁变化的内容
需要显式删除才会被移除

// 节点类型枚举
public enum NodeType {
    TEMPORARY,  // 临时节点
    PERMANENT   // 永久节点
}

// 带节点类型的服务实例
public class ServiceInstanceWithNodeType extends ServiceInstance {
    private NodeType nodeType;  // 节点类型
    
    public ServiceInstanceWithNodeType(String serviceId, String host, int port, NodeType nodeType) {
        super(serviceId, host, port);
        this.nodeType = nodeType;
    }
    
    public NodeType getNodeType() {
        return nodeType;
    }
    
    public void setNodeType(NodeType nodeType) {
        this.nodeType = nodeType;
    }
}

// 支持节点类型的服务注册实现
public class NodeTypeAwareServiceRegistry extends DefaultServiceRegistry {
    private final Map<String, NodeType> nodeTypes = new ConcurrentHashMap<>();
    
    @Override
    public void register(ServiceInstance serviceInstance) {
        super.register(serviceInstance);
        
        // 如果是带节点类型的服务实例，记录节点类型
        if (serviceInstance instanceof ServiceInstanceWithNodeType) {
            ServiceInstanceWithNodeType instanceWithNodeType = 
                (ServiceInstanceWithNodeType) serviceInstance;
            String instanceKey = getInstanceKey(serviceInstance);
            nodeTypes.put(instanceKey, instanceWithNodeType.getNodeType());
        }
    }
    
    // 清理过期的临时节点
    public void cleanupExpiredTemporaryNodes() {
        long currentTime = System.currentTimeMillis();
        List<String> services = getServices();
        
        for (String serviceId : services) {
            List<ServiceInstance> instances = getInstances(serviceId);
            
            for (ServiceInstance instance : instances) {
                String instanceKey = getInstanceKey(instance);
                NodeType nodeType = nodeTypes.get(instanceKey);
                
                // 如果是临时节点且心跳超时，则删除
                if (nodeType == NodeType.TEMPORARY) {
                    Long lastHeartbeat = getLastHeartbeatTime(instanceKey);
                    if (lastHeartbeat != null && 
                        currentTime - lastHeartbeat > getHeartbeatTimeout()) {
                        deregister(instance);
                        nodeTypes.remove(instanceKey);
                    }
                }
            }
        }
    }
    
    // 获取心跳超时时间
    private long getHeartbeatTimeout() {
        // 根据实际配置返回心跳超时时间
        return 30000; // 30秒
    }
    
    // 获取最后心跳时间
    private Long getLastHeartbeatTime(String instanceKey) {
        // 从父类获取最后心跳时间的实现
        // 这里简化处理，实际应该访问父类的lastHeartbeatTime字段
        return System.currentTimeMillis() - 10000; // 模拟返回
    }
}

客户端缓存与订阅模型

为了提高性能和减少对注册中心的压力，客户端通常会实现本地缓存和订阅机制。

客户端缓存实现

// 客户端缓存管理器
public class ClientCacheManager {
    private final Map<String, List<ServiceInstance>> cache = new ConcurrentHashMap<>();
    private final Map<String, Long> cacheUpdateTime = new ConcurrentHashMap<>();
    private final long cacheExpireTime; // 缓存过期时间（毫秒）
    
    public ClientCacheManager(long cacheExpireTime) {
        this.cacheExpireTime = cacheExpireTime;
    }
    
    // 获取缓存的服务实例列表
    public List<ServiceInstance> getCachedInstances(String serviceId) {
        Long updateTime = cacheUpdateTime.get(serviceId);
        if (updateTime != null && 
            System.currentTimeMillis() - updateTime < cacheExpireTime) {
            return cache.get(serviceId);
        }
        return null; // 缓存过期或不存在
    }
    
    // 更新缓存
    public void updateCache(String serviceId, List<ServiceInstance> instances) {
        cache.put(serviceId, new ArrayList<>(instances));
        cacheUpdateTime.put(serviceId, System.currentTimeMillis());
    }
    
    // 清除缓存
    public void clearCache(String serviceId) {
        cache.remove(serviceId);
        cacheUpdateTime.remove(serviceId);
    }
}

订阅模型实现

// 服务实例变更监听器
public interface ServiceInstancesChangeListener {
    void onChange(String serviceId, List<ServiceInstance> instances);
}

// 订阅管理器
public class SubscriptionManager {
    private final Map<String, Set<ServiceInstancesChangeListener>> listeners = new ConcurrentHashMap<>();
    
    // 订阅服务实例变更
    public void subscribe(String serviceId, ServiceInstancesChangeListener listener) {
        listeners.computeIfAbsent(serviceId, k -> ConcurrentHashMap.newKeySet())
                 .add(listener);
    }
    
    // 取消订阅
    public void unsubscribe(String serviceId, ServiceInstancesChangeListener listener) {
        Set<ServiceInstancesChangeListener> serviceListeners = listeners.get(serviceId);
        if (serviceListeners != null) {
            serviceListeners.remove(listener);
        }
    }
    
    // 通知服务实例变更
    public void notifyChange(String serviceId, List<ServiceInstance> instances) {
        Set<ServiceInstancesChangeListener> serviceListeners = listeners.get(serviceId);
        if (serviceListeners != null) {
            for (ServiceInstancesChangeListener listener : serviceListeners) {
                try {
                    listener.onChange(serviceId, new ArrayList<>(instances));
                } catch (Exception e) {
                    System.err.println("通知监听器异常: " + e.getMessage());
                }
            }
        }
    }
}

// 客户端服务发现
public class ClientServiceDiscovery {
    private final String registryUrl;
    private final ClientCacheManager cacheManager;
    private final SubscriptionManager subscriptionManager;
    private final ScheduledExecutorService scheduler;
    
    public ClientServiceDiscovery(String registryUrl) {
        this.registryUrl = registryUrl;
        this.cacheManager = new ClientCacheManager(30000); // 30秒缓存过期时间
        this.subscriptionManager = new SubscriptionManager();
        this.scheduler = Executors.newScheduledThreadPool(2);
        
        // 启动定时刷新任务
        startPeriodicRefresh();
    }
    
    // 发现服务实例
    public List<ServiceInstance> discover(String serviceId) {
        // 首先尝试从缓存获取
        List<ServiceInstance> cachedInstances = cacheManager.getCachedInstances(serviceId);
        if (cachedInstances != null) {
            return cachedInstances;
        }
        
        // 缓存未命中，从注册中心获取
        try {
            List<ServiceInstance> instances = fetchFromRegistry(serviceId);
            cacheManager.updateCache(serviceId, instances);
            return instances;
        } catch (Exception e) {
            System.err.println("从注册中心获取服务实例失败: " + e.getMessage());
            return Collections.emptyList();
        }
    }
    
    // 订阅服务实例变更
    public void subscribe(String serviceId, ServiceInstancesChangeListener listener) {
        subscriptionManager.subscribe(serviceId, listener);
    }
    
    // 取消订阅
    public void unsubscribe(String serviceId, ServiceInstancesChangeListener listener) {
        subscriptionManager.unsubscribe(serviceId, listener);
    }
    
    // 从注册中心获取服务实例
    private List<ServiceInstance> fetchFromRegistry(String serviceId) throws IOException {
        String url = registryUrl + "/services/" + serviceId;
        OkHttpClient client = new OkHttpClient();
        
        Request request = new Request.Builder().url(url).build();
        Response response = client.newCall(request).execute();
        
        if (response.isSuccessful()) {
            String json = response.body().string();
            ObjectMapper mapper = new ObjectMapper();
            return mapper.readValue(json, new TypeReference<List<ServiceInstance>>() {});
        } else {
            throw new IOException("获取服务实例失败: " + response.code());
        }
    }
    
    // 启动定时刷新任务
    private void startPeriodicRefresh() {
        scheduler.scheduleAtFixedRate(this::refreshAllCache, 0, 10, TimeUnit.SECONDS);
    }
    
    // 刷新所有缓存
    private void refreshAllCache() {
        for (String serviceId : cacheManager.getAllCachedServices()) {
            try {
                List<ServiceInstance> instances = fetchFromRegistry(serviceId);
                cacheManager.updateCache(serviceId, instances);
                
                // 通知订阅者
                subscriptionManager.notifyChange(serviceId, instances);
            } catch (Exception e) {
                System.err.println("刷新服务缓存失败: " + serviceId + ", " + e.getMessage());
            }
        }
    }
}

总结

服务注册、发现和心跳机制是微服务架构的核心组件，它们共同保证了服务间通信的可靠性和高效性：

服务注册：服务实例启动时向注册中心注册自己的信息，包括IP、端口、服务名称等
服务发现：服务消费者通过注册中心获取可用的服务实例列表，通常结合负载均衡策略
心跳机制：通过定期发送心跳包来检测服务实例的健康状态，及时清理不健康的实例
节点类型：区分临时节点和永久节点，临时节点在连接断开时自动删除
客户端缓存：客户端实现本地缓存以减少对注册中心的请求压力
订阅模型：通过订阅机制实现服务实例变更的实时通知

通过合理设计和实现这些机制，可以构建一个高可用、高性能的服务注册与发现系统，为微服务架构提供坚实的基础。