• go实现自定义rpc框架 (核心：服务端&客户端、自定义io流、编解码、服务发现、负载均衡、支持多语言网关等)

    git地址：GitHub - radmin-zhangjian/zrpc: zrpc go实现自定义rpc框架

    go开发脚手架 动态/静态路由 (根据gin框架)https://mp.csdn.net/mp_blog/creation/editor/126445356

    目录

    zrpc

    服务端

    客户端

    php客户端 (socket方式)

    网关 支持多语言

    php 连接网关 (curl方式)

    快速开始 - 服务端

    快速开始 - 客户端

    快速开始 - php客户端

    快速开始 - 网关

    zrpc

    说明: zrpc实现了以下基本功能
    1、自定义协议字节流
    2、定义编解码，支持多种方法
    3、注册服务方法
    4、反射调用方法及参数，根据struct方式和map方式传参
    5、支持同步和异步调用
    6、服务发现：redis实现(根据路径查询和redis-scan方式)、其他方式待实现
    7、负载均衡：随机和轮询、其他方式待实现
    8、捕获业务程序异常防止崩溃
    9、支持多语言 网关(http)
    10、php通过socket调用zrpc

    服务端

    go run test/server/main.go -addr=127.0.0.1:8091
    go run test/server/main.go -addr=127.0.0.1:8092
    go run test/server/main.go -addr=127.0.0.1:8093

    客户端

    go run test/client/main.go

    php客户端 (socket方式)

    test/client/client.php

    网关 支持多语言

    go run test/gateway/gateway.go

    php 连接网关 (curl方式)

    test/gateway/gateway.php

    快速开始 - 服务端

    1. package main
    2.  
    3. import (
    4. 	"encoding/gob"
    5. 	"flag"
    6. 	"log"
    7. 	"zrpc/example/service"
    8. 	v1 "zrpc/example/v1"
    9. 	v2 "zrpc/example/v2"
    10. 	"zrpc/rpc"
    11. )
    12.  
    13. // go run main.go -addr=127.0.0.1:8092
    14. var (
    15. 	addr     = flag.String("addr", ":8092", "server address")
    16. 	registry = flag.String("registry", "redis://127.0.0.1:6379", "registry address")
    17. 	basePath = flag.String("basepath", "/zrpc_center", "")
    18. )
    19.  
    20. // 自己定义数据格式的读写
    21. func main() {
    22. 	// 解析参数
    23. 	if !flag.Parsed() {
    24. 		flag.Parse()
    25. 	}
    26.  
    27. 	// gob 编解码时需要注册
    28. 	gob.Register(map[string]interface{}{})
    29. 	gob.Register(service.User{})
    30. 	gob.Register(v1.User{})
    31. 	gob.Register(v2.User{})
    32.  
    33. 	// 创建服务发现
    34. 	sd, err := rpc.CreateServiceDiscovery(*basePath, *registry, "", 0, 100)
    35. 	if err != nil {
    36. 		log.Fatal(err)
    37. 	}
    38.  
    39. 	// 创建服务端
    40. 	srv := rpc.NewServer(*addr, sd)
    41. 	// 将服务端方法，注册一下
    42. 	//srv.Register(new(service.Test))
    43. 	srv.RegisterName(new(service.Test), "service")
    44. 	srv.RegisterName(new(v1.Test), "v1")
    45. 	srv.RegisterName(new(v2.Test), "v2")
    46. 	// 启动服务
    47. 	srv.Serve()
    48. }

    快速开始 - 客户端

    1. package main
    2.  
    3. import (
    4. 	"encoding/gob"
    5. 	"flag"
    6. 	"fmt"
    7. 	"log"
    8. 	"sync"
    9. 	"time"
    10. 	"zrpc/example/service"
    11. 	v1 "zrpc/example/v1"
    12. 	v2 "zrpc/example/v2"
    13. 	"zrpc/rpc"
    14. 	"zrpc/rpc/center"
    15. )
    16.  
    17. var (
    18. 	registry = flag.String("registry", "redis://127.0.0.1:6379", "registry address")
    19. 	basePath = flag.String("basepath", "/zrpc_center", "")
    20.     cli *rpc.Client
    21. )
    22.  
    23. // Args 参数
    24. type Args struct {
    25. 	Id int64
    26. 	X  int64
    27. 	Y  int64
    28. 	Z  string
    29. }
    30.  
    31. // 自己定义数据格式的读写
    32. func main() {
    33. 	// 解析参数
    34. 	if !flag.Parsed() {
    35. 		flag.Parse()
    36. 	}
    37.  
    38. 	// 发现服务
    39. 	sd, err := rpc.ServiceDiscovery(*basePath, *registry, "", 0, 100)
    40. 	if err != nil {
    41. 		log.Fatal(err)
    42. 	}
    43.  
    44. 	// 创建客户端
    45. 	if cli == nil {
    46. 		cli = rpc.NewClient(sd, center.SelectMode(center.Random), true)
    47. 		defer closeCli()
    48. 	}
    49.  
    50. 	// 同步rpc
    51. 	var reply any
    52. 	// 参数 struct 格式
    53. 	str := "我是rpc测试参数！！！"
    54. 	args := Args{
    55. 		Id: 2,
    56. 		X:  20,
    57. 		Z:  str,
    58. 	}
    59. 	errC := cli.Call("service.QueryUser", args, &reply)
    60. 	if errC != nil {
    61. 		fmt.Println("main.call.errC", errC)
    62. 	}
    63. 	reply1 := reply.(map[string]any)
    64. 	fmt.Println("main.call.reply", reply1["Age"])
    65.  
    66. 	fmt.Println("==========================================")
    67.  
    68. 	// 异步rpc
    69. 	var reply2 any
    70. 	call2 := cli.Go("v1.QueryInt", map[string]any{"Id": 10000, "msg": str}, &reply2, nil)
    71. 	<-call2.Done
    72. 	if call2.Error != nil {
    73. 		fmt.Printf("main.go.reply2.error: %v \n", call2.Error)
    74. 	}
    75. 	fmt.Printf("main.go.reply2: %v \n", reply2)
    76.  
    77. 	time.Sleep(2 * time.Second)
    78. }
    79.  
    80.  
    81.

    快速开始 - php客户端

    1. (new TestController())->actionTest()
    2.  
    3. class TestController extends Controller
    4. {
    5.     const HEAD_MSG = "@**@";
    6.     public $socket;
    7.  
    8.     // 析构函数
    9.     public function __destruct() {
    10.         socket_close($this->socket);
    11.     }
    12.  
    13.     public function newRpc($host, $port) {
    14.         $this->socket = socket_create(AF_INET, SOCK_STREAM, SOL_TCP) or die("Unable to create socket");
    15.         @socket_connect($this->socket, $host, $port) or die("Connect error.");
    16.         if ($err = socket_last_error($this->socket)) {
    17.             socket_close($this->socket);
    18.             die(socket_strerror($err));
    19.         }
    20.     }
    21.  
    22.     public function build($method, $data) {
    23.         $response = [
    24.             "ServiceMethod" => $method,
    25.             "Args" => $data,
    26.             "Reply" => null,
    27.         ];
    28.         return $response;
    29.     }
    30.  
    31.     public function write($response) {
    32.         $response = msgpack_pack($response);
    33.         $len = strlen($response);
    34.         $buf = pack("a4", self::HEAD_MSG);
    35.         $buf .= pack("N", $len);
    36.         $buf .= pack("a".$len, $response);
    37.         return socket_write ($this->socket , $buf, strlen($buf));
    38.     }
    39.  
    40.     public function read() {
    41.         // 读取数据
    42.         $hm = socket_read($this->socket, 4, PHP_BINARY_READ);
    43.         if ($hm != self::HEAD_MSG) {
    44.             print_r("HEAD_MSG ERROR: " . $hm);
    45.         }
    46.         $hLen = socket_read($this->socket, 4, PHP_BINARY_READ);
    47.         $hLen = unpack("N", $hLen);
    48.         $hLen = $hLen[1];
    49.         $buffer = socket_read($this->socket, $hLen, PHP_BINARY_READ);
    50.         $buffer = msgpack_unpack($buffer);
    51.         return $buffer;
    52.     }
    53.  
    54.     public function call($api, $data) {
    55.         // 打包数据
    56.         $buf = $this->build($api, $data);
    57.  
    58.         // 发送数据
    59.         $this->write($buf);
    60.  
    61.         // 读取数据
    62.         $buffer = $this->read();
    63.         return $buffer;
    64.     }
    65.  
    66.     // PHP 调用 zrpc 服务
    67.     public function actionTest()
    68.     {
    69.         // 连接服务
    70.         $this->newRpc("127.0.0.1", "8092");
    71.  
    72.         // 数据
    73.         $data["Id"] = 1;
    74.         $data["X"] = 20;
    75.         $data["Z"] = "aaasssdddfffggghhh";
    76.         $data["msg"] = "msg000";
    77.  
    78.         // 接口
    79.         $api[] = "service.QueryUser";
    80.         $api[] = "v1.QueryInt";
    81.  
    82.         // 调用
    83.         for ($i=0; $i<10000; $i++) {
    84.             $rand = mt_rand(0, 1);
    85.             $buffer = $this->call($api[$rand], $data);
    86.             var_dump(json_encode($buffer));
    87.         }
    88.  
    89.     }
    90. }

    快速开始 - 网关

    1、启动网关

    1. package main
    2.  
    3. import (
    4. 	"flag"
    5. 	"log"
    6. 	"zrpc/rpc"
    7. )
    8.  
    9. var (
    10. 	addr     = flag.String("addr", "127.0.0.1:8060", "addr server")
    11. 	registry = flag.String("registry", "redis://127.0.0.1:6379", "registry address")
    12. 	basePath = flag.String("basepath", "/zrpc_center", "")
    13. )
    14.  
    15. func main() {
    16. 	// 解析参数
    17. 	if !flag.Parsed() {
    18. 		flag.Parse()
    19. 	}
    20. 	// http new
    21. 	http := rpc.NewHttp(*addr)
    22. 	// 发现服务
    23. 	sd, err := rpc.ServiceDiscovery(*basePath, *registry, "", 0, 100)
    24. 	if err != nil {
    25. 		log.Fatal(err)
    26. 	}
    27. 	// 注册服务
    28. 	router := http.RegServe(sd)
    29. 	// 启动http服务
    30. 	http.HttpServer(router)
    31. }

    2、客户端连接网关 php例子

    1.  
    2.     $url = 'http://localhost:8060/';
    3.     $data['servicePath'] = 'v1';
    4.     $data['serviceMethod'] = 'queryInt';
    5.     $data['content'] = '{"Id": 10000, "msg": "测试api"}';
    6.  
    7.     $this->curl = new CurlRequest($url);
    8.     $this->curl->setPostFeilds($data);
    9.     $data = $this->curl->post();
    10.     $data = json_decode($data, true);
    11.     var_dump($data);
    12.  
    13.     // curl
    14.     class CurlRequest
    15.     {
    16.         public $version = "1.0";
    17.  
    18.         public $handler;
    19.  
    20.         public $timeOut = 30;
    21.  
    22.         public $header = ['Expect:'];
    23.  
    24.         public $referer;
    25.  
    26.         public $postFields;
    27.  
    28.         public $url;
    29.  
    30.         public $ssl = false;
    31.  
    32.         public $type = "get";
    33.  
    34.         public $agent = "see curl/1.0";
    35.  
    36.         public $returnData;
    37.  
    38.         public function __construct($url)
    39.         {
    40.             $this->url = $url = trim($url);
    41.             $this->handler = curl_init();
    42.             curl_setopt($this->handler, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_0);
    43.             curl_setopt($this->handler, CURLOPT_RETURNTRANSFER, true);
    44.             curl_setopt($this->handler, CURLOPT_URL, $url);
    45.             //支持https
    46.             $this->ssl = stripos($url, 'https://') === 0 ? true : false;
    47.             if ($this->ssl) {
    48.                 curl_setopt($this->handler, CURLOPT_SSL_VERIFYPEER, false);
    49.                 curl_setopt($this->handler, CURLOPT_SSL_VERIFYHOST, false);
    50.             }
    51.         }
    52.  
    53.         public function setTimeOut($timeOut){
    54.             $this->timeOut = $timeOut;
    55.         }
    56.  
    57.         public function setOpt($opt, $value)
    58.         {
    59.             curl_setopt($this->handler, $opt, $value);
    60.         }
    61.  
    62.         public function setHeader($header)
    63.         {
    64.             $this->header = array_merge($this->header, $header);
    65.         }
    66.  
    67.         public function setCookie($cookie)
    68.         {
    69.             curl_setopt($this->handler, CURLOPT_COOKIE, $cookie);
    70.         }
    71.  
    72.         public function setReferer($referer)
    73.         {
    74.             $this->referer = $referer;
    75.         }
    76.  
    77.         public function setPostFeilds($postFields)
    78.         {
    79.             $this->postFields = $postFields;
    80.         }
    81.  
    82.         public function setAgent($agent){
    83.             $this->agent = $agent;
    84.         }
    85.  
    86.         public function exec()
    87.         {
    88.             if (!empty($this->referer)) {
    89.                 curl_setopt($this->handler, CURLOPT_REFERER, $this->referer);
    90.             }else{
    91.                 curl_setopt($this->handler, CURLOPT_AUTOREFERER, true);
    92.             }
    93.  
    94.             //set header
    95.             $this->setHeader(['PHP-SEE-TID:' . 1]);
    96.             $this->setHeader(['PHP-SEE-SEQ:' . 2]);
    97.  
    98.             curl_setopt($this->handler,CURLOPT_HTTPHEADER,$this->header);
    99.             curl_setopt($this->handler, CURLOPT_USERAGENT,$this->agent);
    100.             curl_setopt($this->handler, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_0);
    101.             curl_setopt($this->handler, CURLOPT_IPRESOLVE, CURL_IPRESOLVE_V4);
    102.             curl_setopt($this->handler, CURLOPT_CONNECTTIMEOUT, $this->timeOut);
    103.             curl_setopt($this->handler, CURLOPT_TIMEOUT, $this->timeOut);
    104.  
    105.             if($this->type=='post'){
    106.                 curl_setopt($this->handler, CURLOPT_POST, true);
    107.                 curl_setopt($this->handler, CURLOPT_POSTFIELDS, $this->postFields);
    108.             }
    109.  
    110.             $this->returnData = curl_exec($this->handler);
    111.             $httpcode = curl_getinfo($this->handler, CURLINFO_HTTP_CODE);
    112.             if ($errorNo = curl_errno($this->handler) || $httpcode != 200) {
    113.                 //error message
    114.                 $errorMsg = curl_error($this->handler);
    115.                 \See::$log->warning("curl error, url:%s, type:%s, postData:%s, errorNo:%s, errorMsg:%s, httpcode:%s, return:%s",$this->url,$this->type,$this->postFields,$errorNo,$errorMsg,$httpcode,$this->returnData);
    116.             }else{
    117.                 \See::$log->trace("curl success, url:%s, type:%s, postData:%s, httpcode:%s, return:%s",$this->url,$this->type,$this->postFields,$httpcode,$this->returnData);
    118.             }
    119.             curl_close($this->handler);
    120.             return $this->returnData;
    121.         }
    122.  
    123.         public function get(){
    124.             $this->type = 'get';
    125.             return $this->exec();
    126.         }
    127.  
    128.         public function post(){
    129.             $this->type ='post';
    130.             return $this->exec();
    131.         }
    132.  
    133.     }
  • 相关阅读:
    如何设计金融机构多场景关键应用下的存储架构
    Hikari连接池2--获取和归还连接
    IO模型5-进阶-Netty模型
    1832javaERP管理系统之车间计划管理Myeclipse开发mysql数据库servlet结构java编程计算机网页项目
    linux espidf vscode
    nodejs+vue菜谱美食食谱网站系统
    前端基础建设与架构18 对比 Koa 和 Redux：分析前端中的中间件理念
    Redis - 高级
    苍穹外卖——项目搭建
    springboot+vue+elementUI 健身房私人健身与教练预约管理系统 #毕业设计
  • 原文地址：https://blog.csdn.net/qq_37200336/article/details/126564093