php workman实现socket服务及客户端连接

秒速五厘米 2022-05-19 13:47 379阅读 0赞

*  **1. 解决什么问题，为什么要用workman socket服务**

都知道游戏安装包很大，渠道推广时，需要对游戏进行分包处理，而PHP命令模式是单进程，一次只能分一次包，故这里用workman实现socket服务开启多进程，对游戏进行分包处理（一个进程处理一个分包，多个进程可以同时处理多个分包）

*  **2. 服务端代码**

server.php

<?php
    /**
     * 分包程序.切记不能有die或exit出现.
     *
     * User: yzm
     * Data: 2018/1/16
     */
    
    require_once './vendor/workerman/workerman/Autoloader.php';
    require_once './Lib/Function.php';
    
    require_once __DIR__ . '/Lib/Db.php';
    require_once __DIR__ . '/Lib/DbConnection.php';
    require_once __DIR__ . '/Config/Db.php';
    
    use Workerman\Worker;
    
    // #### create socket and listen 1234 port ####
    $tcp_worker = new Worker("tcp://0.0.0.0:9998");
    
    /**
     * 定义常量.
     */
    define('REP_SUCCESS', 0); // 成功
    define('REP_FAIL', -1); // 失败
    define('REP_FAIL_NO_COMPLETED', 1); // 文件未上传完成
    
    
    // 16 processes,与cpu个数相同
    $tcp_worker->count = 16;
    $msg = '';
    
    define('ORGPKG', '/Volumes/VMware\ Shared\ Folders/orgpkg/');
    define('DISTPKG', '/Volumes/VMware\ Shared\ Folders/');
    //define('SYS_IP', '39.108.223.28');
    define('SYS_IP', '120.92.142.115');
    define('IOS_URL','http://ios.package.tonguu.cn/');
    
    
    // Emitted when new connection come
    $tcp_worker->onConnect = function ($connection) {
        $connection->sized = 0;
    
        // xcode调用脚本
        $certMobile = '/mnt/www/DIVIDE_PKG/Cert/%d/mslabEnt.mobileprovision'; // 证书文件
        $shell = "/mnt/www/DIVIDE_PKG/Lib/dividePkg/resign  sign -ipapath  %s  -destpath %s  -pppath %s -agentid %s";
    
        $connection->shell = $shell;
        $connection->pppath = $certMobile;
    
        echo date("Y-m-d H:i:s") . " connect!" . getclientip() . PHP_EOL;
    
    };
    
    /**
     * 响应结果.
     *
     * @author yzm
     */
    function resonse($conn, $msg, $error = REP_FAIL, $data = [])
    {
        $res = ['msg' => $msg, 'error' => intval($error)];
        if (!empty($data)) {
            $res['content'] = $data;
        }
    
        debug($res);
    
        // 返回JSON数据格式到客户端 包含状态信息
        $rst = json_encode($res);
    
        $conn->send($rst);
    }
    
    
    // Emitted when data received
    $tcp_worker->onMessage = function ($connection, $data) {
        set_time_limit(0);
        ini_set('memory_limit', -1);
    
        $db = \Lib\Db::instance('btmox');
        $data = @json_decode($data, true);
    
        try{
            if (empty($data['authId'])) {
                throw new \Exception('授权文件参数错误');
            }
    
            //1. 查询所有待分包的ios渠道包
            $iosPkg = $db
                ->select('a.id,a.vid,a.filename,a.agent,d.pinyin,b.name,c.package_name')
                ->from('cy_ct_ios_package a')
                ->where("a.status=0 AND c.is_send=1")
                ->leftJoin('cy_ct_ios_mobileversion b','b.id=a.m_v_id')
                ->rightJoin('cy_ct_ios_version c','c.id=a.vid')
                ->leftJoin('cy_game d','d.id=c.game_id')
                ->orderByASC(['a.create_time'])->query();
    
            if(empty($iosPkg)) throw new \Exception('没有需要待分包的数据'.PHP_EOL);
    
            //2. 分包
            foreach($iosPkg as $one){
                try{
                    //对当前正要分的包把状态改为‘分包中’
                    $db->update('cy_ct_ios_package')->cols([
                        'status' => 2,
                    ])->where("id=".$one['id'])->query();
    
                    $filename = $one['pinyin'];
                    // 渠道分包
                    $verId = @$one['vid'];
                    $agent = @$one['agent'];
                    $location = isset($data['location']) ? $data['location'] : 1;
                    $authId = @intval($data['authId']); // 授权文件
    
                    if (empty($verId) || empty($agent)) {
                        throw new \Exception("分包失败：".$one['id']."版本、渠道为空\r\n");
                    }
    
                    // 替换\,否则PHP验证不文件是否存在
                    $orgPkg = str_replace('\\', '', ORGPKG) . "{$filename}.ipa";
    
                    debug($one['id'].'原包：' . $orgPkg);
    
                    debug($one['id'].'是否是文件：' . is_file($orgPkg));
    
                    if (!is_file($orgPkg)) {
                        throw new \Exception("分包失败：".$one['id']."母包不存在-$orgPkg\r\n");
                    }
    
                    // 从新拼接文件
                    $orgPkg = ORGPKG . "{$filename}.ipa";
    
                    // 获取目标包存放路径
                    $distPkgPath = getDistPkgPath($location);
    
                    $distPkg = $distPkgPath . "$filename/vers_{$verId}/{$filename}_$agent.ipa";
                    debug('渠道分包地址：' . $distPkg);
                    if (file_exists($filename)) {
                        @unlink($filename);
                    }
    
                    // 替换授权文件
                    $certMobile = sprintf($connection->pppath, $authId);
    
                    // 渠道分包
                    list($msg, $code) = dividePkg($connection->shell, $orgPkg, $distPkg, $agent, $certMobile);
    
                    debug('$code' . $code);
    
                    if ($code != 0) {
                        throw new \Exception("分包失败：".$msg."\r\n");
                    }
    
                    $distPkg = str_replace($distPkgPath, '', $distPkg);
    
                }catch (\Exception $ex){
                    debug($ex->getMessage());
                    $code = -1;
                    $msg = $ex->getMessage();
                }
    
                //3. 分包后更新分包结果，状态，下载地址
                $status = $code == 0 ? 1 : 2;
                $sdata['status'] = $status;
                $sdata['message'] = $msg;
                if($status == 1){
                    $sdata['url'] = IOS_URL.$distPkg;
                }
                $db->update('cy_ct_ios_package')->cols($sdata)->where("id=".$one['id'])->query();
            }
    
            resonse($connection, $msg,$code);
        }catch (\Exception  $ex){
            resonse($connection, $ex->getMessage());
        }
    };
    
    // Emitted when new connection come
    $tcp_worker->onClose = function ($connection) {
        echo date("Y-m-d H:i:s") . " closed!" . PHP_EOL;
    };
    
    Worker::runAll();

*  **3. 客户端代码**

client.php

<?php
    
    /**
     * 读取socket数据.
     *
     * @author yzm
     *
     * @param $socket
     * @param bool|true $isDividePkg
     * @return array|null|string
     */
    function socketRead($socket, $isDividePkg = true)
    {
        $rst = null;
    
        $buf = socket_read($socket, 8192);
        if ($isDividePkg) {
            $_buf = @json_decode($buf, true);
            $rst = !empty($_buf) ? [$_buf['error'], $_buf['msg'], @$_buf['content']] : $buf;
        } else {
            $rst = $buf;
        }
    
        return $rst;
    }
    
    /**
     * 向物理机发起socket请求.
     *
     * @param $args 参数
     * @return bool
     * @throws \Exception
     */
    function sendSocket($args)
    {
        set_time_limit(0);
        ini_set('memory_limit', -1);
    
        $type = isset($args['type']) ? $args['type'] : 0;
    
        if (!$type) throw new \Exception('类型参数错误');
    
        $port = 9998;
        $ip = "127.0.0.1";
    
        // 创建socket
        $socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP);
    
        if ($socket <= 0) throw new \Exception('创建socket失败,REASON:' . socket_strerror($socket));
    
        try {
    
            // 连接服务器
            $result = socket_connect($socket, $ip, $port);
            if ($result < 0 || is_null($result) || !$result) throw new \Exception('连接失败,REASON:' . socket_strerror($result));
    
            $in = json_encode($args);
    
            // 写入文件信息
            if (!socket_write($socket, $in, strlen($in))) throw new \Exception('消息发送失败,REASON:' . socket_strerror($socket));
    
            // 读取socket返回的数据
            list($error, $msg, $data) = socketRead($socket);
    
            if ($type != 3 && $error != 0) throw new \Exception('104服务器异常,REASON:' . $msg);
    
            // 关闭socket
            socket_close($socket);
    
            switch ($type) {
                case 2: // 分包
                    $rst = $data['url'];
                    break;
                case 3: // 检测文件
                    if ($error == -1) {
                        throw new \Exception('检测文件失败,REASON:' . $msg);
                    }
    
                    $rst = $error;
                    break;
                default:
                    $rst = true;
                    break;
            }
    
        } catch (\Exception $ex) {
    
            // 关闭socket
            @socket_close($socket);
    
            throw new \Exception($ex->getMessage());
        }
    
        return $rst;
    }
    
    
    /**
     * 分包程序.切记不能有die或exit出现.
     *
     * User: yzm
     * Data: 2018/1/16
     */
    require_once './Lib/Function.php';
    
    $i=0;
    while ($i<30){
        try{
            $data['type'] = 1;
            $data['authId'] = 2;
            $data['location'] = 1;
            sendSocket($data);
        }catch (\Exception $ex){
            echo $ex->getMessage();
        }
        $i++;
        sleep(5);
    }

*  **4. 使用**

a. 开启服务

php server.php start //可以看到开启了多个进程

b. 客户端连接

php client.php //从代码知道，里头用了循环，可以多次连接服务，同时发送数据，服务端会把结果返回