swoole方案 实时监控大盘推送中心

业务服务--写--Kafka---Swoole消费--WebSocket推--浏览器ECharts实时刷新 Kafka 当缓冲层业务打点不管推送快不快Swoole 从 Kafka 拉数据有新数据就推给所有看板页面。---代码?php// composer require longlang/phpkafkarequire__DIR__./vendor/autoload.php;uselonglang\phpkafka\Consumer\Consumer;uselonglang\phpkafka\Consumer\ConsumerConfig;uselonglang\phpkafka\Producer\Producer;uselonglang\phpkafka\Producer\ProducerConfig;// 存所有在线的看板连接$clientsnewSwoole\Table(116);$clients-column(fd,Swoole\Table::TYPE_INT);$clients-create();$servernewSwoole\WebSocket\Server(0.0.0.0,9502);$server-set([worker_num2]);// worker 0 启动时跑 Kafka 消费者协程$server-on(workerStart,function($server,$workerId)use($clients){if($workerId!0)return;Swoole\Coroutine::create(function()use($server,$clients){$confignewConsumerConfig();$config-setBrokers(127.0.0.1:9092);$config-setTopic(metrics);$config-setGroupId(dashboard);$config-setAutoCommit(true);$consumernewConsumer($config);while(true){$msg$consumer-consume();if(!$msg){Swoole\Coroutine::sleep(0.05);continue;}// 广播给所有看板foreach($clientsas$row){if($server-isEstablished($row[fd])){$server-push($row[fd],$msg-getValue());}}$consumer-ack($msg);}});});$server-on(open,fn($s,$req)$clients-set($req-fd,[fd$req-fd]));$server-on(close,fn($s,$fd)$clients-del($fd));$server-on(message,fn()null);// 只推不收$server-start();---模拟打点另起一个脚本往 Kafka 写数据?php// producer.php 模拟业务服务打点require__DIR__./vendor/autoload.php;uselonglang\phpkafka\Producer\Producer;uselonglang\phpkafka\Producer\ProducerConfig;$confignewProducerConfig();$config-setBrokers(127.0.0.1:9092);$producernewProducer($config);while(true){$datajson_encode([timedate(H:i:s),cpurand(10,90),memrand(30,80),qpsrand(100,9999),]);$producer-send(metrics,$data);echo发送:$data\n;sleep(1);}---前端 ECharts直接浏览器开!DOCTYPEhtmlscript srchttps://cdn.jsdelivr.net/npm/echarts/dist/echarts.min.js/scriptdiv idchartstylewidth:100%;height:400px/divscriptconstchartecharts.init(document.getElementById(chart));consttimes[],cpuData[],qpsData[];chart.setOption({xAxis:{type:category,data:times},yAxis:[{type:value,name:CPU%},{type:value,name:QPS}],series:[{name:CPU,type:line,data:cpuData,smooth:true},{name:QPS,type:bar,data:qpsData,yAxisIndex:1},]});constwsnewWebSocket(ws://localhost:9502);ws.onmessage({data}){constdJSON.parse(data);times.push(d.time);cpuData.push(d.cpu);qpsData.push(d.qps);if(times.length30){times.shift();cpuData.shift();qpsData.shift();}// 只显示最近30条chart.setOption({xAxis:{data:times},series:[{data:cpuData},{data:qpsData}]});};/script---解释 Kafka 在中间的意义业务服务打点只管往 Kafka 写不管谁在看板。Swoole 挂了重启Kafka 里的数据还在不丢。业务和推送完全解耦Swoole\Coroutine::create在 worker 进程里开一个协程跑 Kafka 消费不阻塞 WebSocket 处理。Kafka 没消息时sleep(0.05)让出CPU有消息立刻处理 只让 worker0消费if($workerId!0)return不然每个 worker 都消费同一条消息广播N遍$consumer-ack($msg)手动提交 offset告诉 Kafka 这条消息处理完了下次不用再发。放在广播之后保证推出去了再 ack Swoole\Table 存 fd共享内存worker0的消费协程能读到 worker1接进来的客户端连接---跑起来# 1. 启动 Kafka本地测试用 dockerdocker run-d-p9092:9092apache/kafka# 2. 启动推送服务php server.php# 3. 启动打点模拟php producer.php# 4. 浏览器打开 index.html看折线图实时跳动