本文目录导读：
我来详细介绍如何在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,或者同时使用两者作为备份方案。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
