Laravel 中使用 swoole 项目实战开发案例二 (后端主动分场景给界面推送消息)life

Myth丶恋晨 2023-03-03 13:28 18阅读 0赞

我的官方群[点击此处][Link 1]**。获取更多的swoole学习资料以及视频源码笔记。**

## 需求分析 ##

我们假设有一个需求，我在后端点击按钮 1，首页弹出 “后端触发了按钮 1”。后端点了按钮 2，列表页弹出 “后端触发了按钮 2”。做到根据不同场景推送到不同页面。

## 代码思路 ##

*    
    Swoole fd  
    客户端浏览器打开或者刷新界面，在 swoole 服务会生成一个进程句柄 **fd** ，每次浏览器页面有打开链接 websocket 的 js 代码，便会生成，每次刷新的时候，会关闭之前打开的 **fd**，重新生成一个新的，关闭界面的时候会生成一个新的。swoole 的 **fd** 生成规则是从 1 开始递增。
 *    
    Redis Hash 存储 fd  
    我们建立一个 key 为 **swoole:fds** redis 哈希类型数据，**fd** 为 hash 的字段，每个字段的值我们存储前端 websocket 请求的 url 参数信息 (根据业务复杂度自己灵活变通，我在项目中会在 url 带上 sessionId)。每次链接打开 swoole 服务的时候我们存储其信息，每次关闭页面时候我们清除其字段。在 redis 存储如下

![format_png][]

*    
    触发分场景推送  
    在界面上当进行了触发操作的时候，通过后台 curl 请求 swoole http 服务，swoole http 服务根据你向我传递的参数分发给对应的逻辑处理。如 curl 请求 `127.0.0.1:9502page=back&func=pushHomeLogic&token=123456` 我们可以根据传入的 **func** 参数，在后台分发给对应逻辑处理。如分发给 **pushHomeLogic** 方法。在其里面实现自己的逻辑。为防止过多的 **ifelse** 以及 **foreach 操作**，我们采用的是闭包，**call\_user\_func** 等方法实现如下

public function onRequest($request,$response)
    {
        if ($this->checkAccess("", $request)) {
            $param = $request->get;
            // 分发处理请求逻辑
            if (isset($param['func'])) {
                if (method_exists($this,$param['func'])) {
                    call_user_func([$this,$param['func']],$request);
                }
            }
        }
    }// 往首页推送逻辑处理
    public function pushHomeLogic($request)
    {
        $callback = function (array $aContent,int $fd,SwooleDemo $oSwoole)use($request) {
            if ($aContent && $aContent['page'] == "home") {
                $aRes['message'] = "后端按了按钮1";
                $aRes['code'] = "200";
                $oSwoole::$server->push($fd,xss_json($aRes));
            }
        };
        $this->eachFdLogic($callback);
    }

## 完整代码 ##

**swool 脚本代码逻辑**

<?php
    
    namespace App\Console\Commands;
    
    use Closure;
    use Illuminate\Console\Command;
    use Illuminate\Support\Facades\Redis;
    
    class SwooleDemo extends Command
    {
        // 命令名称
        protected $signature = 'swoole:demo';
        // 命令说明
        protected $description = '这是关于swoole websocket的一个测试demo';
        // swoole websocket服务
        private static $server = null;
    
        public function __construct()
        {
            parent::__construct();
        }
    
        // 入口
        public function handle()
        {
            $this->redis = Redis::connection('websocket');
            $server = self::getWebSocketServer();
            $server->on('open',[$this,'onOpen']);
            $server->on('message', [$this, 'onMessage']);
            $server->on('close', [$this, 'onClose']);
            $server->on('request', [$this, 'onRequest']);
            $this->line("swoole服务启动成功 ...");
            $server->start();
        }
    
        // 获取服务
        public static function getWebSocketServer()
        {
            if (!(self::$server instanceof \swoole_websocket_server)) {
                self::setWebSocketServer();
            }
            return self::$server;
        }
        // 服务处始设置
        protected static function setWebSocketServer():void
        {
            self::$server  = new \swoole_websocket_server("0.0.0.0", 9502);
            self::$server->set([
                'worker_num' => 1,
                'heartbeat_check_interval' => 60,    // 60秒检测一次
                'heartbeat_idle_time' => 121,        // 121秒没活动的
            ]);
        }
    
        // 打开swoole websocket服务回调代码
        public function onOpen($server, $request)
        {
            if ($this->checkAccess($server, $request)) {
                self::$server->push($request->fd,xss_json(["code"=>200,"message"=>"打开swoole服务成功"]));
            }
        }
        // 给swoole websocket 发送消息回调代码
        public function onMessage($server, $frame)
        {
    
        }
        // http请求swoole websocket 回调代码
        public function onRequest($request,$response)
        {
            if ($this->checkAccess("", $request)) {
                $param = $request->get;
                // 分发处理请求逻辑
                if (isset($param['func'])) {
                    if (method_exists($this,$param['func'])) {
                        call_user_func([$this,$param['func']],$request);
                    }
                }
            }
        }
    
        // websocket 关闭回调代码
        public function onClose($serv,$fd)
        {
            $this->redis->hdel('swoole:fds', $fd);
            $this->line("客户端 {$fd} 关闭");
        }
    
        // 校验客户端连接的合法性,无效的连接不允许连接
        public function checkAccess($server, $request):bool
        {
            $bRes = true;
            if (!isset($request->get) || !isset($request->get['token'])) {
                self::$server->close($request->fd);
                $this->line("接口验证字段不全");
                $bRes = false;
            } else if ($request->get['token'] != 123456) {
                $this->line("接口验证错误");
                $bRes = false;
            }
            $this->storeUrlParamToRedis($request);
            return $bRes;
        }
    
        // 将每个界面打开websocket的url 存储起来
        public function storeUrlParamToRedis($request):void
        {
            // 存储请求url带的信息
            $sContent = json_encode(
                [
                    'page' => $request->get['page'],
                    'fd' => $request->fd,
                ], true);
            $this->redis->hset("swoole:fds", $request->fd, $sContent);
        }
    
        /**
         * @param $request
         * @see 循环逻辑处理
         */
        public function eachFdLogic(Closure $callback = null)
        {
            foreach (self::$server->connections as $fd) {
                if (self::$server->isEstablished($fd)) {
                    $aContent = json_decode($this->redis->hget("swoole:fds",$fd),true);
                    $callback($aContent,$fd,$this);
                } else {
                    $this->redis->hdel("swoole:fds",$fd);
                }
            }
        }
        // 往首页推送逻辑处理
        public function pushHomeLogic($request)
        {
            $callback = function (array $aContent,int $fd,SwooleDemo $oSwoole)use($request) {
                if ($aContent && $aContent['page'] == "home") {
                    $aRes['message'] = "后端按了按钮1";
                    $aRes['code'] = "200";
                    $oSwoole::$server->push($fd,xss_json($aRes));
                }
            };
            $this->eachFdLogic($callback);
        }
        // 往列表页推送逻辑处理
        public function pushListLogic($request)
        {
            $callback = function (array $aContent,int $fd,SwooleDemo $oSwoole)use($request) {
                if ($aContent && $aContent['page'] == "list") {
                    $aRes['message'] = "后端按了按钮2";
                    $aRes['code'] = "200";
                    $oSwoole::$server->push($fd,xss_json($aRes));
                }
            };
            $this->eachFdLogic($callback);
        }
    
        // 启动websocket服务
        public function start()
        {
            self::$server->start();
        }
    }

**控制器代码**

<?php
    
    namespace App\Http\Controllers;
    
    use Illuminate\Http\Request;
    use Illuminate\Support\Facades\Redis;
    class TestController extends Controller
    {
        // 首页
        public function home()
        {
            return view("home");
        }
        // 列表
        public function list()
        {
            return view("list");
        }
        // 后端控制
        public function back()
        {
            if (request()->method() == 'POST') {
               $this->curl_get($this->getUrl());
               return json_encode(['code'=>200,"message"=>"成功"]);
            } else {
                return view("back");
            }
    
        }
        // 获取要请求swoole websocet服务地址
        public function getUrl():string
        {
            // 域名 端口 请求swoole服务的方法
            $sBase = request()->server('HTTP_HOST');
            $iPort = 9502;
            $sFunc = request()->post('func');
            $sPage = "back";
            return $sBase.":".$iPort."?func=".$sFunc."&token=123456&page=".$sPage;
        }
        // curl 推送
        public function curl_get(string $url):string
        {
            $ch_curl = curl_init();
            curl_setopt ($ch_curl, CURLOPT_TIMEOUT_MS, 3000);
            curl_setopt($ch_curl, CURLOPT_SSL_VERIFYPEER, 0);
            curl_setopt ($ch_curl, CURLOPT_HEADER,false);
            curl_setopt($ch_curl, CURLOPT_HTTPGET, 1);
            curl_setopt($ch_curl, CURLOPT_RETURNTRANSFER,true);
            curl_setopt ($ch_curl, CURLOPT_URL,$url);
            $str  = curl_exec($ch_curl);
            curl_close($ch_curl);
            return $str;
        }
    }

### 页面 js 代码 ###

*    
    后端控制页

<!DOCTYPE html>
    <html lang="en">
    <head>
    <meta charset="UTF-8">
    <title>后端界面</title>
    <meta name=viewport content="width=device-width,initial-scale=1,maximum-scale=1,user-scalable=no">
    </head>
    <body>
    <button class="push" data-func="pushHomeLogic">按钮1</button>
    <button class="push" data-func="pushListLogic">按钮2</button>
    </body>
    <script src="{
       { asset("/vendor/tw/global/jQuery/jquery-2.2.3.min.js")}} "></script>
    <script>
    $(function () {
        $(".push").on('click',function(){
            var func = $(this).attr('data-func').trim();
            ajaxGet(func)
        })
        function ajaxGet(func) {
            url = "{
       {route('back')}}";
            token = "{
       {csrf_token()}}";
            $.ajax({
                url: url,
                type: 'post',
                dataType: "json",
                data:{func:func,_token:token},
                error: function (data) {
                    alert("服务器繁忙, 请联系管理员！");
                    return;
                },
                success: function (result) {
    
                },
            })
        }
    
    })
    </script>
    </html>

**首页**

<!DOCTYPE html>
    <html lang="en">
    <head>
    <meta charset="UTF-8">
    <title>swoole首页</title>
    <meta name=viewport content="width=device-width,initial-scale=1,maximum-scale=1,user-scalable=no">
    </head>
    <body>
    <h1>这是首页</h1>
    </body>
    <script>
    var ws;//websocket实例
    var lockReconnect = false;//避免重复连接
    var wsUrl = 'ws://{
       {$_SERVER["HTTP_HOST"]}}:9502?page=home&token=123456';
    
    function initEventHandle() {
        ws.onclose = function () {
            reconnect(wsUrl);
        };
        ws.onerror = function () {
            reconnect(wsUrl);
        };
        ws.onopen = function () {
            //心跳检测重置
            heartCheck.reset().start();
        };
        ws.onmessage = function (event) {
            //如果获取到消息，心跳检测重置
            //拿到任何消息都说明当前连接是正常的
            var data = JSON.parse(event.data);
            if (data.code == 200) {
                console.log(data.message)
            }
            heartCheck.reset().start();
        }
    }
    createWebSocket(wsUrl);
    /**
     * 创建链接
     * @param url
     */
    function createWebSocket(url) {
        try {
            ws = new WebSocket(url);
            initEventHandle();
        } catch (e) {
            reconnect(url);
        }
    }
    function reconnect(url) {
        if(lockReconnect) return;
        lockReconnect = true;
        //没连接上会一直重连，设置延迟避免请求过多
        setTimeout(function () {
            createWebSocket(url);
            lockReconnect = false;
        }, 2000);
    }
    //心跳检测
    var heartCheck = {
        timeout: 60000,//60秒
        timeoutObj: null,
        serverTimeoutObj: null,
        reset: function(){
            clearTimeout(this.timeoutObj);
            clearTimeout(this.serverTimeoutObj);
            return this;
        },
        start: function(){
            var self = this;
            this.timeoutObj = setTimeout(function(){
                //这里发送一个心跳，后端收到后，返回一个心跳消息，
                //onmessage拿到返回的心跳就说明连接正常
                ws.send("heartbeat");
                self.serverTimeoutObj = setTimeout(function(){//如果超过一定时间还没重置，说明后端主动断开了
                    ws.close();//如果onclose会执行reconnect，我们执行ws.close()就行了.如果直接执行reconnect 会触发onclose导致重连两次
                }, self.timeout);
            }, this.timeout);
        },
        header:function(url) {
            window.location.href=url
        }
    
    }
    </script>
    </html>

### 列表页面 ###

<!DOCTYPE html>
    <html lang="en">
    <head>
    <meta charset="UTF-8">
    <title>swoole列表页</title>
    <meta name=viewport content="width=device-width,initial-scale=1,maximum-scale=1,user-scalable=no">
    </head>
    <body>
    <h1>swoole列表页</h1>
    </body>
    <script>
    var ws;//websocket实例
    var lockReconnect = false;//避免重复连接
    var wsUrl = 'ws://{
       {$_SERVER["HTTP_HOST"]}}:9502?page=list&token=123456';
    
    function initEventHandle() {
        ws.onclose = function () {
            reconnect(wsUrl);
        };
        ws.onerror = function () {
            reconnect(wsUrl);
        };
        ws.onopen = function () {
            //心跳检测重置
            heartCheck.reset().start();
        };
        ws.onmessage = function (event) {
            //如果获取到消息，心跳检测重置
            //拿到任何消息都说明当前连接是正常的
            var data = JSON.parse(event.data);
            if (data.code == 200) {
                console.log(data.message)
            }
            heartCheck.reset().start();
        }
    }
    createWebSocket(wsUrl);
    /**
     * 创建链接
     * @param url
     */
    function createWebSocket(url) {
        try {
            ws = new WebSocket(url);
            initEventHandle();
        } catch (e) {
            reconnect(url);
        }
    }
    function reconnect(url) {
        if(lockReconnect) return;
        lockReconnect = true;
        //没连接上会一直重连，设置延迟避免请求过多
        setTimeout(function () {
            createWebSocket(url);
            lockReconnect = false;
        }, 2000);
    }
    //心跳检测
    var heartCheck = {
        timeout: 60000,//60秒
        timeoutObj: null,
        serverTimeoutObj: null,
        reset: function(){
            clearTimeout(this.timeoutObj);
            clearTimeout(this.serverTimeoutObj);
            return this;
        },
        start: function(){
            var self = this;
            this.timeoutObj = setTimeout(function(){
                //这里发送一个心跳，后端收到后，返回一个心跳消息，
                //onmessage拿到返回的心跳就说明连接正常
                ws.send("heartbeat");
                self.serverTimeoutObj = setTimeout(function(){//如果超过一定时间还没重置，说明后端主动断开了
                    ws.close();//如果onclose会执行reconnect，我们执行ws.close()就行了.如果直接执行reconnect 会触发onclose导致重连两次
                }, self.timeout);
            }, this.timeout);
        },
        header:function(url) {
            window.location.href=url
        }
    
    }
    </script>
    </html>

### 界面效果 ###

### 后台控制点击按钮 1 ###

![format_png 1][]

![format_png 2][]

## 后端界面点击按钮 2 ##

![format_png 3][]

希望以上内容能帮助到大家，加入我的官方群[点击此处][Link 1]**。获取更多的swoole学习资料以及视频源码笔记。**

[Link 1]: https://link.zhihu.com/?target=https%3A//jq.qq.com/%3F_wv%3D1027%26k%3DJNZRtm7R
[format_png]: /images/20230208/37c929a9bcb64f74b2548171e22eb731.png
[format_png 1]: /images/20230208/e717035d1dde4e3d88cf65b9f90fb079.png
[format_png 2]: /images/20230208/d6e7713eca6c473cbbc9a2789b8bfe74.png
[format_png 3]: /images/20230208/85176b7c800d4078a1668768410d7ef2.png