当前位置：网站首页>yii2编写swoole的websocket服务

yii2编写swoole的websocket服务

2022-06-28 13:16:00 【赫赫phper】

一、查看PHP的swoole扩展

打开phpinfo,搜索，没有安装的需要安装下

二、开始编写swoole服务代码：

问件目录：/console/controllers/WebSocketController.php

遇到问题1: 无法实现根据商户的uid给特定的商户推送消息，只能通过连接的用户进行所有用户推送

解决方案：1、客户端连接websocket服务的时候，传商户的uid,服务端client接收商户的uid，通过redis将商户的uid作为key，拼接固定前缀，客户端连接的fd为值，进行存储；
2、主动推送server连接服务端client主动推送消息的时候，json串中传要推送的商户的uid；
3、服务端message收到消息后，解析json传中传递的商户uid，通过uid的key值获取redis中存储的商户连接的fd值，通过fd值给商户推送消息；
4、解决根据商户uid给特定商户推送特定消息的问题；

namespace console\controllers;

use \Swoole\WebSocket\Server;
use Yii;
use yii\console\Controller;
use yii\helpers\Json;
class WebSocketController extends Controller
{
    /**
     * @var Server
     */
    public $serv;

    public function actionRun()
    {
        //配置ssl加密连接 , SWOOLE_PROCESS, SWOOLE_SOCK_TCP|SWOOLE_SSL
        $this->serv = new Server("0.0.0.0", 9503);
        $this->serv->set([
            'worker_num'               => 10, //开启的进程数 ⼀般为cup核数 1-4 倍
            'task_worker_num'          => 10, //配置 Task 进程的数量。
            'daemonize'                => 1, //设置 daemonize => true 时，程序将转入后台作为守护进程运行。长时间运行的服务器端程序必须启用此项。如果不启用守护进程，当 ssh 终端退出后，程序将被终止运行。
            'max_request'              => 10000, //设置 worker 进程的最大任务数。
            'dispatch_mode'            => 5, //uid dispatch UID 分配 需要用户代码中调用 Server->bind() 将一个连接绑定 1 个 uid。然后底层根据 UID 的值分配到不同的 Worker 进程。
            'open_eof_check'           => true,
            'package_eof'              => PHP_EOL,
            'log_file'                 => Yii::getAlias('@runtime/swoole_below_cps.log'),
            'reload_async'             => true, //开启的主要目的是为了保证服务重载时，协程或异步任务能正常结束。
            //'enable_reuse_port'        => true, //启用端口重用后，可以重复启动监听同一个端口的 Server 程序
            'heartbeat_idle_time'      => 600,
            'heartbeat_check_interval' => 60,
//            'ssl_cert_file'            => EnvHelper::get('SSL_PEM'), //ssl加密配置文件
//            'ssl_key_file'             => EnvHelper::get('SSL_KEY'), //ssl加密配置key文件
        ]);

        $this->serv->on('Start', [$this, 'onStart']);
        $this->serv->on('Open', [$this, 'onOpen']);
        $this->serv->on('Message', [$this, 'onMessage']);
        $this->serv->on('Request', [$this, 'onRequest']);
        $this->serv->on('Receive', [$this, 'onReceive']);
        $this->serv->on('Task', [$this, 'onTask']);
        $this->serv->on('Finish', [$this, 'onFinish']);

        //dd('先不启');
        $this->serv->start();
    }

    public function onStart(Server $server)
    {
        static::log('平台Websocket启动'.$server->worker_id . PHP_EOL);
    }

    /**
     * Param: 当 WebSocket 客户端与服务器建立连接并完成握手后会回调此函数。
     * User: 赫赫
     * Date: 2022/6/13
     * @param Server $server
     * @param $request
     * @return void
     */
    public function onOpen(Server $server, $request) {
        //接收uid
        $uid = $request->get['uid'];
        //$type = $request->get['type'];
        static::log("活平台-建立连接: 用户uid= {$uid}, fd={$request->fd}建立连接\n");
        if(!empty($uid)){
            //将接收到的用户ID存储在redis中
            $redisKey   = "BELOWCPSWEBSOCKET:".$uid;
            $redis      = \Yii::$app->redis;
            $res = $redis->set($redisKey,$request->fd);
            static::log("平台-将接收到的用户ID存储在redis中:$res\n");
        }
    }

    /** 当服务器收到来自客户端的数据帧时会回调此函数。
     * Param:
     * User: 赫赫
     * Date: 2022/6/13
     * @param Server $server
     * @param $frame
     * @return void
     */
    public function onMessage(Server $server, $frame)
    {
        static::log('websocket服务收到老铁发的消息:' . $frame->data);
        $data = json_decode($frame->data, true);
        static::log('websocket服务收到老铁发的消息解密:' . $data['uid']."\n");
        //根据uid获取redis中的fd,根据fd给用户推送消息
        if(!empty($data['uid'])){
            //获取redis中的用户fd
            $redisKey   = "BELOWCPSWEBSOCKET:".$data['uid'];
            $redis      = \Yii::$app->redis;
            $u_fd = $redis->get($redisKey);
            static::log("websocket服务获取老铁连接fd={$u_fd}发消息\n");
            if($server->isEstablished($u_fd)){
                $server->push($u_fd, $data['msg']);
            }
        }
//        else{
//            //没有uid
//            foreach ($server->connections as $fd) {
//                if ($server->isEstablished($fd)) {
//                    $server->push($fd, $data['msg']);
//                }
//            }
//        }
    }

    /**
     * Param: 接收http请求从get获取message参数的值，给用户推送
     * User: 赫赫
     * Date: 2022/6/13
     * @param Server $serv
     * @param $response
     * @return void
     */
    public function onRequest(Server $serv, $response){
        $uid = $serv->get['uid'];
        static::log('websocket服务收到:'.$uid);
        foreach ($this->serv->connections as $fd) {
            // 需要先判断是否是正确的websocket连接，否则有可能会push失败
            if ($this->serv->isEstablished($fd)) {
                $this->serv->push($fd, $serv->get['message']);
            }
        }
    }
    public function onReceive(Server $serv, $fd, $reactor_id, $data){
        $conn = $serv->connection_info($fd);

        echo "worker_id: " . $serv->worker_id . PHP_EOL;
        if (empty($conn['uid'])) {
            $uid = $fd + 1;
            if ($serv->bind($fd, $uid)) {
                $serv->send($fd, "bind {$uid} success");
            }
        } else {
            if (!isset($serv->fdlist[$fd])) {
                $serv->fdlist[$fd] = $conn['uid'];
            }
            print_r($serv->fdlist);
            foreach ($serv->fdlist as $_fd => $uid) {
                $serv->send($_fd, "{$fd} say:" . $data);
            }
        }
    }
    /**
     * 处理任务
     * @param \Swoole\Server $serv
     * @param int $task_id
     * @param int $from_id
     * @param string $data
     */
    public function onTask(Server $serv, int $task_id, int $from_id, string $data)
    {
        static::log('收到新的任务');
        $data = Json::decode($data, true);
        $serv->finish($data);
    }
    /**
     * @param Server $server
     * @param int $task_id
     * @param mixed $data
     */
    public function onFinish(Server $server, int $task_id, $data)
    {
        static::log('任务完成');
    }

    /**
     * @param mixed $message
     */
    private static function log($message)
    {
        $str = "[" . date('Y-m-d H:i:s') . '] ';

        if (is_string($message)) {
            $str .= $message;
        }
        if (is_array($message)) {
            $str .= var_export($message, true);
        }
        $str .= PHP_EOL;
       // echo $str;
        \Yii::$app->byfLog->channel('BelowCpsWebSocket')->info($str);
    }

}