Swoole 结合TP5搭建文字直播平台

直播模块流程：

 

主进程服务：主进程同时开启两个服务

• http服务，负责向前端传递页面，处理登录等事务
• websocket服务，服务处理直播以及聊天室等事务

在项目根目录（框架代码同级目录）建立script目录，用于存放脚本文件

ws.php ： 主进程服务

<?php /** * Created by PhpStorm. * User: baidu * Date: 18/3/27 * Time: 上午12:50 */
class Ws { CONST HOST = "0.0.0.0"; CONST PORT = 8811; CONST CHART_PORT = 8812; public $ws = null; public function __construct() { // 启动服务时清空redis //开启HTTP服务
        $this->ws = new swoole_websocket_server(self::HOST, self::PORT); //添加端口，用于websocket服务
        $this->ws->listen(self::HOST, self::CHART_PORT, SWOOLE_SOCK_TCP); $this->ws->set( [ 'enable_static_handler' => true,
                'document_root' => "/home/work/hdtocs/swoole_mooc/thinkphp/public/static",
                'worker_num' => 4,
                'task_worker_num' => 4, ] ); $this->ws->on("start", [$this, 'onStart']); $this->ws->on("open", [$this, 'onOpen']); $this->ws->on("message", [$this, 'onMessage']); $this->ws->on("workerstart", [$this, 'onWorkerStart']); $this->ws->on("request", [$this, 'onRequest']); $this->ws->on("task", [$this, 'onTask']); $this->ws->on("finish", [$this, 'onFinish']); $this->ws->on("close", [$this, 'onClose']); $this->ws->start(); } /** * @param $server */
    public function onStart($server) { swoole_set_process_name("live_master"); } /** * @param $server * @param $worker_id */
    public function onWorkerStart($server, $worker_id) { // 定义应用目录
        define('APP_PATH', __DIR__ . '/../../../application/'); // 加载框架里面的文件
        require __DIR__ . '/../thinkphp/base.php'; } /** * request回调 * @param $request * @param $response */
    public function onRequest($request, $response) { if ($request->server['request_uri'] == '/favicon.ico') { $response->status(404); $response->end(); return; } $_SERVER = []; if (isset($request->server)) { foreach ($request->server as $k => $v) { $_SERVER[strtoupper($k)] = $v; } } if (isset($request->header)) { foreach ($request->header as $k => $v) { $_SERVER[strtoupper($k)] = $v; } } $_GET = []; if (isset($request->get)) { foreach ($request->get as $k => $v) { $_GET[$k] = $v; } } $_FILES = []; if (isset($request->files)) { foreach ($request->files as $k => $v) { $_FILES[$k] = $v; } } $_POST = []; if (isset($request->post)) { foreach ($request->post as $k => $v) { $_POST[$k] = $v; } } $this->writeLog(); $_POST['http_server'] = $this->ws; ob_start(); // 执行应用并响应
        try { think\Container::get('app', [APP_PATH]) ->run() ->send(); } catch (\Exception $e) { // todo
 } $res = ob_get_contents(); ob_end_clean(); $response->end($res); } /** * @param $serv * @param $taskId * @param $workerId * @param $data */
    public function onTask($serv, $taskId, $workerId, $data) { //common/task/Task.php 所有任务都保存在这个类下 // 分发 task 任务机制，让不同的任务 调用该类下相应方法
        $obj = new app\common\lib\task\Task; $method = $data['method']; $flag = $obj->$method($data['data'], $serv); return $flag; // 告诉worker
 } /** * @param $serv * @param $taskId * @param $data */
    public function onFinish($serv, $taskId, $data) { echo "taskId:{$taskId}\n"; echo "finish-data-sucess:{$data}\n"; } /** * 监听ws连接事件 * @param $ws * @param $request */
    public function onOpen($ws, $request) { // fd redis [1]
        \app\common\lib\redis\Predis::getInstance()->sAdd(config('redis.live_game_key'), $request->fd); var_dump($request->fd); } /** * 监听ws消息事件 * @param $ws * @param $frame */
    public function onMessage($ws, $frame) { echo "ser-push-message:{$frame->data}\n"; $ws->push($frame->fd, "server-push:" . date("Y-m-d H:i:s")); } /** * close * @param $ws * @param $fd */
    public function onClose($ws, $fd) { // fd del
        \app\common\lib\redis\Predis::getInstance()->sRem(config('redis.live_game_key'), $fd); echo "clientid:{$fd}\n"; } /** * 记录日志 */
    public function writeLog() { $datas = array_merge(['date' => date("Ymd H:i:s")], $_GET, $_POST, $_SERVER); $logs = ""; foreach ($datas as $key => $value) { $logs .= $key . ":" . $value . " "; } swoole_async_writefile(APP_PATH . '../runtime/log/' . date("Ym") . "/" . date("d") . "_access.log", $logs . PHP_EOL, function ($filename) { // todo
        }, FILE_APPEND); } } new Ws();

 直播推送代码： /application/admin/controller/Live.php 

class Live { public function push() { if(empty($_POST)) { //返回错误信息
           　　　　 return Util::show(config('code.error'), 'error'); } $data = [ // 这里获取直播管理页面传递过来的数据
 ]; // 获取连接的用户 // 赛况的基本信息入库 2、数据组织好 push到直播页面
       　　 $taskData = [ 'method' => 'pushLive',
         　　　　   'data' => $data ]; //Task类中 pushLive 方法负责推送数据
       　　 $_POST['http_server']->task($taskData); //返回成功给管理页面
       　　 return Util::show(config('code.success'), 'ok'); } }

Task类 /application/common/task/Task.php

class Task {
    /**
     * 通过task机制发送赛况实时数据给客户端
     * @param $data
     * @param $serv swoole server对象
     */
    public function pushLive($data, $serv) {
        $clients = Predis::getInstance()->sMembers(config("redis.live_game_key"));

        foreach($clients as $fd) {
            $serv->push($fd, json_encode($data));
        }
    }
}