本文目录导读：
我来详细介绍如何在PHP项目中集成Consul和Etcd服务发现与配置管理。
基本概念
Consul
- 服务发现
- 健康检查
- KV存储
- 多数据中心
Etcd
- 分布式KV存储
- 服务发现
- 配置管理
- 分布式锁
Consul集成
1 安装Consul客户端库
composer require sensiolabs/consul-php-sdk # 或 composer require dcarbone/phpconsul-api
2 基础连接配置
<?php
namespace App\Service;
use SensioLabs\Consul\ConsulResponse;
use SensioLabs\Consul\ServiceFactory;
class ConsulService
{
private $consul;
public function __construct()
{
$serviceFactory = new ServiceFactory([
'base_uri' => 'http://127.0.0.1:8500'
]);
$this->consul = $serviceFactory;
}
// 服务注册
public function registerService(array $config)
{
$agent = $this->consul->get('agent');
$serviceConfig = [
'ID' => $config['id'] ?? uniqid(),
'Name' => $config['name'],
'Tags' => $config['tags'] ?? [],
'Port' => $config['port'],
'Address' => $config['address'] ?? '127.0.0.1',
'Check' => [
'HTTP' => sprintf('http://%s:%d/health',
$config['address'] ?? '127.0.0.1',
$config['port']
),
'Interval' => $config['check_interval'] ?? '10s',
'Timeout' => $config['check_timeout'] ?? '5s',
'DeregisterCriticalServiceAfter' => '30s'
]
];
return $agent->registerService($serviceConfig);
}
// 发现服务
public function discoverService(string $serviceName): array
{
$catalog = $this->consul->get('catalog');
$services = $catalog->service($serviceName)->json();
$healthyServices = [];
foreach ($services as $service) {
if ($this->checkServiceHealth($service['ServiceID'])) {
$healthyServices[] = [
'id' => $service['ServiceID'],
'address' => $service['ServiceAddress'],
'port' => $service['ServicePort'],
'tags' => $service['ServiceTags'] ?? []
];
}
}
return $healthyServices;
}
// 检查服务健康状态
private function checkServiceHealth(string $serviceId): bool
{
$health = $this->consul->get('health');
$checks = $health->checks($serviceId)->json();
foreach ($checks as $check) {
if ($check['Status'] !== 'passing') {
return false;
}
}
return true;
}
// KV存储操作
public function getConfig(string $key): ?string
{
$kv = $this->consul->get('kv');
$response = $kv->get($key);
if ($response->getStatusCode() === 200) {
$data = $response->json();
return base64_decode($data[0]['Value']);
}
return null;
}
public function setConfig(string $key, string $value): bool
{
$kv = $this->consul->get('kv');
$response = $kv->put($key, $value);
return $response->getStatusCode() === 200;
}
// 监听配置变化
public function watchConfig(string $key, callable $callback)
{
$lastIndex = 0;
while (true) {
$kv = $this->consul->get('kv');
$response = $kv->get($key, [
'index' => $lastIndex,
'wait' => '5s'
]);
$headers = $response->getHeaders();
$newIndex = (int)$headers['X-Consul-Index'][0];
if ($newIndex > $lastIndex) {
$data = $response->json();
$value = base64_decode($data[0]['Value']);
$callback($value);
$lastIndex = $newIndex;
}
}
}
}
3 健康检查端点
<?php
namespace App\Controller;
use Symfony\Component\HttpFoundation\JsonResponse;
use Symfony\Component\Routing\Annotation\Route;
class HealthController
{
#[Route('/health', name: 'health_check')]
public function check(): JsonResponse
{
$healthStatus = [
'status' => 'UP',
'timestamp' => time(),
'checks' => [
'database' => $this->checkDatabase(),
'cache' => $this->checkCache(),
'disk' => $this->checkDiskSpace()
]
];
$isHealthy = !in_array(false, array_column($healthStatus['checks'], 'healthy'));
return new JsonResponse(
$healthStatus,
$isHealthy ? 200 : 503
);
}
private function checkDatabase(): array
{
try {
// 数据库检查逻辑
return [
'healthy' => true,
'message' => 'Database connection OK'
];
} catch (\Exception $e) {
return [
'healthy' => false,
'message' => $e->getMessage()
];
}
}
private function checkCache(): array
{
try {
// 缓存检查逻辑
return [
'healthy' => true,
'message' => 'Cache connection OK'
];
} catch (\Exception $e) {
return [
'healthy' => false,
'message' => $e->getMessage()
];
}
}
private function checkDiskSpace(): array
{
$freeSpace = disk_free_space('/');
$totalSpace = disk_total_space('/');
$usedPercent = (1 - $freeSpace / $totalSpace) * 100;
return [
'healthy' => $usedPercent < 90,
'message' => sprintf('Disk usage: %.2f%%', $usedPercent),
'freeSpace' => $freeSpace,
'totalSpace' => $totalSpace
];
}
}
Etcd集成
1 安装Etcd客户端
composer require linkorb/etcd-php # 或 composer require etcd-php/etcd
2 Etcd服务类
<?php
namespace App\Service;
use Etcd\Client;
use Etcd\Response;
class EtcdService
{
private $client;
private $config;
public function __construct(array $config = [])
{
$this->config = array_merge([
'host' => '127.0.0.1',
'port' => 2379,
'version' => 'v3',
'timeout' => 5
], $config);
$this->client = new Client(
$this->config['host'],
$this->config['port'],
$this->config['version']
);
}
// 服务注册
public function registerService(array $serviceConfig)
{
$key = sprintf('/services/%s/%s',
$serviceConfig['name'],
$serviceConfig['id'] ?? uniqid()
);
$value = json_encode([
'name' => $serviceConfig['name'],
'address' => $serviceConfig['address'],
'port' => $serviceConfig['port'],
'tags' => $serviceConfig['tags'] ?? [],
'registered_at' => time(),
'ttl' => $serviceConfig['ttl'] ?? 30
]);
// 注册服务（带TTL）
$this->client->put($key, $value, [
'lease' => $this->createLease($serviceConfig['ttl'] ?? 30)
]);
return $key;
}
// 创建租约
private function createLease(int $ttl): string
{
$response = $this->client->grant($ttl);
return $response['ID'];
}
// 续约
public function keepAlive(string $leaseId): bool
{
try {
$this->client->keepAlive($leaseId);
return true;
} catch (\Exception $e) {
return false;
}
}
// 发现服务
public function discoverService(string $serviceName): array
{
$key = sprintf('/services/%s', $serviceName);
$response = $this->client->get($key, ['prefix' => true]);
$services = [];
foreach ($response['kvs'] ?? [] as $kv) {
$service = json_decode($kv['value'], true);
if ($this->isServiceHealthy($service)) {
$services[] = $service;
}
}
return $services;
}
// 检查服务健康状态
private function isServiceHealthy(array $service): bool
{
if (!isset($service['address'], $service['port'])) {
return false;
}
$url = sprintf('http://%s:%d/health',
$service['address'],
$service['port']
);
$ch = curl_init($url);
curl_setopt_array($ch, [
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => 5,
CURLOPT_CONNECTTIMEOUT => 3
]);
curl_exec($ch);
$httpCode = curl_getinfo($ch, CURLINFO_HTTP_CODE);
curl_close($ch);
return $httpCode === 200;
}
// KV存储操作
public function put(string $key, $value): bool
{
try {
$this->client->put($key, is_array($value) ? json_encode($value) : $value);
return true;
} catch (\Exception $e) {
return false;
}
}
public function get(string $key, bool $prefix = false): ?array
{
try {
$options = $prefix ? ['prefix' => true] : [];
$response = $this->client->get($key, $options);
if (!empty($response['kvs'])) {
$result = [];
foreach ($response['kvs'] as $kv) {
$result[] = [
'key' => $kv['key'],
'value' => json_decode($kv['value'], true) ?? $kv['value'],
'create_revision' => $kv['create_revision'],
'mod_revision' => $kv['mod_revision']
];
}
return $result;
}
return null;
} catch (\Exception $e) {
return null;
}
}
// 删除
public function delete(string $key, bool $prefix = false): bool
{
try {
$options = $prefix ? ['prefix' => true] : [];
$this->client->del($key, $options);
return true;
} catch (\Exception $e) {
return false;
}
}
// 监听变化
public function watch(string $key, callable $callback, bool $prefix = false)
{
$options = [
'prefix' => $prefix,
'progress_notify' => true
];
$watch = $this->client->watch($key, $options);
foreach ($watch as $response) {
foreach ($response['events'] ?? [] as $event) {
$type = $event['type'];
$kv = $event['kv'];
$callback([
'type' => $type,
'key' => $kv['key'],
'value' => json_decode($kv['value'], true) ?? $kv['value'],
'create_revision' => $kv['create_revision'],
'mod_revision' => $kv['mod_revision']
]);
}
}
}
// 分布式锁
public function acquireLock(string $lockKey, int $ttl = 10): bool
{
$lockPath = sprintf('/locks/%s', $lockKey);
$lockValue = uniqid('lock_', true);
try {
$this->client->put($lockPath, $lockValue, [
'lease' => $this->createLease($ttl),
'prev_kv' => false
]);
return true;
} catch (\Exception $e) {
return false;
}
}
public function releaseLock(string $lockKey): bool
{
$lockPath = sprintf('/locks/%s', $lockKey);
return $this->delete($lockPath);
}
}
配置管理类
<?php
namespace App\Config;
class DynamicConfig
{
private $consul;
private $etcd;
private $cache = [];
private $backend;
public function __construct($consul, $etcd, string $backend = 'consul')
{
$this->consul = $consul;
$this->etcd = $etcd;
$this->backend = $backend;
}
// 获取配置
public function get(string $key, $default = null)
{
// 优先从缓存获取
if (isset($this->cache[$key])) {
return $this->cache[$key];
}
$value = $this->backend === 'consul'
? $this->consul->getConfig($key)
: $this->etcd->get($key);
if ($value !== null) {
$this->cache[$key] = $value;
return $value;
}
return $default;
}
// 设置配置
public function set(string $key, $value): bool
{
$result = $this->backend === 'consul'
? $this->consul->setConfig($key, $value)
: $this->etcd->put($key, $value);
if ($result) {
$this->cache[$key] = $value;
}
return $result;
}
// 批量获取配置
public function getMultiple(array $keys): array
{
$result = [];
foreach ($keys as $key) {
$result[$key] = $this->get($key);
}
return $result;
}
// 配置热更新
public function watchConfig(string $key, callable $callback)
{
$this->backend === 'consul'
? $this->consul->watchConfig($key, $callback)
: $this->etcd->watch($key, $callback);
}
}
服务注册器
<?php
namespace App\Service;
class ServiceRegistrar
{
private $consulService;
private $etcdService;
private $config;
private $leaseId;
public function __construct($consulService, $etcdService, array $config)
{
$this->consulService = $consulService;
$this->etcdService = $etcdService;
$this->config = $config;
}
// 注册服务
public function register(): bool
{
$serviceConfig = [
'name' => $this->config['service_name'],
'id' => $this->config['service_id'] ?? null,
'address' => $this->config['service_address'],
'port' => $this->config['service_port'],
'tags' => $this->config['service_tags'] ?? [],
'check_interval' => $this->config['check_interval'] ?? '10s',
'health_endpoint' => $this->config['health_endpoint'] ?? '/health'
];
try {
// 注册到Consul
$this->consulService->registerService($serviceConfig);
// 注册到Etcd
$this->etcdService->registerService($serviceConfig);
return true;
} catch (\Exception $e) {
throw $e;
}
}
// 注销服务
public function deregister(): bool
{
try {
// 注销Consul服务
$this->consulService->get('agent')
->deregisterService($this->config['service_id']);
// 删除Etcd服务键
$key = sprintf('/services/%s/%s',
$this->config['service_name'],
$this->config['service_id']
);
$this->etcdService->delete($key);
return true;
} catch (\Exception $e) {
return false;
}
}
// 心跳维护
public function heartbeat()
{
while (true) {
try {
$this->consulService->get('agent')
->passCheck('service:' . $this->config['service_id']);
if ($this->leaseId) {
$this->etcdService->keepAlive($this->leaseId);
}
sleep($this->config['heartbeat_interval'] ?? 10);
} catch (\Exception $e) {
// 重新注册
$this->register();
}
}
}
}
使用示例
<?php
// 1. 服务注册
$registrar = new ServiceRegistrar($consulService, $etcdService, [
'service_name' => 'user-service',
'service_id' => 'user-service-1',
'service_address' => '192.168.1.100',
'service_port' => 8080,
'service_tags' => ['api', 'v1'],
'check_interval' => '10s',
'health_endpoint' => '/health'
]);
$registrar->register();
// 2. 服务发现
$consulService = new ConsulService();
$services = $consulService->discoverService('user-service');
foreach ($services as $service) {
echo sprintf(
"Service: %s at %s:%d\n",
$service['id'],
$service['address'],
$service['port']
);
}
// 3. 配置管理
$config = new DynamicConfig($consulService, $etcdService, 'consul');
// 读取配置
$dbConfig = $config->getMultiple([
'config/database/host',
'config/database/port',
'config/database/name',
'config/database/user',
'config/database/password'
]);
// 写入配置
$config->set('config/feature/new-feature', json_encode([
'enabled' => true,
'percentage' => 50,
'start_date' => '2024-01-01'
]));
// 4. 配置监听
$config->watchConfig('config/feature/new-feature', function($value) {
$featureConfig = json_decode($value, true);
if ($featureConfig['enabled']) {
// 启用新功能
}
});
// 5. 分布式锁使用
$lockKey = 'user-service-1-migration';
if ($etcdService->acquireLock($lockKey, 30)) {
try {
// 执行需要同步的操作
$this->performMigration();
} finally {
$etcdService->releaseLock($lockKey);
}
}
Docker配置
# docker-compose.yml
version: '3.8'
services:
consul:
image: consul:1.15
ports:
- "8500:8500"
- "8600:8600/udp"
command: agent -server -bootstrap-expect=1 -ui -client=0.0.0.0
volumes:
- consul-data:/consul/data
networks:
- service-mesh
etcd:
image: bitnami/etcd:3.5
ports:
- "2379:2379"
- "2380:2380"
environment:
- ALLOW_NONE_AUTHENTICATION=yes
- ETCD_ADVERTISE_CLIENT_URLS=http://etcd:2379
volumes:
- etcd-data:/bitnami/etcd
networks:
- service-mesh
app:
build: .
ports:
- "8080:8080"
environment:
- CONSUL_HOST=consul
- CONSUL_PORT=8500
- ETCD_HOST=etcd
- ETCD_PORT=2379
depends_on:
- consul
- etcd
networks:
- service-mesh
volumes:
consul-data:
etcd-data:
networks:
service-mesh:
driver: bridge
这个集成方案提供了完整的服务发现、配置管理和健康检查功能，可以根据实际需求选择使用Consul或Etcd,或者同时使用两者作为备份方案。
