Hyperf方案 跨地域多活架构

Hyperf 跨地域多活核心工具链：)┌──────────┬────────────────────────────┐ │ 关注点 │ 方案 │ ├──────────┼────────────────────────────┤ │ 流量路由 │ Nacos 命名空间+就近路由 │ ├──────────┼────────────────────────────┤ │ 数据同步 │ Kafka 跨地域复制 │ ├──────────┼────────────────────────────┤ │ 冲突解决 │ CRDT/Last-Write-Wins │ ├──────────┼────────────────────────────┤ │ 全局 ID │ Snowflake（地域位编码） │ ├──────────┼────────────────────────────┤ │ 配置下发 │ hyperf/config-nacos 多集群 │ └──────────┴────────────────────────────┘---安装 composer require hyperf/service-governance-nacos \ hyperf/kafka \ hyperf/config-nacos \ hyperf/snowflake---1.全局唯一 ID（地域编码）<?php namespace App\IdGen;use Hyperf\Snowflake\IdGeneratorInterface;classGeoSnowflake{// 地域编码：cn-north=1, cn-south=2, us-east=3private static array $regionBits=['cn-north'=>1,'cn-south'=>2,'us-east'=>3,];publicfunction__construct(private IdGeneratorInterface $generator){}publicfunctiongenerate():string{$region=self::$regionBits[env('REGION','cn-north')];$id=$this->generator->generate();// 高位注入地域码，全局唯一且可溯源return$region.str_pad((string)$id,18,'0',STR_PAD_LEFT);}public staticfunctionregionOf(string $id):string{returnarray_flip(self::$regionBits)[(int)$id[0]]??'unknown';}}---2.就近路由（同地域优先）<?php namespace App\LoadBalancer;use Hyperf\LoadBalancer\LoadBalancerInterface;use Hyperf\LoadBalancer\Node;classGeoAwareLoadBalancerimplementsLoadBalancerInterface{private array $nodes=[];private string $localRegion;publicfunction__construct(){$this->localRegion=env('REGION','cn-north');}publicfunctionselect():Node{// 优先同地域节点$local=array_filter($this->nodes,fn($n)=>($n->metadata['region']??'')===$this->localRegion);$pool=!empty($local)?$local:$this->nodes;return$pool[array_rand($pool)];}publicfunctiongetNodes():array{return$this->nodes;}publicfunctionsetNodes(array $nodes):static{$this->nodes=$nodes;return$this;}publicfunctionisAutoRefresh():bool{returntrue;}publicfunctionrefresh(callable $callback,int $tickMs=5000):void{$callback();}}---3.多活写入（本地优先+异步同步）<?php namespace App\Service;use App\IdGen\GeoSnowflake;use Hyperf\Kafka\Producer;classMultiActiveOrderService{publicfunction__construct(private GeoSnowflake $idGen,private Producer $kafka,){}publicfunctioncreate(int $userId,array $items):array{$orderId=$this->idGen->generate();$region=env('REGION','cn-north');// 写本地库（低延迟）$order=Order::create(['id'=>$orderId,'user_id'=>$userId,'total'=>collect($items)->sum('price'),'region'=>$region,'version'=>microtime(true),// LWW 时间戳]);// 异步同步到其他地域$this->kafka->send(topic:'geo.order.sync',value:json_encode(['data'=>$order->toArray(),'source'=>$region,'timestamp'=>microtime(true),]),key:(string)$orderId,// 相同 key 保证有序);return$order->toArray();}}---4.跨地域数据同步消费者<?php namespace App\Consumer;use App\MQ\IdempotentConsumer;use Hyperf\Kafka\Annotation\Consumer;use longlang\phpkafka\Consumer\ConsumeMessage;#[Consumer(topic:'geo.order.sync',groupId:'geo-sync-cn-south',// 每个地域独立 groupIdautoCommit:false,nums:4,)]classGeoSyncConsumerextendsIdempotentConsumer{protectedfunctionhandle(array $payload,ConsumeMessage $message):void{$data=$payload['data'];$source=$payload['source'];// 跳过本地产生的数据if($source===env('REGION'))return;// Last-Write-Wins 冲突解决$existing=Order::find($data['id']);if(!$existing||$existing->version<$data['version']){Order::updateOrCreate(['id'=>$data['id']],array_merge($data,['synced_from'=>$source]));}}}---5.流量切换（地域级熔断）<?php namespace App\Governance;use Hyperf\Redis\Redis;classGeoFailover{private string $region;publicfunction__construct(private Redis $redis){$this->region=env('REGION','cn-north');}// 检测本地地域健康状态publicfunctionisLocalHealthy():bool{$key="geo:health:{$this->region}";$failures=(int)$this->redis->get($key);return$failures<10;}publicfunctionrecordFailure():void{$key="geo:health:{$this->region}";$this->redis->incr($key);$this->redis->expire($key,60);}// 获取可用地域列表（降级路由）publicfunctionavailableRegions():array{$all=['cn-north','cn-south','us-east'];returnarray_filter($all,function(string $region){$failures=(int)$this->redis->get("geo:health:{$region}");return$failures<10;});}// 跨地域请求转发publicfunctionforwardTo(string $region,string $path,array $payload):array{$endpoints=['cn-north'=>env('GEO_ENDPOINT_CN_NORTH'),'cn-south'=>env('GEO_ENDPOINT_CN_SOUTH'),'us-east'=>env('GEO_ENDPOINT_US_EAST'),];$response=\Hyperf\Support\make(\GuzzleHttp\Client::class)->post($endpoints[$region].$path,['json'=>$payload,'timeout'=>5,'headers'=>['X-Forwarded-Region'=>$this->region,'X-Trace-Id'=>\Hyperf\Context\Context::get('trace_id'),],]);returnjson_decode($response->getBody(),true);}}---6.多活中间件（自动故障转移）<?php namespace App\Middleware;use App\Governance\GeoFailover;use Psr\Http\Message\ResponseInterface;use Psr\Http\Message\ServerRequestInterface;use Psr\Http\Server\MiddlewareInterface;use Psr\Http\Server\RequestHandlerInterface;classMultiActiveMiddlewareimplementsMiddlewareInterface{publicfunction__construct(private GeoFailover $failover){}publicfunctionprocess(ServerRequestInterface $request,RequestHandlerInterface $handler):ResponseInterface{// 被其他地域转发来的请求，直接处理不再转发if($request->getHeaderLine('X-Forwarded-Region')){return$handler->handle($request);}try{$response=$handler->handle($request);if($response->getStatusCode()>=500){$this->failover->recordFailure();}return$response;}catch(\Throwable $e){$this->failover->recordFailure();// 本地故障，转发到其他可用地域$regions=$this->failover->availableRegions();$target=current(array_filter($regions,fn($r)=>$r!==env('REGION')));if($target){$data=$this->failover->forwardTo(region:$target,path:$request->getUri()->getPath(),payload:(array)$request->getParsedBody(),);return$this->response->json($data);}throw$e;}}}---7.Nacos 多地域配置<?php// config/autoload/config_center.phpreturn['drivers'=>['nacos'=>[// 每个地域独立 Nacos 集群'client'=>['host'=>match(env('REGION')){'cn-north'=>env('NACOS_HOST_CN_NORTH'),'cn-south'=>env('NACOS_HOST_CN_SOUTH'),'us-east'=>env('NACOS_HOST_US_EAST'),},'port'=>8848,],'namespace_id'=>env('REGION'),// 命名空间 = 地域],],];---8.Kafka 跨地域复制（MirrorMaker2） # kafka-mm2.properties clusters=cn-north,cn-south,us-east cn-north.bootstrap.servers=kafka-cn-north:9092cn-south.bootstrap.servers=kafka-cn-south:9092us-east.bootstrap.servers=kafka-us-east:9092# 双向同步 cn-north->cn-south.enabled=truecn-south->cn-north.enabled=truecn-north->us-east.enabled=true# 同步 geo.*前缀 topic cn-north->cn-south.topics=geo\..*cn-south->cn-north.topics=geo\..*# 防止循环复制 replication.policy.class=org.apache.kafka.connect.mirror.DefaultReplicationPolicy---核心要点：-Snowflake 高位注入地域码，ID 全局唯一且可溯源到产生地域-LWW（Last-Write-Wins）用microtime(true)时间戳解决写冲突，适合订单/用户场景-X-Forwarded-Region 防止转发循环，被转发请求直接处理-MirrorMaker2 自动过滤已复制消息（topic 前缀），防止跨地域无限循环同步