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));
        }
    }
}