《Swoole 6实现远程会诊信令服务:WebSocket/WebRTC Signaling+房间管理+录制控制》
·
远程会诊信令服务和前面几篇的根本区别:它是 WebSocket 长连接 + 状态机 +
广播,核心不是吞吐也不是大文件,而是把音视频协商的"暗号"在房间内正确地转发,以及房间/成员/角色/录制的状态管理。
先把一个最关键的认知放在前面,后面代码全据此设计。
---
〇、最重要的一个认知:信令服务器"不碰音视频"
很多人误以为视频会诊的服务器要处理视频流——不是。WebRT的音视频是端到端直连(P2P)或经 SFU
媒体服务器传输的,信令服务器只负责"牵线搭桥":帮两个浏览器交换三样暗号——
- SDP(我支持什么编解码、分辨率)——offer/answer;
- ICE candidate(我在网络上的可达地址)——用于穿透NAT;
- 房间/成员事件(谁进来了、谁说话、谁开始录制)。
所以信令服务器又轻又关键:它转发的全是几 KB 的小 JSON,但一条 ICE 没转对,视频就连不上。Swoole 的 WebSocket
做这个是绝配——海量长连接、低延迟广播。
录制也一样:信令服务器不录视频,它只是下达"开始/停止录制"的控制指令,真正干活的是录制端(浏览器端 MediaRecorder,或服务端
SFU/MediaSoup 的录制模块)。本篇做的是录制控制信令 + 状态/审计——这才是信令服务器的职责。
---
一、大白话:这个信令服务在干嘛
把它想成一个视频会议的"总机接线员":
1. 接线(WebSocket 连接 + 鉴权):医生、专家、患者各自用浏览器连上来,握手时带个 token
表明身份和角色(主诊/会诊专家/观察者)。
2. 进会议室(房间管理):每个会诊是一个"房间"。接线员维护"哪个房间里有谁",新人进来要通知房间里所有老成员"来人了",并把"房
里已有谁"告诉新人——这样双方才知道要和谁建立视频连接。
3. 牵线(信令转发,核心):医生 A 想和专家 B 通视频,A 把自己的"暗号(SDP/ICE)"发给接线员说"转给 B",接线员精准转发给
B(不是广播,是点对点定向转发)。B 回复暗号同理转回 A。暗号对上,A、B 的浏览器就直接建立视频流了,不再经过服务器。
4. 会议控制(录制/静音/踢人):主诊医生喊"开始录制",接线员广播这条控制指令给全房间(各端据此启动录制),并记录"本次会诊从 X
时刻开始录制"做合规审计。停止、静音、移出成员同理。
5. 善后(断线清理):有人掉线,接线员要通知房里其他人"某某离开了",让他们关掉对应的视频窗口;房间空了就回收。
---
二、完整代码(单文件:信令服务器 + 浏览器测试页 + 命令行假客户端)
<?php
declare(strict_types=1);
use Swoole\WebSocket\Server;
use Swoole\WebSocket\Frame;
use Swoole\Http\Request;
use Swoole\Http\Response;
use Swoole\Table;
use Swoole\Atomic;
use Swoole\Timer;
/**
* ============================================================
* 远程会诊信令服务(Swoole 6 WebSocket)
* WebRTC Signaling 转发 + 房间管理 + 角色权限 + 录制控制
* ------------------------------------------------------------
* php sig.php server # 启动信令服务(ws://0.0.0.0:9400)
* 浏览器开 http://127.0.0.1:9400/ -> 自带测试页,两个标签页可互通
* ------------------------------------------------------------
* 鉴权(演示版):连接 URL 带 ?room=R1&user=张医生&role=host&token=...
* 角色 role: host(主诊,可控会议/录制) | expert(会诊专家) | observer(观察者)
* ============================================================
*/
final class Cfg
{
public const HOST = '0.0.0.0';
public const PORT = 9400;
public const HEARTBEAT = 40; // 秒;超时未 ping 视为掉线
public const AUDIT = __DIR__ . '/sig_audit.log';
// 演示用静态 token;生产应校验 JWT / 调鉴权服务
public const TOKENS = ['demo-secret'];
}
function runServer(): void
{
$server = new Server(Cfg::HOST, Cfg::PORT);
$server->set([
'worker_num' => 4,
'enable_coroutine' => true,
'max_request' => 0, // 长连接服务不回收 Worker
'heartbeat_idle_time' => 0, // 用业务层 ping/pong 判活
'heartbeat_check_interval' => 0,
'open_websocket_ping_frame' => true, // 接管 ping/pong
'open_websocket_pong_frame' => true,
]);
/* ---- 共享内存:连接、房间成员(跨 Worker)---- */
$conns = new Table(1 << 16); // fd -> 会话
$conns->column('room', Table::TYPE_STRING, 64);
$conns->column('user', Table::TYPE_STRING, 64);
$conns->column('role', Table::TYPE_STRING, 16);
$conns->column('peer', Table::TYPE_STRING, 40); // 会内唯一 peerId
$conns->column('last', Table::TYPE_INT);
$conns->create();
$rooms = new Table(8192); // room -> 状态
$rooms->column('recording', Table::TYPE_INT);
$rooms->column('rec_by', Table::TYPE_STRING, 64);
$rooms->column('rec_since', Table::TYPE_INT);
$rooms->column('count', Table::TYPE_INT);
$rooms->create();
$peerSeq = new Atomic(0);
/* ---------- 工具:发送 / 房间广播 / 定向转发 ---------- */
$send = function (Server $srv, int $fd, array $msg) {
if ($srv->isEstablished($fd)) $srv->push($fd, json_encode($msg, JSON_UNESCAPED_UNICODE));
};
// 房间内成员列表(fd 列表)
$roomFds = function (Table $conns, string $room, int $exceptFd = 0): array {
$fds = [];
foreach ($conns as $fd => $c) if ($c['room'] === $room && (int) $fd !== $exceptFd) $fds[] = (int) $fd;
return $fds;
};
// 按 peerId 找 fd
$fdOfPeer = function (Table $conns, string $room, string $peer): ?int {
foreach ($conns as $fd => $c) if ($c['room'] === $room && $c['peer'] === $peer) return (int) $fd;
return null;
};
$audit = function (array $rec) {
$rec['ts'] = date('c');
@file_put_contents(Cfg::AUDIT, json_encode($rec, JSON_UNESCAPED_UNICODE) . "\n", FILE_APPEND | LOCK_EX);
};
/* ---------- 内嵌测试页(HTTP GET /) ---------- */
$server->on('request', function (Request $req, Response $res) {
if (($req->server['request_uri'] ?? '/') === '/') {
$res->header('Content-Type', 'text/html; charset=utf-8');
$res->end(testPageHtml());
} else { $res->status(404); $res->end('not found'); }
});
/* ---------- WebSocket 握手即鉴权 + 入房 ---------- */
$server->on('open', function (Server $srv, Request $req)
use ($conns, $rooms, $peerSeq, $send, $roomFds, $audit) {
$q = $req->get ?? [];
$room = (string) ($q['room'] ?? '');
$user = (string) ($q['user'] ?? '');
$role = (string) ($q['role'] ?? 'observer');
$token = (string) ($q['token'] ?? '');
$fd = $req->fd;
// 鉴权(演示:校验静态 token;生产换 JWT 验签 + 会诊权限校验)
if ($room === '' || $user === '' || !in_array($token, Cfg::TOKENS, true)) {
$send($srv, $fd, ['type' => 'error', 'reason' => 'auth-failed']);
$srv->disconnect($fd, 4001, 'auth-failed');
return;
}
if (!in_array($role, ['host', 'expert', 'observer'], true)) $role = 'observer';
$peer = 'peer' . $peerSeq->add(1);
$conns->set((string) $fd, ['room' => $room, 'user' => $user, 'role' => $role,
'peer' => $peer, 'last' => time()]);
$rooms->set($room, ['recording' => $rooms->get($room)['recording'] ?? 0,
'rec_by' => $rooms->get($room)['rec_by'] ?? '',
'rec_since' => $rooms->get($room)['rec_since'] ?? 0,
'count' => ($rooms->get($room)['count'] ?? 0) + 1]);
// 告诉新人:你的 peerId + 房里已有谁(据此发起 P2P 连接)
$members = [];
foreach ($roomFds($conns, $room, $fd) as $ofd) {
$o = $conns->get((string) $ofd);
$members[] = ['peer' => $o['peer'], 'user' => $o['user'], 'role' => $o['role']];
}
$r = $rooms->get($room);
$send($srv, $fd, ['type' => 'welcome', 'peer' => $peer, 'room' => $room, 'role' => $role,
'members' => $members,
'recording' => $r['recording'], 'rec_since' => $r['rec_since']]);
// 通知房里老成员:来新人了
foreach ($roomFds($conns, $room, $fd) as $ofd)
$send($srv, $ofd, ['type' => 'peer-joined', 'peer' => $peer, 'user' => $user, 'role' => $role]);
$audit(['ev' => 'join', 'room' => $room, 'user' => $user, 'role' => $role, 'peer' => $peer]);
echo "[sig] {$user}({$role}) 进入房间 {$room},当前 " . $rooms->get($room)['count'] . " 人\n";
});
/* ---------- 消息:信令转发 + 房间/录制控制 ---------- */
$server->on('message', function (Server $srv, Frame $frame)
use ($conns, $rooms, $send, $roomFds, $fdOfPeer, $audit) {
$fd = $frame->fd;
$me = $conns->get((string) $fd);
if ($me === false) return;
$conns->set((string) $fd, ['last' => time()]); // 刷新活跃
$m = json_decode($frame->data, true);
if (!is_array($m) || !isset($m['type'])) return;
$room = $me['room'];
switch ($m['type']) {
// ——WebRTC 三大信令:定向转发给目标 peer(不广播)——
case 'offer':
case 'answer':
case 'candidate':
$target = $fdOfPeer($conns, $room, $m['to'] ?? '');
if ($target !== null) {
$send($srv, $target, [
'type' => $m['type'],
'from' => $me['peer'], // 告诉对方这是谁发来的
'sdp' => $m['sdp'] ?? null,
'candidate' => $m['candidate'] ?? null,
]);
}
break;
// ——房间内聊天/标注等业务消息:广播 ——
case 'chat':
foreach ($roomFds($conns, $room, $fd) as $ofd)
$send($srv, $ofd, ['type' => 'chat', 'from' => $me['user'], 'text' => (string) ($m['text'] ??
'')]);
break;
// ——录制控制:仅 host 可操作 ——
case 'rec-start':
case 'rec-stop':
if ($me['role'] !== 'host') {
$send($srv, $fd, ['type' => 'error', 'reason' => 'permission-denied: host only']);
break;
}
$on = $m['type'] === 'rec-start';
$rooms->set($room, ['recording' => $on ? 1 : 0,
'rec_by' => $on ? $me['user'] : '',
'rec_since' => $on ? time() : 0]);
// 广播控制指令:各端据此启动/停止本地或 SFU 录制
foreach ($roomFds($conns, $room) as $ofd)
$send($srv, $ofd, ['type' => $on ? 'recording-started' : 'recording-stopped',
'by' => $me['user'], 'at' => time()]);
$audit(['ev' => $on ? 'rec-start' : 'rec-stop', 'room' => $room, 'by' => $me['user']]);
echo "[sig] 房间 {$room} 录制" . ($on ? '开始' : '停止') . "(by {$me['user']})\n";
break;
// ——主持人控制:静音/移出成员(仅 host)——
case 'mute':
case 'kick':
if ($me['role'] !== 'host') { $send($srv, $fd, ['type' => 'error', 'reason' => 'host only']); break; }
$tfd = $fdOfPeer($conns, $room, $m['target'] ?? '');
if ($tfd === null) break;
if ($m['type'] === 'mute') {
$send($srv, $tfd, ['type' => 'force-mute', 'by' => $me['user']]);
} else {
$send($srv, $tfd, ['type' => 'kicked', 'by' => $me['user']]);
$srv->disconnect($tfd, 4003, 'kicked-by-host');
}
$audit(['ev' => $m['type'], 'room' => $room, 'by' => $me['user'], 'target' => $m['target'] ?? '']);
break;
case 'ping':
$send($srv, $fd, ['type' => 'pong', 't' => time()]);
break;
}
});
/* ---------- 断开:通知房间 + 清理 ---------- */
$server->on('close', function (Server $srv, int $fd)
use ($conns, $rooms, $send, $roomFds, $audit) {
$me = $conns->get((string) $fd);
if ($me === false) return;
$room = $me['room'];
$conns->del((string) $fd);
foreach ($roomFds($conns, $room, $fd) as $ofd)
$send($srv, $ofd, ['type' => 'peer-left', 'peer' => $me['peer'], 'user' => $me['user']]);
$r = $rooms->get($room);
if ($r !== false) {
$left = max(0, $r['count'] - 1);
if ($left === 0) $rooms->del($room);
else $rooms->set($room, ['count' => $left]);
}
$audit(['ev' => 'leave', 'room' => $room, 'user' => $me['user'], 'peer' => $me['peer']]);
echo "[sig] {$me['user']} 离开房间 {$room}\n";
});
/* ---------- 心跳巡检:踢掉僵尸连接(只在 Worker 0)---------- */
$server->on('workerStart', function (Server $srv, int $wid) use ($conns) {
if ($wid !== 0) return;
echo "会诊信令服务启动 ws://" . Cfg::HOST . ":" . Cfg::PORT . " (测试页同地址 http://127.0.0.1:" . Cfg::PORT
. "/)\n";
Timer::tick(10000, function () use ($srv, $conns) {
$now = time();
foreach ($conns as $fd => $c)
if ($now - $c['last'] > Cfg::HEARTBEAT && $srv->isEstablished((int) $fd))
$srv->disconnect((int) $fd, 4000, 'heartbeat-timeout');
});
});
$server->start();
}
/* ============================================================
* 内嵌浏览器测试页:真实 WebRTC 端到端(两个标签页可互看视频)
* ============================================================ */
function testPageHtml(): string
{
return <<<'HTML'
<!doctype html><html lang="zh"><head><meta charset="utf-8">
<title>会诊信令测试</title>
<style>body{font-family:sans-serif;margin:20px}video{width:320px;background:#222;margin:6px}
input,button{padding:6px;margin:3px}#log{white-space:pre-wrap;background:#f4f4f4;padding:8px;height:160px;overflow:aut
o}</style>
</head><body>
<h3>远程会诊信令测试(开两个标签页,房间填同一个)</h3>
<div>
房间<input id="room" value="R1"> 姓名<input id="user" value="医生A">
角色<select id="role"><option value="host">host主诊</option><option value="expert">expert专家</option><option
value="observer">observer观察</option></select>
<button onclick="join()">加入</button>
<button onclick="recStart()">开始录制</button>
<button onclick="recStop()">停止录制</button>
</div>
<div><video id="local" autoplay muted playsinline></video><video id="remote" autoplay playsinline></video></div>
<div id="log"></div>
<script>
const log=(...a)=>{document.getElementById('log').textContent+=a.join(' ')+'\n';};
let ws, myPeer, localStream, pcs={}; // pcs: peerId -> RTCPeerConnection
const ICE={iceServers:[{urls:'stun:stun.l.google.com:19302'}]};
async function join(){
localStream=await navigator.mediaDevices.getUserMedia({video:true,audio:true});
document.getElementById('local').srcObject=localStream;
const room=room_.value,user=user_.value,role=role_.value;
ws=new
WebSocket(`ws://${location.host}/?room=${room}&user=${encodeURIComponent(user)}&role=${role}&token=demo-secret`);
ws.onmessage=async e=>{const m=JSON.parse(e.data); await onMsg(m);};
ws.onclose=e=>log('断开',e.code,e.reason);
}
const
room_=document.getElementById('room'),user_=document.getElementById('user'),role_=document.getElementById('role');
function pc(peer){
if(pcs[peer])return pcs[peer];
const p=new RTCPeerConnection(ICE);
localStream.getTracks().forEach(t=>p.addTrack(t,localStream));
p.onicecandidate=e=>{if(e.candidate)ws.send(JSON.stringify({type:'candidate',to:peer,candidate:e.candidate}));};
p.ontrack=e=>{document.getElementById('remote').srcObject=e.streams[0];};
pcs[peer]=p; return p;
}
async function onMsg(m){
switch(m.type){
case 'welcome':
myPeer=m.peer; log('已加入,我是',myPeer,'录制中:',m.recording);
for(const mem of m.members){ const p=pc(mem.peer); // 对已有成员主动发 offer
const o=await p.createOffer(); await p.setLocalDescription(o);
ws.send(JSON.stringify({type:'offer',to:mem.peer,sdp:o})); }
break;
case 'peer-joined': log('新成员',m.user); break;
case 'offer':{ const p=pc(m.from); await p.setRemoteDescription(m.sdp);
const a=await p.createAnswer(); await p.setLocalDescription(a);
ws.send(JSON.stringify({type:'answer',to:m.from,sdp:a})); break;}
case 'answer': await pcs[m.from]?.setRemoteDescription(m.sdp); break;
case 'candidate': await pcs[m.from]?.addIceCandidate(m.candidate); break;
case 'peer-left': log('离开',m.user); pcs[m.peer]?.close(); delete pcs[m.peer]; break;
case 'recording-started': log('🔴 录制开始 by',m.by); break;
case 'recording-stopped': log('⏹录制停止 by',m.by); break;
case 'force-mute': localStream.getAudioTracks().forEach(t=>t.enabled=false); log('被主持人静音'); break;
case 'error': log('错误:',m.reason); break;
}
}
function recStart(){ws&&ws.send(JSON.stringify({type:'rec-start'}));}
function recStop(){ws&&ws.send(JSON.stringify({type:'rec-stop'}));}
</script></body></html>
HTML;
}
/* ============================================================
* 命令行假客户端:不开浏览器也能验证“信令转发/房间/录制”逻辑
* ============================================================ */
function runFakeClient(string $room, string $user, string $role): void
{
Swoole\Coroutine\run(function () use ($room, $user, $role) {
$cli = new Swoole\Coroutine\Http\Client(Cfg::HOST === '0.0.0.0' ? '127.0.0.1' : Cfg::HOST, Cfg::PORT);
$cli->set(['timeout' => -1]);
$path = "/?room=$room&user=" . rawurlencode($user) . "&role=$role&token=demo-secret";
if (!$cli->upgrade($path)) { echo "升级 WebSocket 失败\n"; return; }
echo "[$user] 已连接\n";
// 收消息协程
Swoole\Coroutine::create(function () use ($cli, $user) {
while (true) {
$f = $cli->recv();
if ($f === false || $f === '') break;
if (is_object($f) && isset($f->data)) echo "[$user] <= {$f->data}\n";
}
});
// host 演示:5 秒后开始录制,10 秒后停止
if ($role === 'host') {
Swoole\Coroutine::sleep(5);
$cli->push(json_encode(['type' => 'rec-start']));
Swoole\Coroutine::sleep(5);
$cli->push(json_encode(['type' => 'rec-stop']));
}
Swoole\Coroutine::sleep(30);
$cli->close();
});
}
// -------- 入口 --------
$cmd = $argv[1] ?? 'server';
match ($cmd) {
'server' => runServer(),
'client' => runFakeClient($argv[2] ?? 'R1', $argv[3] ?? '医生A', $argv[4] ?? 'host'),
default => print("用法:\n php sig.php server\n php sig.php client <room> <user> <role>\n"),
};
---
三、怎么跑
方式一:真实浏览器端到端(推荐,能看到视频)
php sig.php server
# 浏览器开两个标签页,都访问 http://127.0.0.1:9400/
# 标签1:房间 R1、姓名"医生A"、角色 host,点"加入"(允许摄像头)
# 标签2:房间 R1、姓名"专家B"、角色 expert,点"加入"
# →两个标签页互相看到对方视频(P2P 直连,服务器只转了信令)
# →在 host 标签点"开始录制",两边日志都出现 "🔴 录制开始"
▎ 注:getUserMedia 在 127.0.0.1 上允许;若用域名/局域网 IP,浏览器要求 HTTPS/WSS(生产必须上 TLS)。
方式二:命令行验证信令逻辑(不用浏览器)
php sig.php server
php sig.php client R1 医生A host # 5秒后自动触发录制开始/停止
php sig.php client R1 专家B expert # 会收到 peer-joined / recording-started 等
你会看到 welcome / peer-joined / recording-started / recording-stopped 等消息在两个客户端间正确流转,审计写进
sig_audit.log。
---
四、工程要点(大白话)
- 信令服务器只转"暗号"不碰视频:offer/answer/candidate 三类消息是定向转发(to: 某peer →精准 push 给那个
fd),不是广播。视频流在两个浏览器之间 P2P 直连,完全不经过服务器——所以一台Swoole
信令服务器能撑起海量并发会诊,因为它只搬几 KB 的小 JSON。
- 房间状态放共享内存 Table:谁在哪个房间、什么角色、录制开没开,全在 Swoole\Table,所有 Worker
共享。新人进来要"通知老成员 +
把成员列表给新人",断线要"通知房间其他人关窗口"——这套加入/离开广播是多人会诊视频能正确显示每个人画面的关键。
- peerId 而非 fd 对外:对客户端暴露的是会内唯一的 peerId,不是底层 fd。这样转发用 to: peerX,逻辑清晰,也不泄露连接细节。
- 角色权限在服务端强校验:"开始录制/静音/踢人"这类敏感操作,服务端检查
role==='host',不信任客户端。医疗场景这点尤其重要——录制涉及患者知情同意,不能让观察者随便录。
- 录制是"控制信令 + 状态 + 审计",不是录流本身:host 一声令下,服务器广播 recording-started
并记下"谁、几点开始录",各端据此启动录制(浏览器 MediaRecorder 或服务端
SFU)。服务器留的是合规审计轨迹(谁在何时录了什么房间),这才是信令层该负责的。
- 业务层心跳判活:WebRTC 连接可能"网在人不在",靠业务 ping 刷新
last,巡检定时器踢掉超时僵尸连接,避免房间里挂着一堆"幽灵成员"。
---
五、诚实提示(重要)
1. 这套信令 + 浏览器 WebRTC 是真能跑通端到端视频的——P2P(mesh)模式下,2~人会诊完全够用,代码里测试页就是真实可用的
WebRTC。这部分我有把握。
2. 人多了必须上 SFU:mesh 模式下每个人要和其他所有人各建一条连接,5 人以上上行带宽和 CPU 就扛不住。多人会诊的工业标准是
SFU 媒体服务器(MediaSoup / Janus / LiveKit / Jitsi)——浏览器只把流发给SFU,SFU 再转发。这时 Swoole
信令服务器依然是你这套(负责房间/鉴权/录制控制),只是 offer/answer
的对端从"另一个浏览器"变成"SFU"。架构不变,值钱的正是这层信令。
3. 真正的录制要靠 SFU 或专门录制端:服务端混流录制(把多路视频合成一个 MP4 存档,供病历归档/复盘)是 SFU 的能力(如
MediaSoup 配 FFmpeg、LiveKit Egress)。本篇做的是录制的控制与审计信令,把录下来的文件落地归档,正好可以接你前面那篇
DICOM/大文件网关 的存储与分发能力。
4. 生产必须 WSS + 正规鉴权:演示用静态 token 和明文 ws。医疗合规要求 TLS(WSS)、JWT/OAuth 鉴权、会诊权限校验、TURN
服务器(很多医院内网 NAT/防火墙严格,只靠 STUN 连不通,必须自建 coturn 做中继)。
---
到这里,你这套系列已经覆盖得相当全了——从金融六篇(撮合/留痕/网关/对账/国密/分布式)到行业应用三篇(充电桩/DICOM
影像/会诊信令)。它们共享同一套 Swoole 6 工程内核:协程 + 共享内存状态 + io_uring + 长连接管理。
要不要我帮你做一个总收尾:把这 9
篇整理成一张技术能力矩阵——每篇用到的核心模式(协程/Table/io_uring/WebSocket/背压/状态机…)、真实瓶颈、以及"原生可/
需配专业组件(SFU、DCMTK、国密库、etcd)"的边界一览表?这样你无论是写成专栏、做技术分享还是对外讲方案,都有一张能"镇场子"
且经得起追问的全局图。
更多推荐


所有评论(0)