在hyperf框架中使用基于protobuf的RPC生成器实现rpc服务

swoole  /  管理员 发布于 2年前   840

ROC RPC PHP演示git

https://github.com/Gemini-D/roc-rpc-php-demo

环境架构：

hyperf2.2
ROCGenerator

安装RPC生成工具

有两种安装方式

1.使用源码

2.使用打包好的二进制文件

git:

https://github.com/hyperf/roc-generator

编写rpc.proto文件

syntax = "proto3";

option php_namespace = "ROC\\RPC";

package rpc;

service UserInterface {
 rpc info(UserInput) returns (User) {}}

message UserInput{
 uint64 id = 1;}

message User {
 uint64 id = 1; string name = 2; uint32 gender = 3;}

根据文件生成代码

cd rpc
roc-php gen:roc rpc.proto -O src

接下来，我们就可以在 rpc/src 目录下看到了生成的文件。

完善自定义rpc组件代码

正常情况下，我们需要按照官网配置，增加对应配置文件，但这样会比较麻烦，

所以我们通过事件的方式，将对应配置自动加载进来

编写监听器

增加rpc/src/Listener/BootConsumerListener

<?php

declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */namespace ROC\RPC\Listener;

use Hyperf\Contract\ConfigInterface;
use Hyperf\Event\Contract\ListenerInterface;
use Hyperf\Framework\Event\BootApplication;
use Hyperf\RpcMultiplex\Constant;
use Psr\Container\ContainerInterface;
use ROC\RPC\UserInterface;

/**
 * Must handle the event before `Hyperf\RpcClient\Listener\AddConsumerDefinitionListener`. * You can set ROC\RPC\Listener\BootConsumerListener::class => 99. */class BootConsumerListener implements ListenerInterface
{
 /** * @var ContainerInterface */ private $container;
 public function __construct(ContainerInterface $container) { $this->container = $container; }
 public function listen(): array { return [ BootApplication::class, ]; }
 public function process(object $event): void { $config = $this->container->get(ConfigInterface::class);
 $consumers = $config->get('services.consumers', []); $services = [ UserInterface::class => ['127.0.0.1', 9504], ];
 foreach ($services as $interface => [$host, $port]) { $consumers[] = static::getDefaultConsumer($interface, $host, $port); }
 $config->set('services.consumers', $consumers); }
 public static function getDefaultConsumer(string $interface, string $host, int $port): array { return [ 'name' => $interface, 'service' => $interface, 'id' => $interface, 'protocol' => Constant::PROTOCOL_DEFAULT, 'load_balancer' => 'random', 'nodes' => [ ['host' => $host, 'port' => $port], ], 'options' => [ 'connect_timeout' => 5.0, 'recv_timeout' => 5.0, 'settings' => [ // 包体最大值，若小于 Server 返回的数据大小，则会抛出异常，故尽量控制包体大小
 'package_max_length' => 1024 * 1024 * 2, ], // 重试次数，默认值为 2 'retry_count' => 2, // 重试间隔，毫秒
 'retry_interval' => 10, // 多路复用客户端数量
 'client_count' => 4, // 心跳间隔
 'heartbeat' => 20, ], ]; }}

编写RPC服务端代码

增加对应仓库

首先我们现在 server/composer.json 中增加对应的仓库配置

{
 "repositories": { "rpc": { "type": "path", "url": "../rpc" } }}

接下来再通过执行脚本，载入对应组件包

composer require roc/rpc

导入RPC相关组件

composer require "hyperf/rpc-client:3.0.x-dev" -W
composer require "hyperf/rpc:3.0.x-dev" -W
composer require "hyperf/rpc-multiplex:3.0.x-dev" -W
composer require "hyperf/rpc-server:3.0.x-dev" -W

ps：

因为相关的代码，还没有发布Release版本，所以需要导入x-dev包

增加RPC服务配置

让我们修改config/server.php文件，增加rpc相关配置

<?php

declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */use Hyperf\Engine\Constant\SocketType;
use Hyperf\Server\Event;
use Hyperf\Server\Server;

return [
 'mode' => SWOOLE_BASE, 'type' => Hyperf\Server\CoroutineServer::class, 'servers' => [ [ 'name' => 'http', 'type' => Server::SERVER_HTTP, 'host' => '0.0.0.0', 'port' => 9501, 'sock_type' => SocketType::TCP, 'callbacks' => [ Event::ON_REQUEST => [Hyperf\HttpServer\Server::class, 'onRequest'], ], ], [ 'name' => 'rpc', 'type' => Server::SERVER_BASE, 'host' => '0.0.0.0', 'port' => 9504, 'sock_type' => SWOOLE_SOCK_TCP, 'callbacks' => [ Event::ON_RECEIVE => [Hyperf\RpcMultiplex\TcpServer::class, 'onReceive'], ], 'settings' => [ 'open_length_check' => true, 'package_length_type' => 'N', 'package_length_offset' => 0, 'package_body_offset' => 4, 'package_max_length' => 1024 * 1024 * 2, ], ], ], 'settings' => [ 'enable_coroutine' => true, 'worker_num' => 4, 'pid_file' => BASE_PATH . '/runtime/hyperf.pid', 'open_tcp_nodelay' => true, 'max_coroutine' => 100000, 'open_http2_protocol' => true, 'max_request' => 0, 'socket_buffer_size' => 2 * 1024 * 1024, 'package_max_length' => 2 * 1024 * 1024, ], 'callbacks' => [ Event::ON_BEFORE_START => [Hyperf\Framework\Bootstrap\ServerStartCallback::class, 'beforeStart'], Event::ON_WORKER_START => [Hyperf\Framework\Bootstrap\WorkerStartCallback::class, 'onWorkerStart'], Event::ON_PIPE_MESSAGE => [Hyperf\Framework\Bootstrap\PipeMessageCallback::class, 'onPipeMessage'], Event::ON_WORKER_EXIT => [Hyperf\Framework\Bootstrap\WorkerExitCallback::class, 'onWorkerExit'], ],];

增加服务实现

新增app/RPC/UserService.php 文件，增加以下代码

<?php
declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */namespace App\RPC;
use Hyperf\RpcMultiplex\Constant;
use Hyperf\RpcServer\Annotation\RpcService;
use ROC\RPC\User;
use ROC\RPC\UserInput;
use ROC\RPC\UserInterface;
#[RpcService(name: UserInterface::class, server: 'rpc', protocol: Constant::PROTOCOL_DEFAULT)]
class UserService implements UserInterface
{
 public function info(UserInput $input): User { return new User( $input->id, 'Hyperf', 1 ); }
}

配置监听器

修改server/config/autoload/listeners.php

<?php
declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */return [
 Hyperf\ExceptionHandler\Listener\ErrorExceptionHandler::class, Hyperf\Command\Listener\FailToHandleListener::class, ROC\RPC\Listener\BootConsumerListener::class => 99,];

测试RPC

编写RegisterProtocolListener

因为默认的RPC组件，使用的是默认的Normalizer，而代码生成的接口文件，

是需要支持Json转Object，所以我们需要进行替换

<?php
declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */namespace App\Listener;
use Hyperf\Event\Annotation\Listener;
use Hyperf\Event\Contract\ListenerInterface;
use Hyperf\Framework\Event\BootApplication;
use Hyperf\Rpc\ProtocolManager;
use Hyperf\RpcMultiplex\Constant;
use Hyperf\RpcMultiplex\DataFormatter;
use Hyperf\RpcMultiplex\Packer\JsonPacker;
use Hyperf\RpcMultiplex\PathGenerator;
use Hyperf\RpcMultiplex\Transporter;
use Hyperf\Utils\Serializer\JsonDeNormalizer;
use Psr\Container\ContainerInterface;
#[Listener(priority: -1)]
class RegisterProtocolListener implements ListenerInterface
{
 public function __construct(protected ContainerInterface $container) { }
 public function listen(): array { return [ BootApplication::class, ]; }
 public function process(object $event): void { $this->container->get(ProtocolManager::class)->register(Constant::PROTOCOL_DEFAULT, [ 'packer' => JsonPacker::class, 'transporter' => Transporter::class, 'path-generator' => PathGenerator::class, 'data-formatter' => DataFormatter::class, 'normalizer' => JsonDeNormalizer::class, ]); }}

编写测试代码

<?php
declare(strict_types=1);
/**
 * This file is part of Hyperf. * * @link     https://www.hyperf.io * @document https://hyperf.wiki * @contact  [email protected] * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE */namespace App\Controller;
use Hyperf\Di\Annotation\Inject;
use ROC\RPC\UserInput;
use ROC\RPC\UserInterface;
class IndexController extends Controller
{
 #[Inject] protected UserInterface $user;
 public function index() { $result = $this->user->info(new UserInput(1));
 return $this->response->success($result); }}

启动服务

php bin/hyperf.php start

访问接口

curl http://127.0.0.1:9501/
{"code":0,"data":{"id":1,"name":"Hyperf","gender":1}}