Laravel整合PHPSocketIo实现web消息推送

PHPSocket.IO，PHP 跨平台实时通讯框架
PHPSocket.IO 是 PHP 版本的 Socket.IO 服务端实现，基于 workerman 开发，用于替换 node.js 版本 Socket.IO 服务端。PHPSocket.IO 底层采用 websocket 协议通讯，如果客户端不支持 websocket 协议，则会自动采用 http 长轮询的方式通讯。

环境

• Ubuntu 18
• Laravel 5.8
• PHPSocket.IO 1.1

安装依赖

composer require workerman/phpsocket.io
composer require guzzlehttp/guzzle

启动程序整合到 artisan 命令中

创建文件命令php artisan make:command MsgPush

app/Console/Commands/MsgPush.php

<?php

namespace App\Console\Commands;

use Illuminate\Console\Command;
use Workerman\Worker;
use Workerman\Lib\Timer;
use PHPSocketIO\SocketIO;

class MsgPush extends Command
{
    protected $signature = 'msg-push
    {action=start : start | restart | reload(平滑重启) | stop | status | connetions}
    {--d : deamon or debug}';
    
    protected $description = 'web 消息推送服务';
    
    // 全局数组保存 uid 在线数据
    private static $uidConnectionCounter = [];
    // 广播的在线用户数, 一个 uid 代表一个用户
    private static $onlineCount = 0;
    // 广播的在线页面数, 同一个 uid 可能开启多个页面
    private static $onlinePageCount = 0;
    //PHPSocketIO 服务
    private static $senderIo = null;
    

    public function __construct()
    {parent::__construct();
    }
    
    /**
     * 根据脚本参数开启 PHPSocketIO 服务
     * PHPSocketIO 服务的端口是 `2120`
     * 传递数据的端口是 `2121`
     */
    public function handle()
    {
        global $argv;
        // 启动 php 脚本所需的命令行参数
        $argv[0] = 'MsgPush';
        $argv[1] = $this->argument('action'); // start | restart | reload(平滑重启) | stop | status | connetions
        $argv[2] = $this->option('d') ? '-d' : ''; // 守护进程模式或调试模式启动
        
        // PHPSocketIO 服务
        self::$senderIo = new SocketIO(2120);
        
        // 客户端发起连接事件时，设置连接 socket 的各种事件回调
        self::$senderIo->on('connection', function ($socket) {
            
            // 当客户端发来登录事件时触发,$uid 目前由页面传值决定, 当然也可以根据业务需要由服务端来决定
            $socket->on('login', function ($uid) use ($socket) {
                // 已经登录过了
                if (isset($socket->uid)) return;
                
                // 更新对应 uid 的在线数据
                $uid = (string)$uid;
                // 这个 uid 有 self::$uidConnectionCounter[$uid]个 socket 连接
                self::$uidConnectionCounter[$uid] = isset(self::$uidConnectionCounter[$uid]) ? self::$uidConnectionCounter[$uid] + 1 : 1;
                
                // 将这个连接加入到 uid 分组，方便针对 uid 推送数据
                $socket->join($uid);
                $socket->uid = $uid;
                // 更新这个 socket 对应页面的在线数据
                self::emitOnlineCount();});
            
            // 当客户端断开连接是触发（一般是关闭网页或者跳转刷新导致）$socket->on('disconnect', function () use ($socket) {if (!isset($socket->uid)) {return;}
                
                // 将 uid 的在线 socket 数减一
                if (--self::$uidConnectionCounter[$socket->uid] <= 0) {unset(self::$uidConnectionCounter[$socket->uid]);
                }
            });
            
        });
        
        // 当 self::$senderIo 启动后监听一个 http 端口，通过这个端口可以给任意 uid 或者所有 uid 推送数据
        self::$senderIo->on('workerStart', function () {
            // 监听一个 http 端口
            $innerHttpWorker = new Worker('http://0.0.0.0:2121');
            // 当 http 客户端发来数据时触发
            $innerHttpWorker->onMessage = function ($httpConnection, $data) {$type = $_REQUEST['type'] ?? '';
                $content = htmlspecialchars($_REQUEST['content'] ?? '');
                $to = (string)($_REQUEST['to'] ?? '');
                
                // 推送数据的 url 格式 type=publish&to=uid&content=xxxx
                switch ($type) {
                    case 'publish':
                        // 有指定 uid 则向 uid 所在 socket 组发送数据
                        if ($to) {self::$senderIo->to($to)->emit('new_msg', $content);
                        } else {
                            // 否则向所有 uid 推送数据
                            self::$senderIo->emit('new_msg', $content);
                        }
                        // http 接口返回，如果用户离线 socket 返回 fail
                        if ($to && !isset(self::$uidConnectionCounter[$to])) {return $httpConnection->send('offline');
                        } else {return $httpConnection->send('ok');
                        }
                }
                return $httpConnection->send('fail');
            };
            // 执行监听
            $innerHttpWorker->listen();
            
            // 一个定时器，定时向所有 uid 推送当前 uid 在线数及在线页面数
            Timer::add(1, [self::class, 'emitOnlineCount']);
        });

//        Worker::$daemonize = true;
        Worker::runAll();}
    
    /**
     * 将在线数变化推送给所有登录端
     * 须是 public 方法, 可供其它类调用
     */
    public static function emitOnlineCount()
    {$newOnlineCount = count(self::$uidConnectionCounter);
        $newOnlinePageCount = array_sum(self::$uidConnectionCounter);
        
        // 只有在客户端在线数变化了才广播，减少不必要的客户端通讯
        if ($newOnlineCount != self::$onlineCount || $newOnlinePageCount != self::$onlinePageCount) {//            var_dump('emitOnlineCount:', self::$uidConnectionCounter);
            // 将在线数变化推送给所有登录端
            self::$senderIo->emit(
                'update_online_count',
                [
                    'onlineCount' => $newOnlineCount,
                    'onlinePageCount' => $newOnlinePageCount
                ]
            );
            self::$onlineCount = $newOnlineCount;
            self::$onlinePageCount = $newOnlinePageCount;
        }
    }
}

启动 PHPSocket.Io 服务

# 守护进程模式启动
php artisan msg-push start -d
#调式模式启动
php artisan msg-push start

web 页面

resources/views/socketio.blade.php

<!DOCTYPE html>
<html>
<head>
    <meta http-equiv="content-type" content="text/html;charset=utf-8">
    <title>laravel 整合 phpSocketIo</title>
</head>
<body>
<h1>laravel 整合 phpSocketIo</h1>
<h2> 实现 laravel 服务端推送消息到 web 端 </h2>
<h5> 效果查看 console</h5>


<script src='https://cdn.bootcss.com/socket.io/2.0.3/socket.io.js'></script>
<script>
document.addEventListener('DOMContentLoaded', () => {const uid = Date.now(), // 这个识别 id 可以换成项目相应业务的 id, 同一个 id 可以多端登录, 能同时收到消息
      domain = document.domain, // 当前打开页面的域名或 ip
      sendToOneApi = `http://${domain}:2121/?type=publish&content=msg_content&to=${uid}`,
      sendToAllApi = `http://${domain}:2121/?type=publish&content=msg_content`,
      socket = io(`http://${domain}:2120`); // 连接 socket 服务端

  console.log('给指定 uid 登录端发送消息接口:', sendToOneApi); // 支持 get 和 post 方法
  console.log('给所有登录端发送消息接口:', sendToAllApi);

  // 连接后登录
  socket.on('connect', function () {socket.emit('login', uid);
  });

  // 后端推送来消息时
  socket.on('new_msg', function (msg) {console.log('收到消息:' + msg);
  });

  // 后端推送来在线数据时
  socket.on('update_online_count', function (online_stat) {console.log('即时在线数据:', online_stat);
  });

});
</script>
</body>
</html>

web 页面路由

routes/web.php

Route::get('/socketio', function () {return view('socketio');
});

laravel 内以触发事件方式推送消息

app/Providers/EventServiceProvider.php

// 定义事件
//App/Providers/EventServiceProvider
public function boot()
    {parent::boot();
        
        // 推送消息到 web 端, 这个闭包只能传入一个参数
        Event::listen('send-msg', function (object $data) {//            dump($data);
            $response = (new \GuzzleHttp\Client())->post('http://127.0.0.1:2121', [
                'form_params' => [
                    'content' => $data->content,
                    'to' => $data->to ?? '','type'=> $data->type ??'publish',
                ],
            ]);
        
            return (string)$response->getBody();});
    }

浏览器方式测试推送

地址栏输入 http://${domain}:2121/?type=publish&content=Are_you_ok 推送给全体成员,${domain}是你实际的 ip 或域名

tinker 方式测试推送

# 进入 tinker
php artisan tinker
#推送给全体
event('send-msg',(object)['content'=>'hello'])
#推送给个体,`to` 改成你的实际值
event('send-msg',(object)['content'=>'hello','to'=>1556645595484])

通过以上操作即可在 php 服务端向 web 端推送消息啦, 解锁新功能是不是有点小兴奋呢?
感谢推动着时代进步的巨人们, 是你们让我等看到了更多的可能!

参考

workerman 手册
PHPSocket.IO 跨平台实时通讯框架简介