Swoole-WebSocket服务端主动推送消息

Swoole-WebSocket服务端主动推送消息

标签（空格分隔）： websocket swoole php

WebSocket服务器代码

<?php
namespace swoolewebsocket;

class WebSocketServer
{

    var $server;

    public function __construct()
    {
        $this->server = new SwooleWebSocketServer("0.0.0.0", 9991);
        print "WebSocketServer: 0.0.0.0, 9991" . PHP_EOL;
        $this->server->set([
            'task_worker_num' =>4
        ]);
        $this->server->on('open', [$this, 'open']);
        $this->server->on('message', [$this, 'message']);
        $this->server->on('task', [$this, 'task']);
        $this->server->on('finish', [$this, 'finish']);
        $this->server->on('request', [$this, 'request']);
        $this->server->on('close', [$this, 'close']);
        $this->server->start();
    }


    /**
     * $request 是一个 HTTP 请求对象，包含了客户端发来的握手请求信息
     * onOpen 事件函数中可以调用 push 向客户端发送数据或者调用 close 关闭连接
     * onOpen 事件回调是可选的
     * @param SwooleWebSocketServer $server
     * @param SwooleHttpRequest $request
     */
    public function open($server, $request)
    {
        $fd = $request->fd;
        echo "server: handshake success with fd{$fd}" . PHP_EOL;
    }


    /**
     * $frame 是 SwooleWebSocketFrame 对象，包含了客户端发来的数据帧信息
     * onMessage 回调必须被设置，未设置服务器将无法启动
     * 客户端发送的 ping 帧不会触发 onMessage，底层会自动回复 pong 包，也可设置 open_websocket_ping_frame 参数手动处理
     * @param SwooleWebSocketServer $server
     * @param SwooleWebSocketFrame $frame
     */
    public function message($server, $frame)
    {
        $data = json_decode($frame->data, true);
        var_dump("---收到消息----fd:{$frame->fd}");
        var_dump($data);
        $server->task($data);

        // 保持连接 心跳检测
        if ($data['type'] == 'ping') {
            $server->push($frame->fd, json_encode(['type' => 'ping', 'data' => 'tong']));
        }
    }

    /**
     * @param SwooleWebSocketServer $server
     * @param $fd
     */
    public function close($server, $fd)
    {
        echo "fd:{$fd} close" . PHP_EOL;
    }


    /**
     * @param SwooleServer $server
     * @param int $taskId
     * @param int $srcWorkerId
     * @param mixed $data
     */
    public function task($server, $taskId, $srcWorkerId, $data)
    {
        echo "正在处理任务----taskId:{$taskId}" . PHP_EOL;
        sleep("3");
        $server->finish("ok");
    }


    /**
     * @param SwooleServer $server
     * @param int $task_id
     * @param mixed $data
     */
    public function finish($server, $taskId, $data)
    {
        echo "任务处理成功-----taskId:{$taskId}";
    }


    public function request(SwooleHttpRequest $request, SwooleHttpResponse $response)
    {
        echo "服务端主动推送消息......" . PHP_EOL;
        if ($request->server['path_info'] == "/websocket/") {
            foreach ($this->server->connections as $fd) {
                if ($this->server->isEstablished($fd)) {
                    $this->server->push($fd, json_encode(['type' => 3, 'data' => '系统消息推送' . mt_rand(1000, 9999)]));
                }
            }
        }
    }
}

new WebSocketServer();

客户端(JS)

<script>
document.addEventListener("DOMContentLoaded", function() {
    var wsServer = 'ws://web.test.com/wss';
    var websocket = new WebSocket(wsServer);
    websocket.onopen = function (evt) {
        console.log("已连接上服务器");
    };

    //每隔5秒钟发送一次心跳，避免websocket连接因超时而自动断开
    window.setInterval(function(){
        var ping = {"type":"ping"};
        websocket.send(JSON.stringify(ping));
    },10000);

    websocket.onclose = function (evt) {
        console.log("连接已关闭");
    };

    websocket.onmessage = function (evt) {
        data = eval("("+evt.data+")");
        if (data.type == "ping") {
            return;
        }
        console.log('接受信息: ' + data.data);
    };
});

客户端(php)

$result = curl_get("http://0.0.0.0:9991/websocket/");