在 Laravel 中使用 PHPSocketIO 提供服务

参考测试文章，https://www.workerman.net/a/1621 可以了解到通过 laraman 与 phpsocketio 结合的使用方式。接下来便可以通过自定义进程为项目提供即时通讯服务。下方是部分 demo 代码。

config/socketio.php

<?php

use Itinysun\Laraman\Events\MessageDone;
use Itinysun\Laraman\Events\MessageReceived;
use Itinysun\Laraman\Events\RequestReceived;
use Itinysun\Laraman\Events\TaskReceived;
use Itinysun\Laraman\Listeners\CleanBaseState;
use Itinysun\Laraman\Listeners\CleanWebState;
use Itinysun\Laraman\Process\Web;

/*
 * 这里是进程配置文件
 * 配置文件格式：
 * 'process name'=>[
 *      'handler'=>class name, 必须，写进程类名。必须继承 ProcessBase 类
 *      'options'= [ 进程类构造函数的参数，用于进程内部使用
 *          'events'=>[event name =>array(listeners)] 可选，订阅事件。
 *          'clearMode'=>bool , 可选，默认false，是否开启洁癖模式
 *       ],
 *      'workerman'=>array config 可选，构造 worker时的参数，参考workerman官方手册，如果是全局属性，请在server中配置

 * ]
 *
 */
return [
    'workerman' => [
        'count' => 1,

        //是否支持平滑重启，如果是持续性任务，请设置为false
        'reloadable' => true,

        //运行命令使用的用户及组，windows无效
        'user' => '',
        'group' => '',
    ],
    'handler' => \App\Process\SocketIO::class,
];

app/Process/SocketIO.php

<?php

namespace App\Process;

use Workerman\Timer;
use Workerman\Worker;
use Workerman\Protocols\Http\Request;
use Workerman\Connection\TcpConnection;
use Itinysun\Laraman\Process\ProcessBase;
use Itinysun\Laraman\Server\LaramanWorker;
use Itinysun\Laraman\Traits\HasLaravelApplication;

class SocketIO extends ProcessBase
{
    use HasLaravelApplication;

    protected function onWorkerStart(LaramanWorker $worker): void
    {
        // 全局数组保存uid在线数据
        $uidConnectionMap = array();
        // 记录最后一次广播的在线用户数
        $last_online_count = 0;
        // 记录最后一次广播的在线页面数
        $last_online_page_count = 0;

        // PHPSocketIO服务
        $sender_io = new \PHPSocketIO\SocketIO(2120);

        // 客户端发起连接事件时，设置连接socket的各种事件回调
        $sender_io->on('connection', function ($sender_io) use (&$uidConnectionMap, $last_online_count, $last_online_page_count) {
            // 当客户端发来登录事件时触发
            $sender_io->on('login', function ($uid) use ($sender_io, &$uidConnectionMap, $last_online_count, $last_online_page_count) {
                // 已经登录过了
                if (isset($sender_io->uid)) {
                    return;
                }
                // 更新对应uid的在线数据
                $uid = (string)$uid;
                if (!isset($uidConnectionMap[$uid])) {
                    $uidConnectionMap[$uid] = 0;
                }
                // 这个uid有++$uidConnectionMap[$uid]个socket连接
                ++$uidConnectionMap[$uid];
                // 将这个连接加入到uid分组，方便针对uid推送数据
                $sender_io->join($uid);
                $sender_io->uid = $uid;
                // 更新这个socket对应页面的在线数据
                $sender_io->emit('update_online_count', "当前<b>{$last_online_count}</b>人在线，共打开<b>{$last_online_page_count}</b>个页面");
            });

            // 当客户端断开连接是触发（一般是关闭网页或者跳转刷新导致）
            $sender_io->on('disconnect', function () use ($sender_io, &$uidConnectionMap) {
                if (!isset($sender_io->uid)) {
                    return;
                }
                if (empty($uidConnectionMap[$sender_io->uid])) {
                    return;
                }
                // 将uid的在线socket数减一
                if (--$uidConnectionMap[$sender_io->uid] <= 0) {
                    unset($uidConnectionMap[$sender_io->uid]);
                }
            });
        });

        // 监听一个http端口
        $inner_http_worker = new Worker('http://0.0.0.0:2121');
        // 当http客户端发来数据时触发
        $inner_http_worker->onMessage = function (TcpConnection $http_connection, Request $request) use ($sender_io, &$uidConnectionMap) {
            $response = new \Workerman\Protocols\Http\Response();
            $response->withHeaders([
                'Access-Control-Allow-Credentials' => 'true',
                'Access-Control-Allow-Origin' => $request->header('origin', '*'),
                'Access-Control-Allow-Methods' => $request->header('access-control-request-method', '*'),
                'Access-Control-Allow-Headers' => $request->header('access-control-request-headers', '*'),
                'Content-Type' => 'application/json',
            ]);

            $post = $request->post();
            $post = $post ? $post : $request->get();
            // 推送数据的url格式 type=publish&to=uid&content=xxxx
            switch (@$post['type']) {
                case 'publish':
                    $to = @$post['to'];
                    $post['content'] = htmlspecialchars(@$post['content']);
                    // 有指定uid则向uid所在socket组发送数据
                    if ($to) {
                        $sender_io->to($to)->emit('new_msg', $post['content']);
                        // 否则向所有uid推送数据
                    } else {
                        $sender_io->emit('new_msg', @$post['content']);
                    }
                    // http接口返回，如果用户离线socket返回fail
                    if ($to && !isset($uidConnectionMap[$to])) {
                        $response->withBody('offline');
                        return $http_connection->send($response);
                    } else {
                        $response->withBody('ok');
                        return $http_connection->send($response);
                    }
            }
            $response->withBody('fail');
            return $http_connection->send($response);
        };
        // 执行监听
        $inner_http_worker->listen();
        $sender_io->worker->listen();

        // 一个定时器，定时向所有uid推送当前uid在线数及在线页面数
        Timer::add(1, function () use (&$uidConnectionMap, $sender_io, &$last_online_count, &$last_online_page_count) {
            $online_count_now = count($uidConnectionMap);
            $online_page_count_now = array_sum($uidConnectionMap);
            // 只有在客户端在线数变化了才广播，减少不必要的客户端通讯
            if ($last_online_count != $online_count_now || $last_online_page_count != $online_page_count_now) {
                $sender_io->emit('update_online_count', "当前<b>{$online_count_now}</b>人在线，共打开<b>{$online_page_count_now}</b>个页面");
                $last_online_count = $online_count_now;
                $last_online_page_count = $online_page_count_now;
            }
        });
    }
}