thinkphp8.0用workerman根據(jù)用戶userid主動推送給多個(gè)客戶端（包含一對一聊天）
效果圖片：
前端測試代碼（按F12）：
測試鏈接：http://test.com/?userid=1000
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>WebSocket Example</title>
</head>
<body>
<p>用戶：<input type="text" id="userToId" value="1000"></p>
<p>內(nèi)容：<input type="text" id="msg"></p>
<p><button id="sendMessageButton">發(fā)送消息</button></p>
<script>
function getUrlParams() {
const url = window.location.search
const paramsRegex = /[?&]+([^=&]+)=([^&]*)/gi;
const params = {};
let match;
while (match = paramsRegex.exec(url)) {
params[match[1]] = match[2];
}
return params;
}
// 創(chuàng)建 WebSocket 連接
const ws = new WebSocket('ws://localhost:8481');
let data = getUrlParams()
// 連接打開時(shí)觸發(fā)
ws.onopen = function() {
console.log('已連接到服務(wù)器.',data.userid);
ws.send(JSON.stringify({
type: 'bind',
userId: data.userid,
}));
// 間隔30秒通知一次服務(wù)器我還在線
setInterval(() => {
ws.send(JSON.stringify({
type: 'ping'
}));
},50000);
};
// 收到服務(wù)器消息時(shí)觸發(fā)
ws.onmessage = function(event) {
console.log(event.data);
// 可以在這里添加更多的處理邏輯，比如更新頁面內(nèi)容
// alert('Received from server: ' + event.data);
};
// 連接關(guān)閉時(shí)觸發(fā)
ws.onclose = function() {
console.log('已斷開與服務(wù)器的連接');
};
// 連接出錯(cuò)時(shí)觸發(fā)
ws.onerror = function(error) {
console.error('WebSocket Error:', error);
};
// 獲取按鈕元素
const sendMessageButton = document.getElementById('sendMessageButton');
const msg = document.getElementById('msg');
const userToId = document.getElementById('userToId');
// 為按鈕添加點(diǎn)擊事件監(jiān)聽器
sendMessageButton.addEventListener('click', function() {
// 檢查 WebSocket 連接是否已打開
if (ws.readyState === WebSocket.OPEN) {
// 發(fā)送消息給服務(wù)器'發(fā)送：' + msg.value
ws.send(JSON.stringify({
type: 'text',
userToId: userToId.value,
message: msg.value,
}));
// console.log('Message sent to server: Hello Server!');
} else {
console.error('WebSocket未打開。就緒狀態(tài):', ws.readyState);
}
});
</script>
</body>
</html>
后端發(fā)送代碼：
測試鏈接：http://test.com/tests/t
```php
public function t()
{
// 建立socket連接到內(nèi)部推送端口
$client = stream_socket_client('tcp://127.0.0.1:5678', $errno, $errmsg, 1);
// 推送的數(shù)據(jù)，包含uid字段，表示是給這個(gè)uid推送
$data = [
'type' => 'text',
'userToId' => '1000',
'message' => '服務(wù)器端TTT的通知消息'.rand(10,99)
];
// 發(fā)送數(shù)據(jù)，注意5678端口是Text協(xié)議的端口，Text協(xié)議需要在數(shù)據(jù)末尾加上換行符
fwrite($client, json_encode($data)."\n");
// 讀取推送結(jié)果
echo fread($client, 8192);
return 11;
}運(yùn)行：php think websocket:server
```php
<?php
//文件路徑： app\command\WebSocketServer.php
namespace app\command;
use think\console\Command;
use think\console\Input;
use think\console\Output;
use Workerman\Connection\TcpConnection;
use Workerman\Timer;
use Workerman\Worker;
/*
* 配置信息 路徑： \config\console.php
* return [
// 指令定義
'commands' => [
'websocket:server' => app\command\WebSocketServer::class,
],
];
*/
class WebSocketServer extends Command
{
protected function configure(): void
{
$this->setName('websocket:server')->setDescription('Start WebSocket Server');
}
protected function execute(Input $input, Output $output): void
{
// 心跳間隔55秒
define('HEARTBEAT_TIME', 55);
$worker = new Worker('websocket://0.0.0.0:8481');
$worker->count = 1;
$worker->connectionList = [];
// 進(jìn)程啟動后設(shè)置一個(gè)每10秒運(yùn)行一次的定時(shí)器
$worker->onWorkerStart = function ($worker) use ($output){
Timer::add(10, function()use($worker){
$time_now = time();
foreach($worker->connections as $connection) {
// 有可能該connection還沒收到過消息，則lastMessageTime設(shè)置為當(dāng)前時(shí)間
if (empty($connection->lastMessageTime)) {
$connection->lastMessageTime = $time_now;
continue;
}
// 上次通訊時(shí)間間隔大于心跳間隔，則認(rèn)為客戶端已經(jīng)下線，關(guān)閉連接
if ($time_now - $connection->lastMessageTime > HEARTBEAT_TIME) {
$connection->send('您長時(shí)間未請求服務(wù)器，鏈接已經(jīng)斷開');
$connection->close();
}
}
});
// 開啟一個(gè)內(nèi)部端口，方便內(nèi)部系統(tǒng)推送數(shù)據(jù)，Text協(xié)議格式 文本+換行符
$inner_text_worker = new Worker('Text://0.0.0.0:5678');
$inner_text_worker->onMessage = function($connection, $buffer) use ($worker)
{
// $data數(shù)組格式，里面有uid，表示向那個(gè)uid的頁面推送數(shù)據(jù)
$data = json_decode($buffer, true);
//聊天
if ($data['type'] == 'text')
{
if (isset($worker->connectionList[$data['userToId']]))
{
$conns = $worker->connectionList[$data['userToId']];
foreach ($conns as $conn){
$conn->send('用戶'.$data['userToId'].'：'.$data['message']);
}
$connection->send('我：成功了');
}else{
$connection->send('我：對方下線了哈');
}
}
};
$inner_text_worker->listen();
// $output->writeln('用戶onWorkerStart' );
};
$worker->onConnect = function(TcpConnection $connection) use ($output, $worker)
{
// 指令輸出
$output->writeln('用戶：'.$worker->id.'-'.$connection->id.'鏈接成功');
};
$worker->onMessage = function ($connection, $data) use ($output, $worker) {
// 給connection臨時(shí)設(shè)置一個(gè)lastMessageTime屬性，用來記錄上次收到消息的時(shí)間
$connection->lastMessageTime = time();
$data = json_decode($data, true);
//綁定用戶ID
if ($data['type'] == 'bind' && !isset($connection->userId))
{
$connection->userId = $data['userId'];
$worker->connectionList[$connection->userId][$connection->id] = $connection;
}
//聊天
if ($data['type'] == 'text')
{
if (isset($worker->connectionList[$data['userToId']]))
{
$conns = $worker->connectionList[$data['userToId']];
foreach ($conns as $conn){
$conn->send('用戶'.$data['userToId'].'：'.$data['message']);
}
}else{
$connection->send('我：對方下線了');
}
}
// // 指令輸出
// $output->writeln('用戶：'.$connection->id.'，客戶端接收到數(shù)據(jù)：'.json_encode($worker->connections) );
//向客戶端自己發(fā)送數(shù)據(jù)
if (!empty($data['message'])){
$connection->send('我：'.$data['message']);
}
};
// 客戶端連接斷開時(shí)，斷開對應(yīng)的鏈接連接
$worker->onClose = function(TcpConnection $connection) use ($output, $worker)
{
if(isset($worker->connectionList[$connection->userId]))
{
// 連接斷開時(shí)刪除映射
if (count($worker->connectionList[$connection->userId]) == 1){
unset($worker->connectionList[$connection->userId]);
}else{
unset($worker->connectionList[$connection->userId][$connection->id]);
}
}
// 指令輸出
$output->writeln('用戶：'.$connection->id.'已經(jīng)斷開鏈接。'.$connection->userId);
};
Worker::runAll();
}
}0個(gè)評論
haomc
320
積分
0
獲贊數(shù)
0
粉絲數(shù)
2025-02-18 加入