Swoole RPC 的實現

開始今天的文章吧，這篇文章實現了一個簡單的 RPC 遠程調用，在實現以前須要先了解什麼是 RPC，不清楚的能夠看下以前發的這篇文章 《我眼中的 RPC》。

下面的演示代碼主要使用了 Swoole 的 Task 任務池，經過 OnRequest/OnReceive 得到信息交給 Task 去處理。

舉個工做中的例子吧，在電商系統中的兩個模塊，我的中心模塊和訂單管理模塊，這兩個模塊是獨立部署的，可能不在一個機房，可能不是一個域名，如今我的中心須要經過 用戶ID 和 訂單類型 獲取訂單數據。
實現效果
客戶端

HTTP 請求

    //代碼片斷    
    <?php    
    $demo = [    
        'type'  => 'SW',    
        'token' => 'Bb1R3YLipbkTp5p0',    
        'param' => [    
            'class'  => 'Order',    
            'method' => 'get_list',    
            'param' => [    
                'uid'  => 1,    
                'type' => 2,    
            ],    
        ],    
    ];    
    $ch = curl_init();    
    $options = [    
        CURLOPT_URL  => 'http://10.211.55.4:9509/',    
        CURLOPT_POST => 1,    
        CURLOPT_POSTFIELDS => json_encode($demo),    
    ];    
    curl_setopt_array($ch, $options);    
    curl_exec($ch);    
    curl_close($ch);


TCP 請求

    //代碼片斷    
    $demo = [    
        'type'  => 'SW',    
        'token' => 'Bb1R3YLipbkTp5p0',    
        'param' => [    
            'class'  => 'Order',    
            'method' => 'get_list',    
            'param' => [    
                'uid'  => 1,    
                'type' => 2,    
            ],    
        ],    
    ];    
    $this->client->send(json_encode($demo));

 

請求方式

    SW 單個請求，等待結果

發出請求後，分配給 Task ，並等待 Task 執行完成後，再返回。

    SN 單個請求，不等待結果

發出請求後，分配給 Task 以後，就直接返回。
發送數據

 

  $demo = [    
        'type'  => 'SW',    
        'token' => 'Bb1R3YLipbkTp5p0',    
        'param' => [    
            'class'  => 'Order',    
            'method' => 'get_list',    
            'param' => [    
                'uid'  => 1,    
                'type' => 2,    
            ],    
        ],    
    ];

 

    type 同步/異步設置

    token 可進行權限驗證

    class 請求的類名

    method 請求的方法名

    uid 參數一

    type 參數二

返回數據

640?wx_fmt=png

    request_method 請求方式

    request_time 請求開始時間

    response_time 請求結束時間

    code 標識

    msg 標識值

    data 約定數據

    query 請求參數

代碼
OnRequest.php

    <?php    
    if (!defined('SERVER_PATH')) exit("No Access");    
    class OnRequest    
    {    
        private static $query;    
        private static $code;    
        private static $msg;    
        private static $data;    
        public static function run($serv, $request, $response)    
        {    
            try {    
                $data = decrypt($request->rawContent());    
                self::$query   = $data;    
                if (empty($data)) {    
                    self::$code = '-1';    
                    self::$msg  = '非法請求';    
                    self::end($request, $response);    
                }    
                //TODO 驗證Token    
                switch ($data['type']) {    
                    case 'SW': //單個請求,等待結果    
                        $task = [    
                            'request' => $data,    
                            'server'  => 'http'    
                        ];    
                        $rs = $serv->task(json_encode($task), -1, function ($serv, $task_id, $rs_data) use ($request, $response) {    
                            self::$code = '1';    
                            self::$msg  = '成功';    
                            self::$data = $rs_data['response'];    
                            self::end($request, $response);    
                        });    
                        if ($rs === false) {    
                            self::$code = '-1';    
                            self::$msg  = '失敗';    
                            self::end($request, $response);    
                        }    
                        break;    
                    case 'SN': //單個請求,不等待結果    
                        $task = [    
                            'request' => $data,    
                            'server'  => 'http'    
                        ];    
                        $rs = $serv->task(json_encode($task));    
                        if ($rs === false) {    
                            self::$code = '-1';    
                            self::$msg  = '失敗';    
                        } else {    
                            self::$code = '1';    
                            self::$msg  = '成功';    
                        }    
                        self::end($request, $response);    
                        break;    
                    default:    
                        self::$code = '-1';    
                        self::$msg  = '非法請求';    
                        self::end($request, $response);    
                }    
            } catch(Exception $e) {    
            }    
        }    
        private static function end($request = null, $response = null)    
        {    
            $rs['request_method'] = $request->server['request_method'];    
            $rs['request_time']   = $request->server['request_time'];    
            $rs['response_time']  = time();    
            $rs['code']           = self::$code;    
            $rs['msg']            = self::$msg;    
            $rs['data']           = self::$data;    
            $rs['query']          = self::$query;    
            $response->end(json_encode($rs));    
            self::$data = [];    
            return;    
        }    
    }

 

OnReceive.php

    <?php    
    if (!defined('SERVER_PATH')) exit("No Access");    
    class OnReceive    
    {    
        private static $request_time;    
        private static $query;    
        private static $code;    
        private static $msg;    
        private static $data;    
        public static function run($serv, $fd, $reactor_id, $data)    
        {    
            try {    
                self::$request_time = time();    
                $data = decrypt($data);    
                self::$query = $data;    
                //TODO 驗證Token    
                switch ($data['type']) {    
                    case 'SW': //單個請求,等待結果    
                        $task = [    
                            'fd'           => $fd,    
                            'request'      => $data,    
                            'server'       => 'tcp',    
                            'request_time' => self::$request_time,    
                        ];    
                        $rs = $serv->task(json_encode($task));    
                        if ($rs === false) {    
                            self::$code = '-1';    
                            self::$msg  = '失敗';    
                            self::handlerTask($serv, $fd);    
                        }    
                        break;    
                    case 'SN': //單個請求,不等待結果    
                        $task = [    
                            'fd'           => $fd,    
                            'request'      => $data,    
                            'server'       => 'tcp',    
                            'request_time' => self::$request_time,    
                        ];    
                        $rs = $serv->task(json_encode($task));    
                        if ($rs === false) {    
                            self::$code = '-1';    
                            self::$msg  = '失敗';    
                        } else {    
                            self::$code = '1';    
                            self::$msg  = '成功';    
                        }    
                        self::handlerTask($serv, $fd);    
                        break;    
                    default:    
                        self::$code = '-1';    
                        self::$msg  = '非法請求';    
                        self::handlerTask($serv, $fd);    
                }    
            } catch(Exception $e) {    
            }    
        }    
        private static function handlerTask($serv, $fd)    
        {    
            $rs['request_method'] = 'TCP';    
            $rs['request_time']   = self::$request_time;    
            $rs['response_time']  = time();    
            $rs['code']           = self::$code;    
            $rs['msg']            = self::$msg;    
            $rs['data']           = self::$data;    
            $rs['query']          = self::$query;    
            $serv->send($fd, json_encode($rs));    
        }    
    }