<?php
// 安装：pecl install swoole
$server = new Swoole\HTTP\Server('0.0.0.0', 9501);// 百万并发配置（16进程+32任务线程）
$server->set(['worker_num' => 16,       // 16个Worker进程'reactor_num' => 8,        // 每个Worker配8个线程'task_worker_num' => 32,   // 32个异步任务进程'enable_coroutine' => true // 开启协程
]);// 设备数据上报入口
$server->on('request', function ($request, $response) use ($server) {// 1. 投递异步任务（0毫秒等待）$taskId = $server->task(['device_id' => $request->get['id'],'sensor_data' => $request->post]);// 2. 立即响应202 Accepted$response->header('Content-Type', 'application/json');$response->end(json_encode(['code' => 202, 'task_id' => $taskId]));
});// 异步任务处理器（实际业务逻辑）
$server->on('task', function ($server, $taskId, $workerId, $data) {// 3. 处理设备数据（不阻塞主进程）$result = (new SensorProcessor())->process($data);// 4. 存储到Redis分片$shard = crc32($data['device_id']) % 32;Redis::select($shard);Redis::set('device:'.$data['device_id'], json_encode($result));return $taskId; // 返回任务ID
});$server->start(); // 启动服务

<?php
class ShardingProcessor {const SHARDS = 32; // 分片数量public function handle(array $data) {// 1. 按设备ID哈希分片$shardIndex = crc32($data['device_id']) % self::SHARDS;$queueName = "sensor_queue_{$shardIndex}";// 2. 写入分片队列Redis::rpush($queueName, json_encode($data));// 3. 动态启停消费者$this->adjustConsumer($shardIndex);}private function adjustConsumer(int $shardIndex) {$queueLen = Redis::llen("sensor_queue_{$shardIndex}");// 每500条消息启动1个消费者$required = ceil($queueLen / 500);$current = $this->getConsumerCount($shardIndex);if ($required > $current) {// 启动新消费者$this->startConsumer($shardIndex, $required - $current);} elseif ($current > $required) {// 停止多余消费者$this->stopConsumer($shardIndex, $current - $required);}}private function startConsumer(int $shardIndex, int $count) {for ($i = 0; $i < $count; $i++) {$cmd = "php consumer.php {$shardIndex} > /dev/null 2>&1 &";shell_exec($cmd);}}
}// consumer.php 示例
$shardIndex = $argv[1];
$queueName = "sensor_queue_{$shardIndex}";$redis = new Redis();
$redis->connect('127.0.0.1');while (true) {// 30秒超时阻塞获取$data = $redis->blpop($queueName, 30);if ($data) {(new SensorAnalyser())->analyse(json_decode($data[1], true));}
}

<?php
class InferenceBalancer {private $gpuNodes = ['gpu01' => ['addr' => '192.168.10.101:5000', 'load' => 0],'gpu02' => ['addr' => '192.168.10.102:5000', 'load' => 0],'gpu03' => ['addr' => '192.168.10.103:5000', 'load' => 0]];public function predict(array $data) {// 1. 选择负载最低的GPU节点$node = $this->selectNode();// 2. gRPC调用（毫秒级响应）$client = new GrpcClient($node['addr']);$result = $client->predict($data);// 3. 更新节点负载$this->updateNodeLoad($node['id']);return $result;}private function selectNode(): array {$lowestLoad = 100;$selected = null;foreach ($this->gpuNodes as $id => $node) {if ($node['load'] < $lowestLoad) {$lowestLoad = $node['load'];$selected = ['id' => $id, 'addr' => $node['addr']];}}return $selected;}private function updateNodeLoad(string $nodeId) {// 模拟负载增加（实际通过API获取）$this->gpuNodes[$nodeId]['load'] = min($this->gpuNodes[$nodeId]['load'] + 5, 100);// 10秒后负载下降Swoole\Timer::after(10000, function() use ($nodeId) {$this->gpuNodes[$nodeId]['load'] = max($this->gpuNodes[$nodeId]['load'] - 8, 0);});}
}

# Nginx动态扩缩容配置
upstream edge_nodes {zone edge_pool 64k;server 192.168.1.101:9501;server 192.168.1.102:9501;
}# 扩容策略：CPU>80%持续30秒
@scale_up {cpu_usage 80 30s;add_server 192.168.1.103:9501;add_server 192.168.1.104:9501;
}# 缩容策略：CPU<30%持续5分钟
@scale_down {cpu_usage 30 5m;remove_server 192.168.1.104:9501;
}server {listen 80;location / {dynamic_upstream fallback=sticky interval=5s up=@scale_up down=@scale_down;proxy_pass http://edge_nodes;}
}

<?php
class TimeSeriesWriter {const BATCH_SIZE = 1000; // 每批1000条public function write(array $sensorPoints) {$batch = [];foreach ($sensorPoints as $point) {// 生成InfluxDB行协议$batch[] = sprintf('sensor,device=%s,type=%s value=%.2f %d',$point['device_id'],$point['sensor_type'],$point['value'],$point['timestamp']);if (count($batch) >= self::BATCH_SIZE) {$this->flush($batch);$batch = [];}}$this->flush($batch);}private function flush(array $lines) {$ch = curl_init('http://influxdb:8086/write?db=factory');curl_setopt_array($ch, [CURLOPT_POST => true,CURLOPT_POSTFIELDS => implode("\n", $lines),CURLOPT_TIMEOUT_MS => 300 // 300ms超时]);curl_exec($ch);}
}

class DynamicConsumer {public function adjust(string $queueName) {$length = Redis::llen($queueName);$required = ceil($length / 500); // 每500条1个消费者// 获取当前消费者数$current = count($this->getConsumers($queueName));if ($required > $current) {$this->addConsumers($queueName, $required - $current);} elseif ($current > $required) {$this->removeConsumers($queueName, $current - $required);}}
}

class EdgeFallback {const CLOUD_TIMEOUT = 50; // 50mspublic function processRequest(array $request) {$start = microtime(true);// 1. 尝试云端处理try {$result = $this->cloudProcess($request, self::CLOUD_TIMEOUT);return $result;} catch (TimeoutException $e) {// 2. 超时后本地降级处理return $this->localProcess($request);}}private function localProcess(array $request) {// 使用简化模型快速推理return (new FastPredictor())->predict($request);}
}

class GPULoadMonitor {public function updateAll() {foreach ($this->nodes as $id => $node) {$load = $this->getNodeLoad($node['ip']);Redis::hset('gpu_load', $id, $load);}}public function getOptimalNode(): string {$lowestLoad = 100;$selected = '';foreach (Redis::hgetall('gpu_load') as $id => $load) {if ($load < $lowestLoad) {$lowestLoad = $load;$selected = $this->nodes[$id]['addr'];}}return $selected;}
}