跨平臺

Rpc 的請求響應是通過 tcp 作為傳輸層協(xié)議實現(xiàn)，服務廣播使用 udp 協(xié)議，所以當我們使用其他語言作為 Rpc 客戶端時，只需要實現(xiàn)對應的應用層網(wǎng)絡協(xié)議即可。

下面客戶端使用的服務端是 微服務 - 服務端章節(jié) 基于自定義節(jié)點管理器 Redis 節(jié)點管理器 實現(xiàn)的。

具體 RPC 服務端 demo 代碼可查看 Github RPC 5.x Demo Github 或者 Gitee RPC 5.x Demo Gitee

PHP RPC 客戶端示例代碼

<?php
/**
 * This file is part of EasySwoole.
 *
 * @link http://www.zbjtqy.com
 * @document http://www.zbjtqy.com
 * @contact http://www.zbjtqy.com/Preface/contact.html
 * @license https://github.com/easy-swoole/easyswoole/blob/3.x/LICENSE
 */

$data = [
    'service' => 'Goods', // 需要調用的服務名稱
    'module'  => 'GoodsModule', // 需要調用的服務下的子模塊名稱
    'action'  => 'list',  // 需要調用的服務下的子模塊的方法名稱
    'arg'     => ['a', 'b', 'c'], // 需要傳遞的參數(shù)
];

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

// tcp://127.0.0.1:9600（示例請求地址） 是 rpc 服務端的地址，這里是本地，所以使用 127.0.0.1
// 開發(fā)者需要根據(jù)實際情況調整進行調用
$fp = stream_socket_client('tcp://127.0.0.1:9600');
fwrite($fp, pack('N', strlen($raw)) . $raw); // pack 數(shù)據(jù)校驗

$try = 3;
$data = fread($fp, 4);
if (strlen($data) < 4 && $try > 0) {
    $data .= fread($fp, 4);
    $try--;
    usleep(1);
}

// 做長度頭部校驗
$len = unpack('N', $data);
$data = '';
$try = 3;
if (strlen($data) < $len[1] && $try > 0) {
    $data .= fread($fp, $len[1]);
    $try--;
    usleep(1);
}

if (strlen($data) != $len[1]) {
    echo 'data error';
} else {
    $data = json_decode($data, true);
    // 這就是服務端返回的結果
    var_dump($data);
}

fclose($fp);

/**
 * 調用結果如下：
 * 其中 
 * statue 為服務端返回給客戶端的調用狀態(tài)碼 （具體可查看服務端：http://www.zbjtqy.com/Microservices/Rpc/server.html）
 * result 為服務端返回給客戶端的調用結果
 * msg    為服務端返回給客戶端的調用狀態(tài)信息
 * responseUUID 為服務端響應客戶端的唯一標識
 */
array(4) {
  ["status"]=>
  int(0)
  ["result"]=>
  array(2) {
    [0]=>
    array(3) {
      ["goodsId"]=>
      string(6) "100001"
      ["goodsName"]=>
      string(7) "商品1"
      ["prices"]=>
      int(1124)
    }
    [1]=>
    array(3) {
      ["goodsId"]=>
      string(6) "100002"
            ["goodsName"]=>
      string(7) "商品2"
            ["prices"]=>
      int(599)
    }
  }
  ["msg"]=>
  string(22) "get goods list success"
  ["responseUUID"]=>
  string(36) "3897f7ea-12a0-39c1-8948-ee9b9bc37274"
}

注意：可能由于網(wǎng)絡問題，并不是一次就能 recv 獲取到調用結果。

Go RPC 客戶端示例代碼

/**
 * This file is part of EasySwoole.
 *
 * @link http://www.zbjtqy.com
 * @document http://www.zbjtqy.com
 * @contact http://www.zbjtqy.com/Preface/contact.html
 * @license https://github.com/easy-swoole/easyswoole/blob/3.x/LICENSE
 */

package main

import (
    "encoding/binary"
    "net"
)

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

func sendEasyswooleMsg(conn *net.TCPConn) {
    var sendData []byte
    data := `{"service":"Goods","module":"GoodsModule","action":"list","arg":["a","b","c"]}`
    b := []byte(data)
    // 大端字節(jié)序(網(wǎng)絡字節(jié)序)大端就是將高位字節(jié)放到內存的低地址端，低位字節(jié)放到高地址端。
    // 網(wǎng)絡傳輸中(比如TCP/IP)低地址端(高位字節(jié))放在流的開始，對于2個字節(jié)的字符串(AB)，傳輸順序為：A(0-7bit)、B(8-15bit)。
    sendData = int32ToBytes8(int32(len(data)))
    // 將數(shù)據(jù)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 RPC 客戶端示例代碼

/**
 * This file is part of EasySwoole.
 *
 * @link http://www.zbjtqy.com
 * @document http://www.zbjtqy.com
 * @contact http://www.zbjtqy.com/Preface/contact.html
 * @license https://github.com/easy-swoole/easyswoole/blob/3.x/LICENSE
 */

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 = "{\"service\":\"Goods\",\"module\":\"GoodsModule\",\"action\":\"list\",\"arg\":[\"a\",\"b\",\"c\"]}".getBytes();
        byte[] head = Main.toLH(msg.length);
        byte[] data = Main.mergeByteArr(head, msg);

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

        // 讀取服務器返回的數(shù)據(jù)，使用 socket 套接字對象中的字節(jié)輸入流
        InputStream in = socket.getInputStream();
        byte[] response = new byte[1024];
        int len = in.read(response);
        // 這里是 rpc 服務端返回的結果為 json 字符串
        System.out.println(new String(response, 4, len-4));
        socket.close();
    }

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

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

/**
 * 服務端返回結果如下：
 */
{"status":0,"result":[{"goodsId":"100001","goodsName":"商品1","prices":1124},{"goodsId":"100002","goodsName":"商品2","prices":599}],"msg":"get goods list success","responseUUID":"66b81f45-10f7-1a3e-fecd-9b57b021e31e"}