swoole方案 WebSocket 下推消息优先级队列

WebSocket 推消息优先级队列大白话先说清楚普通弹幕 哈哈哈哈哈 优先级 1 (低) 礼物打赏 送了火箭 优先级 2 (中) 系统广播 服务器维护通知 优先级 3 (高) 队列里同时有这三条 先发系统广播 → 再发礼物 → 最后发弹幕为什么要优先队列直播间一秒几千条弹幕普通消息把队列塞满了礼物消息卡在后面半天没发出去打赏的人当然骂娘。代码?php// 优先队列数字越大越先出队classMsgQueueextendsSplPriorityQueue{// 相同优先级按插入顺序排先进先出publicfunctioncompare($a,$b):int{if($a[priority]$b[priority]){return$a[seq]$b[seq]?1:-1;}return$a[priority]$b[priority]?-1:1;}}$servernewSwoole\WebSocket\Server(0.0.0.0,9502);$queuenewMsgQueue();$seq0;// 插入序号保证同优先级有序// 优先级常量define(PRI_CHAT,1);// 普通弹幕define(PRI_GIFT,2);// 礼物打赏define(PRI_SYSTEM,3);// 系统广播$server-on(open,function($server,$req){echofd{$req-fd}进来了\n;});$server-on(message,function($server,$frame)use($queue,$seq){$datajson_decode($frame-data,true);// 根据消息类型定优先级$primatch($data[type]??){systemPRI_SYSTEM,giftPRI_GIFT,defaultPRI_CHAT,};// 塞进队列$queue-insert([fd$frame-fd,payload$data,priority$pri,seq$seq,],$pri);echo收到消息 type{$data[type]}pri{$pri}\n;});$server-on(close,function($server,$fd){echofd{$fd}走了\n;});// 每10ms消费队列往所有连接推$server-on(workerStart,function($server)use($queue){Swoole\Timer::tick(10,function()use($server,$queue){// 每次最多发50条防止一次发太多卡住$limit50;while(!$queue-isEmpty()$limit--0){$msg$queue-extract();foreach($server-connectionsas$fd){if(!$server-isEstablished($fd))continue;$server-push($fd,json_encode($msg[payload]));}}});});$server-start();测试客户端?php// 模拟同时发三种消息go(function(){$cnewSwoole\Coroutine\Http\Client(127.0.0.1,9502);$c-upgrade(/);// 同时扔进去看谁先出来$c-push(json_encode([typechat,msg哈哈哈哈]));$c-push(json_encode([typegift,msg送了火箭]));$c-push(json_encode([typesystem,msg系统维护]));// 接收推过来的消息while(true){$frame$c-recv();if($frame)echo收到{$frame-data}\n;}});队列出队顺序示意塞入顺序 出队顺序 ① chat pri1 ③ system pri3 ← 先出 ② gift pri2 ② gift pri2 ③ system pri3 ① chat pri1 ← 最后出核心就三件事1. SplPriorityQueue PHP自带优先队列不用装任何包 2. match 判类型 决定这条消息值几分 3. Timer::tick(10ms) 每10毫秒消费一批高分的先出去