EasySwoole RPC

很多傳統的 Phper 并不懂 RPC 是什么，RPC 全稱 Remote Procedure Call，中文譯為 遠程過程調用，其實你可以把它理解為是一種架構性上的設計，或者是一種解決方案。

例如在某龐大商場系統中，你可以把整個商場拆分為 N 個微服務（理解為 N 個獨立的小模塊也行），例如：

訂單系統
用戶管理系統
商品管理系統
等等

那么在這樣的架構中，就會存在一個 API 網關 的概念，或者是叫 服務集成者。我的 API 網關 的職責，就是把一個請求，拆分成 N 個小請求，分發到各個小服務里面，再整合各個小服務的結果，返回給用戶。例如在某次下單請求中，那么大概發送的邏輯如下：

API 網關接受請求
API 網關提取用戶參數，請求用戶管理系統，獲取用戶余額等信息，等待結果
API 網關提取商品參數，請求商品管理系統，獲取商品剩余庫存和價格等信息，等待結果
API 網關融合用戶管理系統、商品管理系統的返回結果，進行下一步調用（假設滿足購買條件）
API 網關調用用戶管理信息系統進行扣款，調用商品管理系統進行庫存扣減，調用訂單系統進行下單（事務邏輯和撤回可以用 請求 id 保證，或者自己實現其他邏輯調度）
API 網關返回綜合信息給用戶

而在以上發生的行為，就稱為 遠程過程調用。而調用過程實現的通訊協議可以有很多，比如常見的 HTTP 協議。而 EasySwoole RPC 采用自定義短鏈接的 TCP 協議實現，每個請求包，都是一個 JSON，從而方便實現跨平臺調用。

全新特性

協程調度
服務自動發現
服務熔斷
服務降級
Openssl 加密
跨平臺、跨語言支持
支持接入第三方注冊中心

安裝

composer require easyswoole/rpc=4.x

執行流程

服務端：
注冊RPC服務，創建相應的服務swoole table表（ps:記錄調用成功和失敗的次數）注冊worker,tick進程

woker進程監聽：
客戶端發送請求->解包成相對應的格式->執行對應的服務->返回結果->客戶端

tick進程：
注冊定時器發送心跳包到本節點管理器
啟用廣播：每隔幾秒發送本節點各個服務信息到其他節點
啟用監聽：監聽其他節點發送的信息，發送相對應的命令（心跳|下線）到節點管理器處理
進程關閉：主動刪除本節點的信息，發送下線廣播到其他節點

Rpc-Server

場景

例如在一個商場系統中，我們將商品庫和系統公告兩個服務切分開到不同的服務器當中。當用戶打開商場首頁的時候，我們希望App向某個網關發起請求，該網關可以自動的幫我們請求商品列表和系統公共等數據，合并返回。

服務定義

每一個Rpc服務其實就一個EasySwoole\Rpc\AbstractService類。如下：

定義商品服務

namespace App\RpcService;

use EasySwoole\Rpc\AbstractService;

class Goods extends AbstractService
{

    /**
     *  重寫onRequest(比如可以對方法做ip攔截或其它前置操作)
     *
     * @param string $action
     * @return bool
     * CreateTime: 2020/6/20 下午11:12
     */
    protected function onRequest(?string $action): ?bool
    {
        return true;
    }

    public function serviceName(): string
    {
        return 'goods';
    }

    public function list()
    {
        $this->response()->setResult([
            [
                'goodsId'=>'100001',
                'goodsName'=>'商品1',
                'prices'=>1124
            ],
            [
                'goodsId'=>'100002',
                'goodsName'=>'商品2',
                'prices'=>599
            ]
        ]);
        $this->response()->setMsg('get goods list success');
    }
}

定義公共服務

namespace App\RpcService;

use EasySwoole\Rpc\AbstractService;

class Common extends AbstractService
{
    public function serviceName(): string
    {
        return 'common';
    }

    public function mailBox()
    {
        $this->response()->setResult([
            [
                'mailId'=>'100001',
                'mailTitle'=>'系統消息1',
            ],
            [
                'mailId'=>'100001',
                'mailTitle'=>'系統消息1',
            ],
        ]);
        $this->response()->setMsg('get mail list success');
    }

    public function serverTime()
    {
        $this->response()->setResult(time());
        $this->response()->setMsg('get server time success');
    }
}

服務注冊

在Easyswoole全局的Event文件中，進行服務注冊。至于節點管理、服務類定義等具體用法請看對應章節。

namespace EasySwoole\EasySwoole;

use App\RpcService\Common;
use App\RpcService\Goods;
use EasySwoole\EasySwoole\Swoole\EventRegister;
use EasySwoole\EasySwoole\AbstractInterface\Event;
use EasySwoole\Http\Request;
use EasySwoole\Http\Response;
use EasySwoole\Redis\Config\RedisConfig;
use EasySwoole\RedisPool\RedisPool;
use EasySwoole\Rpc\NodeManager\RedisManager;
use EasySwoole\Rpc\Config as RpcConfig;
use EasySwoole\Rpc\Rpc;

class EasySwooleEvent implements Event
{

    public static function initialize()
    {
        // TODO: Implement initialize() method.
        date_default_timezone_set('Asia/Shanghai');
    }

    public static function mainServerCreate(EventRegister $register)
    {
        /*
         * 定義節點Redis管理器
         */
        $redisPool = new RedisPool(new RedisConfig([
            'host'=>'127.0.0.1'
        ]));
        $manager = new RedisManager($redisPool);
        //配置Rpc實例
        $config = new RpcConfig();
        //這邊用于指定當前服務節點ip，如果不指定，則默認用UDP廣播得到的地址
        $config->setServerIp('127.0.0.1');
        $config->setNodeManager($manager);
        /*
         * 配置初始化
         */
        Rpc::getInstance($config);
        //添加服務
        Rpc::getInstance()->add(new Goods());
        Rpc::getInstance()->add(new Common());
        Rpc::getInstance()->attachToServer(ServerManager::getInstance()->getSwooleServer());
    }

    public static function onRequest(Request $request, Response $response): bool
    {
        // TODO: Implement onRequest() method.
        return true;
    }

    public static function afterRequest(Request $request, Response $response): void
    {
        // TODO: Implement afterAction() method.
    }
}

為了方便測試，我把兩個服務放在同一臺機器中注冊。實際生產場景應該是N臺機注冊商品服務，N臺機器注冊公告服務，把服務分開。

Rpc-Client

控制器聚合調用

namespace App\HttpController;

use EasySwoole\Http\AbstractInterface\Controller;
use EasySwoole\Rpc\Response;
use EasySwoole\Rpc\Rpc;

class Index extends Controller
{

    function index()
    {
        $ret = [];
        $client = Rpc::getInstance()->client();
        /*
         * 調用商品列表
         */
        $client->addCall('goods','list',['page'=>1])
            ->setOnSuccess(function (Response $response)use(&$ret){
                $ret['goods'] = $response->toArray();
            })->setOnFail(function (Response $response)use(&$ret){
                $ret['goods'] = $response->toArray();
            });
        /*
         * 調用信箱公共
         */
        $client->addCall('common','mailBox')
            ->setOnSuccess(function (Response $response)use(&$ret){
                $ret['mailBox'] = $response->toArray();
            })->setOnFail(function (Response $response)use(&$ret){
                $ret['mailBox'] = $response->toArray();
            });
        /*
        * 獲取系統時間
        */
        $client->addCall('common','serverTime')
            ->setOnSuccess(function (Response $response)use(&$ret){
                $ret['serverTime'] = $response->toArray();
            });

        $client->exec(2.0);

        $this->writeJson(200,$ret);
    }
}

注意，控制器中可以這樣調用，是因為服務端章節中，在EasySwoole的全局啟動事件已經對當前的Rpc實例定義注冊了節點管理器。因此在控制器中調用的時候該Rpc實例可以找到對應的節點。一般來說，在做聚合網關的節點，是不需要注冊服務進去的，僅需注冊節點管理器即可。

客戶端

當rpc服務和客戶端不在同一服務中時，并且服務端客戶端使用的都是es

<?php
require_once 'vendor/autoload.php';

use EasySwoole\Rpc\Config;
use EasySwoole\Rpc\Rpc;
use EasySwoole\Rpc\NodeManager\RedisManager;
use EasySwoole\Rpc\Response;
$redisConfig = new \EasySwoole\Redis\Config\RedisConfig();
$redisConfig->setHost('127.0.0.1'); // 服務端使用的redis節點地址
$redisConfig->setPort('6379'); // 服務端使用的redis節點端口
$pool=new \EasySwoole\RedisPool\RedisPool($redisConfig);
$config = new Config();
$config->setServerIp('127.0.0.1'); // 指定rpc服務地址
$config->setListenPort(9502); // 指定rpc服務端口
$config->setNodeManager(new RedisManager($pool));
$rpc = new Rpc($config);

\Swoole\Coroutine::create(function () use ($rpc) {
    $client = $rpc->client();
    $client->addCall('UserService', 'register', ['arg1', 'arg2'])
        ->setOnFail(function (Response $response) {
            print_r($response->toArray());
        })
        ->setOnSuccess(function (Response $response) {
            print_r($response->toArray());
        });

    $client->exec();
});
swoole_timer_clear_all();

跨平臺

Rpc的請求響應通過tcp協議,服務廣播使用udp協議,我們只需要實現網絡協議即可。

PHP示例代碼

<?php
/**
 * Created by PhpStorm.
 * User: xcg
 * Date: 2019/6/17
 * Time: 14:30
 */
$data = [
    'command' => 1,//1:請求,2:狀態rpc 各個服務的狀態
    'request' => [
        'serviceName' => 'UserService',
        'action' => 'register',//行為名稱
        'arg' => [
            'args1' => 'args1',
            'args2' => 'args2'
        ]
    ]
];

//$raw = serialize($data);//注意序列化類型,需要和RPC服務端約定好協議 $serializeType

$raw = json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES);

$fp = stream_socket_client('tcp://127.0.0.1:9600');
fwrite($fp, pack('N', strlen($raw)) . $raw);//pack數據校驗

$data = fread($fp, 65533);
//做長度頭部校驗
$len = unpack('N', $data);
$data = substr($data, '4');
if (strlen($data) != $len[1]) {
    echo 'data error';
} else {
    $data = json_decode($data, true);
//    //這就是服務端返回的結果，
    var_dump($data);//默認將返回一個response對象 通過$serializeType修改
}
fclose($fp);

Go示例代碼

package main

import (
    "encoding/binary"
    "net"
)

func main() {
    var tcpAddr *net.TCPAddr
    tcpAddr,_ = net.ResolveTCPAddr("tcp","127.0.0.1:9600")
    conn,_ := net.DialTCP("tcp",nil,tcpAddr)
    defer conn.Close()
    sendEasyswooleMsg(conn)
}

func sendEasyswooleMsg(conn *net.TCPConn) {
    var sendData []byte
    data := `{"command":1,"request":{"serviceName":"UserService","action":"register","arg":{"args1":"args1","args2":"args2"}}}`
    b := []byte(data)
    // 大端字節序(網絡字節序)大端就是將高位字節放到內存的低地址端，低位字節放到高地址端。
    // 網絡傳輸中(比如TCP/IP)低地址端(高位字節)放在流的開始，對于2個字節的字符串(AB)，傳輸順序為：A(0-7bit)、B(8-15bit)。
    sendData = int32ToBytes8(int32(len(data)))
    // 將數據byte拼裝到sendData的后面
    for _, value := range b {
        sendData = append(sendData, value)
    }
    conn.Write(sendData)
}

func int32ToBytes8(n int32) []byte {
    var buf = make([]byte, 4)
    binary.BigEndian.PutUint32(buf, uint32(n))
    return buf
}

Java

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;

public class Main {
    public static void main(String[] args) throws IOException {
        byte[] msg = "{\"command\":1,\"request\":{\"serviceName\":\"UserService\",\"action\":\"register\",\"arg\":{\"args1\":\"args1\",\"args2\":\"args2\"}}}".getBytes();
        byte[] head = Main.toLH(msg.length);
        byte[] data = Main.mergeByteArr(head, msg);

        //創建Socket對象，連接服務器
        Socket socket=new Socket("127.0.0.1",9600);
        //通過客戶端的套接字對象Socket方法，獲取字節輸出流，將數據寫向服務器
        OutputStream out=socket.getOutputStream();
        out.write(data);

        //讀取服務器發回的數據，使用socket套接字對象中的字節輸入流
        InputStream in=socket.getInputStream();
        byte[] response=new byte[1024];
        int len=in.read(response);
        System.out.println(new String(response,4, len-4));
        socket.close();
    }

    static byte[] toLH(int n) {
        byte[] b = new byte[4];
        b[3] = (byte) (n & 0xff);
        b[2] = (byte) (n >> 8 & 0xff);
        b[1] = (byte) (n >> 16 & 0xff);
        b[0] = (byte) (n >> 24 & 0xff);
        return b;
    }

    static byte[] mergeByteArr(byte[] a, byte[] b) {
        byte[] c= new byte[a.length + b.length];
        System.arraycopy(a, 0, c, 0, a.length);
        System.arraycopy(b, 0, c, a.length, b.length);
        return c;
    }
}

其他語言只需要實現tcp協議即可

EasySwoole RPC 自定義注冊中心

EasySwoole默認為通過UDP廣播+自定義進程定時刷新自身節點信息的方式來實現無主化/注冊中心的服務發現。在服務正常關閉的時候，自定義定時進程的onShutdown方法會執行deleteServiceNode方法來實現節點下線。在非正常關閉的時候，心跳超時也會被節點管理器踢出。

有些情況，不方便用UDP廣播的情況下，那么EasySwoole支持你自定義一個節點管理器，來變更服務發現方式。

例如用Redis來實現

namespace EasySwoole\Rpc\NodeManager;

use EasySwoole\RedisPool\RedisPool;
use EasySwoole\Rpc\ServiceNode;
use EasySwoole\Utility\Random;

class RedisManager implements NodeManagerInterface
{
    protected $redisKey;

    protected $pool;

    function __construct(RedisPool $pool, string $hashKey = 'rpc')
    {
        $this->redisKey = $hashKey;
        $this->pool = $pool;
    }

    function getServiceNodes(string $serviceName, ?string $version = null): array
    {
        $redis = $this->pool->getObj(15);
        try {
            $nodes = $redis->hGetAll("{$this->redisKey}_{$serviceName}");
            $nodes = $nodes ?: [];
            $ret = [];
            foreach ($nodes as $nodeId => $node) {
                $node = new ServiceNode(json_decode($node,true));
                /**
                 * @var  $nodeId
                 * @var  ServiceNode $node
                 */
                if (time() - $node->getLastHeartBeat() > 30) {
                    $this->deleteServiceNode($node);
                }
                if ($version && $version != $node->getServiceVersion()) {
                    continue;
                }
                $ret[$nodeId] = $node;
            }
            return $ret;
        } catch (\Throwable $throwable) {
            //如果該redis斷線則銷毀
            $this->pool->unsetObj($redis);
        } finally {
            $this->pool->recycleObj($redis);
        }
        return [];
    }

    function getServiceNode(string $serviceName, ?string $version = null): ?ServiceNode
    {
        $list = $this->getServiceNodes($serviceName, $version);
        if (empty($list)) {
            return null;
        }
        return Random::arrayRandOne($list);
    }

    function deleteServiceNode(ServiceNode $serviceNode): bool
    {
        $redis = $this->pool->getObj(15);
        try {
            $redis->hDel($this->generateNodeKey($serviceNode), $serviceNode->getNodeId());
            return true;
        } catch (\Throwable $throwable) {
            $this->pool->unsetObj($redis);
        } finally {
            $this->pool->recycleObj($redis);
        }
        return false;
    }

    function serviceNodeHeartBeat(ServiceNode $serviceNode): bool
    {
        if (empty($serviceNode->getLastHeartBeat())) {
            $serviceNode->setLastHeartBeat(time());
        }
        $redis = $this->pool->getObj(15);
        try {
            $redis->hSet($this->generateNodeKey($serviceNode), $serviceNode->getNodeId(), $serviceNode->__toString());
            return true;
        } catch (\Throwable $throwable) {
            $this->pool->unsetObj($redis);
        } finally {
            //這邊需要測試一個對象被unset后是否還能被回收
            $this->pool->recycleObj($redis);
        }
        return false;
    }

    protected function generateNodeKey(ServiceNode $node)
    {
        return "{$this->redisKey}_{$node->getServiceName()}";
    }
}

即使關閉了UDP定時廣,EasySwoole Rpc的tick進程依舊會每3秒執行一次serviceNodeHeartBeat用于更新自身的節點心跳信息。

亚洲精品成人_精品成人一区_999视频在线播放_免费黄色在线_亚洲成人久久久_久久www免费视频