thinkPHP搭建webSocket服务端和客户端

仅供参考,提供的都是代码片段
这里的场景为通过PHP发送webSocket推送通知消息给webSocket服务端,服务端在把消息推送给web客户端

搭建webSocket服务端,在thinkPHP的application文件夹下创建一个worker/controller/Worket.php文件:

<?php
namespace app\worker\controller;
use think\worker\Server;
use Workerman\Lib\Timer;

class Worker extends Server
{
    protected $socket = 'websocket://127.0.0.1:2346';
    protected $processes = 1;
    protected $worker; 
    protected $HEARTBEAT_TIME = '60';

    public function __construct()
    {
        parent::__construct();
        $this->worker->uidConnections = array();
    }

    public function onMessage($connection, $data)
    {
        $uid = $connection->uid;
        if($uid) {
            // 【关键修改1】：在这里接收 PHP 后端推送来的订单数据
            $arr = json_decode($data, true);
            if(isset($arr['order_type'])){
                // 收到推送，查找管理员并分发
                if(!empty($arr['store_id'])){
                    $pushAuth = model('admin')->field('id,username')->where("find_in_set(:store_ids, store_ids)", ['store_ids' => $arr['store_id']])->whereOr('store_ids','')->select();
                }else{
                    $pushAuth = model('admin')->field('id,username')->where('store_ids','')->select();
                }
                foreach ($pushAuth as $value) {
                    $this->sendMessageByUid($value['id'], $arr);
                }
                print_r("收到后端推送并已转发\n");
                return;
            }

            print_r("收到前端消息: ".$data."\n");
            // 下面你原来的业务逻辑保持不变...
            $data=json_decode($data,true);
            $message=['code'=>0,'msg'=>'你还活着','type'=>0,'data'=>[]];
            // 【关键修改2】：去掉 json_encode 的第二个参数 true！否则会乱码
            $message = json_encode($message); 
            if(!$data){
                return true;
            }
            if(isset($data['uuid'])){
                $this->worker->uidConnections[$data['uuid']]=$connection;
            }
            $method = 'send';
            $is_push=0;
            $this->broadcast($message);
            switch ($method) {
                case 'add_chat':
                    $is_push=1;
                    $chat_id=1111;
                    $message=['code'=>0,'msg'=>'创建成功','type'=>-1,'data'=>$chat_id];
                    break;
                case 'send':
                    $is_push=1;
                    $message='你处理的消息信息';
                    break;
                case 'list':    
                    $is_push=1;
                    $message='聊天消息列表';
                    break;
                case 'chat_list':
                    $is_push=1;
                    $post=['user_id'=>$uid,'chat_id'=>$data['chat_id'],'page'=>$data['page']];
                    $message=ChatList::getList($post);
                    break;
            }
            if($is_push){
                $this->sendMessageByUid($uid,$message);
            }
        }else{
            $this->onClose($connection);
        }
    }

    function broadcast($message)
    {
        foreach($this->worker->uidConnections as $connection)
        {
            $connection->send($message);
        }
    }

    function sendMessageByUid($uid, $message)
    {
        if(isset($this->worker->uidConnections[$uid]))
        {
            $connection = $this->worker->uidConnections[$uid];
            // 【关键修改3】：去掉 json_encode 的 true
            $connection->send(json_encode($message)); 
        }
    }

    public function onConnect($connection)
    {
        $connection->onWebSocketConnect = function($connection , $http_header)
        {
            // 【关键修改4】：删掉你原来这里截取最后一行解析 order_type 的逻辑
            // 握手阶段只管握手，业务数据已经转移到 onMessage 里完美接收了
            
            if(!isset($connection->uid))
            {
                $connection->uid = time();
                $this->worker->uidConnections[$connection->uid]=$connection;
            }
            print_r("用户连接成功".$connection->uid."\n");
        };
    }

    public function onClose($connection)
    {
        if(isset($connection->uid))
        {
            print_r("用户断开".$connection->uid."\n");
            unset($this->worker->uidConnections[$connection->uid]);
        }
    }

    public function onError($connection, $code, $msg)
    {
        echo "error $code $msg\n";
    }

    public function onWorkerStart($worker)
    {
        // 你的心跳逻辑原封不动保留，保证不会主动断开客户端
        Timer::add(10, function()use($worker){
            $time_now = time();
            foreach($worker->connections as $connection) {
                if (empty($connection->lastMessageTime)) {
                    $connection->lastMessageTime = $time_now;
                    continue;
                }
                $connection->send('ping');
                if ($time_now - $connection->lastMessageTime > $this->HEARTBEAT_TIME) {
                    // 保持注释状态，不断开
                    // $connection->close();
                }
            }
        });
    }
}

<?php
define('APP_PATH', __DIR__ . '/../application/');
define('BIND_MODULE','worker/Worker');
// 加载框架引导文件
require __DIR__ . '/../thinkphp/start.php';

location ~* (runtime|application)/{
	return 403;
}
location / {
	if (!-e $request_filename){
		rewrite  ^(.*)$  /index.php?s=$1  last;   break;
	}
}
location /wss
    {
            proxy_pass http://127.0.0.1:2346;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header Host $host;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_http_version 1.1;
            proxy_set_header Upgrade $http_upgrade;
            proxy_set_header Connection "upgrade";
            rewrite /wss/(.*) /$1 break;
            proxy_redirect off;
    }

map $http_upgrade $connection_upgrade {
  default upgrade;
  ''      close;
}
# 这里的backend可以改成自己需要的名称，我自己用的是wss。
upstream wss {
  # 这里的localhost是映射本地服务器，也可以是外网ip，2345是我ws开启的端口。
  server localhost:2346;
}

server
{
    listen 80 default_server;
		listen 443 ssl http2 default_server;
    server_name xxxxxxxxxxx.cn www.xxxxxxxxxxx.cn;
    index index.php index.html index.htm default.php default.htm default.html;
    root /www/wwwroot/xxxxxxxxxxx.cn/hotel/public;

    #SSL-START SSL相关配置，请勿删除或修改下一行带注释的404规则
    #error_page 404/404.html;
    ssl_certificate    /www/server/panel/vhost/cert/xxxxxxxxxxx.cn/fullchain.pem;
    ssl_certificate_key    /www/server/panel/vhost/cert/xxxxxxxxxxx.cn/privkey.pem;
    ssl_protocols TLSv1.1 TLSv1.2 TLSv1.3;
    ssl_ciphers EECDH+CHACHA20:EECDH+CHACHA20-draft:EECDH+AES128:RSA+AES128:EECDH+AES256:RSA+AES256:EECDH+3DES:RSA+3DES:!MD5;
    ssl_prefer_server_ciphers on;
    ssl_session_cache shared:SSL:10m;
    ssl_session_timeout 10m;
    add_header Strict-Transport-Security "max-age=31536000";
    error_page 497  https://$host$request_uri;
		#SSL-END

    #ERROR-PAGE-START  错误页配置，可以注释、删除或修改
    #error_page 404 /404.html;
    #error_page 502 /502.html;
    #ERROR-PAGE-END

    #PHP-INFO-START  PHP引用配置，可以注释或修改
    

	include enable-php-74.conf;
    #PHP-INFO-END

    #REWRITE-START URL重写规则引用,修改后将导致面板设置的伪静态规则失效
    include /www/server/panel/vhost/rewrite/laiyashiji.cn.conf;
    #REWRITE-END

    #禁止访问的文件或目录
    location ~ ^/(\.user.ini|\.htaccess|\.git|\.env|\.svn|\.project|LICENSE|README.md)
    {
        return 404;
    }

    #一键申请SSL证书验证目录相关设置
    location ~ \.well-known{
        allow all;
    }

    #禁止在证书验证目录放入敏感文件
    if ( $uri ~ "^/\.well-known/.*\.(php|jsp|py|js|css|lua|ts|go|zip|tar\.gz|rar|7z|sql|bak)$" ) {
        return 403;
    }

    location ~ .*\.(gif|jpg|jpeg|png|bmp|swf)$
    {
        expires      30d;
        error_log /dev/null;
        access_log /dev/null;
    }
    location ~ .*\.(js|css)?$
    {
        expires      12h;
        error_log /dev/null;
        access_log /dev/null;
    }
	access_log  /www/wwwlogs/xxxxxxxxxxx.cn.log;
    error_log  /www/wwwlogs/xxxxxxxxxxx.cn.error.log;
}

/bin/bash -c "kill -9 $(lsof -t -i :2346); php server.php start;"

<?php
namespace app\common\library;
use think\Log;

class WebSocketPush
{
    public static function push($order_type=1, $status = 1, $order_id=0, $store_id=0, $msg='您有新订单消息!'){
        
        $host = "127.0.0.1";
        $port = "2346";
        $path = "/wss"; 
        
        // 组装消息
        if($order_type==1){
            $msg = '您有新酒店订单消息';
        }else if($order_type==2){
            $msg = '您有储值卡充值订单消息';
        }else if($order_type==3){
            $msg = '您有购买会员订单消息';
        }else{
            $msg = '您有商城订单消息';
        }

        $arr = [
            'order_type' => $order_type,
            'status'     => $status,
            'msg'        => $msg,
            'order_id'   => $order_id,
            'store_id'   => $store_id,
            'data'       => ['name'=>111]
        ];
        $message = json_encode($arr); 

        // 1. 创建 TCP Socket
        $socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
        if(!$socket){
            Log::error("WebSocket Push: 创建socket失败");
            return false;
        }
        
        // 2. 连接服务端
        $result = @socket_connect($socket, $host, $port);
        if (!$result) {
            Log::error("WebSocket Push: 连接失败 - " . socket_strerror(socket_last_error($socket)));
            socket_close($socket);
            return false;
        }

        // 3. 发送 WebSocket 握手请求 (重点修复：Host 必须带端口)
        $key = base64_encode(random_bytes(16));
        $upgradeHeader = "GET $path HTTP/1.1\r\n" .
                    "Host: $host:$port\r\n" .  // <--- 这里必须改成 $host:$port
                    "Upgrade: websocket\r\n" .
                    "Connection: Upgrade\r\n" .
                    "Sec-WebSocket-Key: $key\r\n" .
                    "Sec-WebSocket-Version: 13\r\n\r\n";
                    
        socket_write($socket, $upgradeHeader, strlen($upgradeHeader));

        // 4. 读取服务端握手响应 (增强检测)
        $response = '';
        $timeout = 3; // 握手超时时间3秒
        $start_time = time();
        
        while (true) {
            // 检查是否超时
            if (time() - $start_time > $timeout) {
                Log::error("WebSocket Push: 握手等待超时");
                socket_close($socket);
                return false;
            }
            
            $chunk = @socket_read($socket, 1024);
            
            // socket_read 返回 false 或 空字符串，说明连接已断开
            if ($chunk === false || $chunk === '') {
                Log::error("WebSocket Push: 握手失败，服务端关闭了连接。可能原因: Host字段不对或端口不匹配");
                socket_close($socket);
                return false;
            }
            
            $response .= $chunk;
            
            // 检查是否收到了完整的HTTP头
            if (strpos($response, "\r\n\r\n") !== false) {
                break;
            }
        }
        
        // 检查是否真正返回了 101 状态码
        if (strpos($response, '101') === false) {
            Log::error("WebSocket Push: 握手被拒绝, 响应头: " . $response);
            socket_close($socket);
            return false;
        }

        // 5. 握手成功，封装 WebSocket 数据帧 (加 Mask 掩码)
        $frame = self::encodeWebSocketFrame($message);
        
        // 6. 发送数据帧
        $write_result = @socket_write($socket, $frame, strlen($frame));
        if ($write_result === false) {
             Log::error("WebSocket Push: 发送数据帧失败");
        }

        // 7. 短暂等待后关闭连接 (确保数据已发出)
        usleep(50000); // 等待 50 毫秒
        socket_close($socket);
        
        return true;
    }

    /**
     * 核心方法：按照 RFC6455 标准封装带 Mask 掩码的 WebSocket 数据帧
     */
         /**
     * 核心方法：按照 RFC6455 标准封装带 Mask 掩码的 WebSocket 数据帧
     */
    private static function encodeWebSocketFrame($payload) {
        $length = strlen($payload);
        
        // 第一个字节：0x81 表示 FIN=1, opcode=1 (文本帧)
        $frame = chr(0x81);
        
        // 第二个字节：客户端必须置 Mask 位为 1 (0x80)
        if ($length <= 125) {
            $frame .= chr(0x80 | $length);
        } elseif ($length <= 65535) {
            $frame .= chr(0x80 | 126) . pack("n", $length);
        } else {
            $frame .= chr(0x80 | 127) . pack("J", $length);
        }
        
        // 生成 4 字节的 Masking-key
        $mask = random_bytes(4);
        $frame .= $mask;
        
        // 对载荷数据进行异或计算 (严格对齐，防止错位)
        $maskedPayload = '';
        for ($i = 0; $i < $length; $i++) {
            // 注意：这里必须严格用 $i % 4 保证与 Mask 的 0,1,2,3 逐字节对齐
            $maskedPayload .= $payload[$i] ^ $mask[$i % 4];
        }
        $frame .= $maskedPayload;
        
        return $frame;
    }


}

 sendWebSocket();
            function sendWebSocket(){
                var lockReconnect = false;  //避免ws重复连接
                var ws = null;          // 判断当前浏览器是否支持WebSocket
                var wsUrl = 'wss://xxxxxxx.cn/wss';
                createWebSocket(wsUrl);   //连接ws
                Fast.config.toastr.positionClass = "toast-top-center";
                function createWebSocket(url) {
                    try{
                        if('WebSocket' in window){
                            ws = new WebSocket(url);
                        }
                        initEventHandle();
                    }catch(e){
                        reconnect(url);
                        console.log(e);
                    }     
                }
                
                function initEventHandle() {
                    ws.onclose = function () {
                        // Toastr.error('连接已断开');
                        reconnect(wsUrl);
                        setStatus()
                        console.log("llws连接关闭!"+new Date().toLocaleString());
                    };
                    ws.onerror = function () {
                        reconnect(wsUrl);
                        // Toastr.error('连接错误');
                        console.log("llws连接错误!");
                        setStatus()
                    };
                    ws.onopen = function () {
                        heartCheck.reset().start();      //心跳检测重置
                         ws.send(
                            JSON.stringify({
                                type: 'send',
                                uuid: Config.admin.id
                             })
                         );
                        console.log("llws连接成功!"+new Date().toLocaleString());
                    };
                    ws.onmessage = function (event) {    //如果获取到消息，心跳检测重置
                        heartCheck.reset().start();      //拿到任何消息都说明当前连接是正常的
                        var audio1 = new Audio('/audio/新订单.mp3');
                        var audio2 = new Audio('/audio/取消订单.mp3');
                        var audio3 = new Audio('/audio/申请取消订单.mp3');
                        setStatus(1)
                        // console.log("llws收到消息啦:" +event.data);
                        if(event.data!='ping'){
                            try {
                                let json = JSON.parse(decodeUnicode(event.data))
                                if(json['order_type']!=undefined){
                                    if(json['status']==1){
                                        Toastr.info(json.msg,'订单通知');
                                        audio1.play();
                                    }else{
                                            Toastr.warning(json.msg,'订单通知');
                                        //  Layer.alert(json.msg, {icon: 3, title:'通知'}, 
                                        //     function(index){
                                        //         //点击确定按钮的回调函数
                                        //         layer.close(index); //关闭弹窗
                                        //         //执行相关操作代码
                                        //     }
                                        // );
                                        audio2.play();
                                        
                                    }
                                }
                                
                                console.log(json)
                            } catch (error) {
                                // 异常处理
                                setStatus()
                                console.log(error.message); // 输出错误信息
                            } finally {
                                // 清理代码或执行必须发生的代码
                                // console.log('无论是否发生异常，finally块中的代码都会执行。');
                            }
                        }
                    };
                }
                function setStatus(status=0){
                    if(status){
                        $('.online-status').css('color','#18bc9c')
                        $('.online-status-text').html('在线')
                    }else{
                        $('.online-status').css('color','red')
                        $('.online-status-text').html('业务通知掉线')
                    }
                    
                }
                // 监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
                window.onbeforeunload = function() {
                    ws.close();
                }  
                
                function reconnect(url) {
                    if(lockReconnect) return;
                    lockReconnect = true;
                    setTimeout(function () {     //没连接上会一直重连，设置延迟避免请求过多
                        createWebSocket(url);
                        lockReconnect = false;
                    }, 2000);
                }
               function decodeUnicode(str) {
                str = str.replace(/\\/g, "%");  
                    return unescape(str);  
                }
                //心跳检测
                var heartCheck = {
                    timeout: 10000,        //1分钟发一次心跳
                    timeoutObj: null,
                    serverTimeoutObj: null,
                    reset: function(){
                        
                        clearTimeout(this.timeoutObj);
                        clearTimeout(this.serverTimeoutObj);
                        return this;
                    },
                    start: function(){
                        var self = this;
                        // console.log(this.timeoutObj )
                        this.timeoutObj = setTimeout(function(){
                            //这里发送一个心跳，后端收到后，返回一个心跳消息，
                            //onmessage拿到返回的心跳就说明连接正常
                            ws.send("ping");
                            // console.log("ping!")
                            self.serverTimeoutObj = setTimeout(function(){//如果超过一定时间还没重置，说明后端主动断开了
                                ws.close();     //如果onclose会执行reconnect，我们执行ws.close()就行了.如果直接执行reconnect 会触发onclose导致重连两次
                            }, self.timeout)
                        }, this.timeout)
                    }
                }
            }