README

FiberPHP JSON-RPC 是一个 JSON-RPC 2.0 远程过程调用框架，包含契约接口、打包器、TCP/HTTP 客户端与服务端、批量调用、异步客户端和注解自动注册。

架构总览

┌─────────────────────────────────────────────────────────┐
│                     业务层（调用方）                       │
│        $client->call()  /  asyncCall()  /  batchCall()  │
├─────────────────────────────────────────────────────────┤
│              ClientInterface（协议驱动实现）               │
│     AsyncTcpClient  │  TcpClient  │  HttpClient          │
├─────────────────────────────────────────────────────────┤
│              PackerInterface（协议层打包器）               │
│              JsonRpcPacker  │  JsonPacker                 │
├─────────────────────────────────────────────────────────┤
│              传输层（TCP / HTTP）                         │
│   AsyncTcpConnection  │  stream_socket  │  cURL           │
╞═════════════════════════════════════════════════════════╡
│                      网络边界                             │
╞═════════════════════════════════════════════════════════╡
│              传输层（TCP / HTTP）                         │
│    Workerman Worker  │  PHP-FPM / SAPI                   │
├─────────────────────────────────────────────────────────┤
│              ServerInterface（协议驱动实现）               │
│           TcpServer  │  HttpServer                       │
├─────────────────────────────────────────────────────────┤
│              ServiceContainer（服务路由）                  │
│        服务名 → 实例映射  │  方法校验  │  反射调用          │
├─────────────────────────────────────────────────────────┤
│                     业务层（服务端）                       │
│           GoodsService  │  StockService  │  ...           │
└─────────────────────────────────────────────────────────┘

包结构

包名	职责
`fiberphp/jsonrpc`	JSON-RPC 2.0 全栈：契约接口、打包器、TCP/HTTP 客户端与服务端、批量调用、异步客户端、注解自动注册

核心概念

RpcRequest / RpcResponse

协议无关的请求/响应上下文对象，贯穿中间件管线：

// 请求对象
$request = new RpcRequest(
    service: 'Goods',           // 目标服务名
    method: 'getGoods',         // 目标方法名
    params: [1],                // 调用参数
    metadata: ['traceId' => 'xxx'], // 元数据（traceId、鉴权等）
    id: 123,                    // 请求 ID（用于匹配响应）
);

// 响应对象
$response = RpcResponse::success($result, $id);
$response = RpcResponse::error(-32601, 'Service not found', $id);

ServiceInterface

服务标记接口。所有对外暴露的 RPC 服务类需实现此接口：

use FiberPHP\Rpc\Contract\ServiceInterface;

class GoodsService implements ServiceInterface
{
    public function getGoods(int $id): array
    {
        return ['id' => $id, 'name' => 'iPhone 16 Pro'];
    }
}

ServiceContainer

服务注册表，负责服务名 → 实例映射、方法合法性校验、反射调用：

$container = new ServiceContainer();
$container->add('Goods', new GoodsService());      // 按名注册
$container->addInstance(new GoodsService());        // 自动注册（短名 Goods + 全限定类名）

// 调用
$result = $container->call('Goods', 'getGoods', [1]);

`#[RpcService]` 注解自动注册

在服务类上标注 #[RpcService] 注解，由 ServiceScanner 自动发现并注册，无需手动调用 addService()：

use FiberPHP\Rpc\Attribute\RpcService;
use FiberPHP\Rpc\Contract\ServiceInterface;

#[RpcService]
class GoodsService implements ServiceInterface
{
    public function getGoods(int $id): array { ... }
}

#[RpcService(name: 'Stock', middleware: [AuthMiddleware::class])]
class StockService implements ServiceInterface
{
    public function getStock(int $id): array { ... }
}

服务名推断规则：

name 为 null 时，取类短名并去掉 Service 后缀（GoodsService → Goods）
同时注册全限定类名作为别名，支持两种调用方式

自动扫描机制：

在 config/rpc/app.php 中配置扫描目录：

return [
    'service_scan_dirs' => [
        'App\\Rpc\\Service\\' => app_path('Rpc/Service'),
    ],
];

RpcProvider::boot() 阶段自动扫描这些目录，发现带 #[RpcService] 注解的类，通过容器实例化（支持构造函数 DI）后注册到 ServiceContainer。TcpServer / HttpServer 共享同一个 ServiceContainer 引用，扫描注册后自动可见。

手动扫描（独立脚本）：

use FiberPHP\JsonRpc\Server\TcpServer;
use FiberPHP\Rpc\Service\ServiceScanner;

$server = new TcpServer();
$scanner = new ServiceScanner($server->getServiceContainer());
$scanner->scan([
    'App\\Rpc\\Service\\' => __DIR__ . '/app/Rpc/Service',
]);
$server->start('0.0.0.0', 9501);

方法校验规则：

方法必须存在且为 public
魔术方法（__ 开头）禁止调用
服务不存在 → 错误码 -32601
方法不存在 → 错误码 -32601
方法不可访问 → 错误码 -32602
调用异常 → 错误码 -32603

中间件

采用洋葱模型，与 fiberphp/pipeline 兼容，可用于鉴权、日志、限流、Trace：

use FiberPHP\Rpc\Contract\MiddlewareInterface;
use Closure;

class AuthMiddleware implements MiddlewareInterface
{
    public function handle(object $request, Closure $next): mixed
    {
        $token = $request->getMetadata('token');
        if ($token === null) {
            throw new RpcException('Unauthorized', -32001);
        }
        // 验证 token...
        return $next($request);
    }
}

// 注册到客户端或服务端
$client->withMiddleware(new AuthMiddleware());
$server->withMiddleware(new AuthMiddleware());

快速开始

1. 创建 RPC 服务

<?php

declare(strict_types=1);

namespace App\Rpc\Service;

use FiberPHP\Rpc\Attribute\RpcService;
use FiberPHP\Rpc\Contract\ServiceInterface;

#[RpcService]
class GoodsService implements ServiceInterface
{
    public function getGoods(int $id): array
    {
        return ['id' => $id, 'name' => 'iPhone 16 Pro', 'price' => 8999.00];
    }

    public function listGoods(array $ids = []): array
    {
        // ...
    }

    public function getByCategory(string $category): array
    {
        // ...
    }
}

2. 创建 RPC 服务端

#!/usr/bin/env php
<?php

declare(strict_types=1);

use FiberPHP\JsonRpc\Server\TcpServer;
use FiberPHP\Rpc\Service\ServiceScanner;

define('BASE_PATH', __DIR__);

require_once __DIR__ . '/vendor/autoload.php';

$server = new TcpServer();

// 注解自动扫描注册（替代手动 addService）
$scanner = new ServiceScanner($server->getServiceContainer());
$scanner->scan([
    'App\\Rpc\\Service\\' => __DIR__ . '/app/Rpc/Service',
]);

// 可选：添加全局中间件
// $server->withMiddleware(new AuthMiddleware());

$server->start('0.0.0.0', 9501, [
    'worker_num' => 2,
    'name'       => 'goods-service',
]);

3. 创建 RPC 客户端

use FiberPHP\JsonRpc\Client\AsyncTcpClient;

// 同步调用（Fiber 协程中自动挂起/恢复）
$client = new AsyncTcpClient();
$client->to('127.0.0.1', 9501);
$goods = $client->call('Goods', 'getGoods', [1]);

// 异步调用（立即返回 Channel，不阻塞）
$ch = $client->asyncCall('Goods', 'getGoods', [1]);
// ... 执行其他逻辑 ...
$response = $ch->pop();  // 按需获取结果

// 批量调用（一次 TCP 往返）
$results = $client->batchCall([
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [1]],
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [2]],
    ['service' => 'Goods', 'method' => 'getByCategory', 'params' => ['手机']],
]);

三种调用模式

同步调用 `call()`

在 Fiber 协程中自动挂起当前协程，响应到达后恢复。代码风格为同步阻塞，底层不阻塞 Worker。

$client = new AsyncTcpClient();
$client->to('127.0.0.1', 9501);

$goods  = $client->call('Goods', 'getGoods', [1]);  // 挂起 → 恢复
$stock  = $client->call('Stock', 'getStock', [1]);  // 挂起 → 恢复
// 总耗时 = 请求1 + 请求2（串行）

异步调用 `asyncCall()`

立即返回 Channel，不阻塞当前协程。多个请求可并行发起，总耗时 ≈ max(单请求)。

$client = new AsyncTcpClient();
$client->to('127.0.0.1', 9501);

// 并行发起 3 个请求
$ch1 = $client->asyncCall('Goods', 'getGoods', [1]);
$ch2 = $client->asyncCall('Goods', 'getGoods', [2]);
$ch3 = $client->asyncCall('Goods', 'listGoods', [[1, 2, 3]]);

// 此处可执行其他业务逻辑...

// 按需获取结果（pop() 会自动挂起直到响应到达）
$r1 = $ch1->pop()->result;
$r2 = $ch2->pop()->result;
$r3 = $ch3->pop()->result;
// 总耗时 ≈ max(请求1, 请求2, 请求3)

批量调用 `batchCall()`

一次 TCP 往返发送多个请求到同一服务端，符合 JSON-RPC 2.0 批量规范。适用于从同一服务获取多条数据。

$client = new AsyncTcpClient();
$client->to('127.0.0.1', 9501);

$results = $client->batchCall([
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [1]],
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [2]],
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [3]],
    ['service' => 'Goods', 'method' => 'getByCategory', 'params' => ['手机']],
]);
// $results 是数组，顺序与请求一致
// 仅一次 TCP 往返

混合调用

在同一方法中组合三种模式：

$goodsClient = new AsyncTcpClient();
$goodsClient->to('127.0.0.1', 9501);

$stockClient = new AsyncTcpClient();
$stockClient->to('127.0.0.1', 9502);

// 1. 异步发起库存查询（不等待）
$chStock = $stockClient->asyncCall('Stock', 'getStock', [2]);

// 2. 批量获取商品信息（一次往返）
$goodsBatch = $goodsClient->batchCall([
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [1]],
    ['service' => 'Goods', 'method' => 'getGoods', 'params' => [2]],
]);

// 3. 同步调用创建新商品
$newGoods = $goodsClient->call('Goods', 'createGoods', ['iPad Pro M4', 6999.00, '平板']);

// 4. 获取异步库存结果
$stock = $chStock->pop()->result;

客户端选择指南

客户端	传输	适用场景	异步	批量
`AsyncTcpClient`	TCP（Workerman）	Workerman 事件循环内（HTTP 服务、Queue 消费者）	支持	支持
`TcpClient`	TCP（同步 Socket）	CLI 脚本、非 Workerman 环境	不支持	支持
`HttpClient`	HTTP（cURL）	需通过 HTTPS 网关、Nginx 代理访问	不支持	支持

推荐：在 Workerman 应用中始终使用 AsyncTcpClient，它同时支持同步、异步和批量三种调用模式。

服务端选择指南

服务端	传输	适用场景
`TcpServer`	TCP（Workerman）	高性能内部服务间通信
`HttpServer`	HTTP（SAPI/FPM）	需通过 HTTP 网关暴露、或 PHP-FPM 部署

JSON-RPC 2.0 协议格式

单个请求

{
    "jsonrpc": "2.0",
    "method": "Goods.getGoods",
    "params": [1],
    "id": 1
}

单个响应（成功）

{
    "jsonrpc": "2.0",
    "result": {"id": 1, "name": "iPhone 16 Pro", "price": 8999},
    "id": 1
}

单个响应（失败）

{
    "jsonrpc": "2.0",
    "error": {"code": -32601, "message": "RPC service \"Foo\" not found"},
    "id": 1
}

批量请求

[
    {"jsonrpc": "2.0", "method": "Goods.getGoods", "params": [1], "id": 1},
    {"jsonrpc": "2.0", "method": "Goods.getGoods", "params": [2], "id": 2},
    {"jsonrpc": "2.0", "method": "Goods.getByCategory", "params": ["手机"], "id": 3}
]

批量响应

[
    {"jsonrpc": "2.0", "result": {"id": 1, "name": "iPhone 16 Pro"}, "id": 1},
    {"jsonrpc": "2.0", "result": {"id": 2, "name": "MacBook Air M4"}, "id": 2},
    {"jsonrpc": "2.0", "result": [{"id": 1, "name": "iPhone 16 Pro"}], "id": 3}
]

TCP 分包

TCP 传输使用 \r\n（EOF）作为消息分隔符：

{"jsonrpc":"2.0","method":"Goods.getGoods","params":[1],"id":1}\r\n

错误码

错误码	含义
`-32700`	Parse error（JSON 解析失败）
`-32600`	Invalid Request（无效请求）
`-32601`	Method not found（服务/方法不存在）
`-32602`	Invalid params（方法不可访问）
`-32603`	Internal error（调用异常）
`-32004`	Connect error（连接失败，异步客户端）

API 参考

ClientInterface

方法	说明
`to(string $host, int $port): self`	设置目标服务端地址
`call(string $service, string $method, array $params = [], array $options = []): mixed`	同步调用，返回结果
`batchCall(array $calls): array`	批量调用，返回结果数组
`withPacker(PackerInterface $packer): self`	设置打包器
`withMiddleware(MiddlewareInterface\\|callable $m): self`	追加中间件
`withTimeout(int $connect, int $recv): self`	设置超时

AsyncTcpClient（扩展方法）

方法	说明
`asyncCall(string $service, string $method, array $params = [], array $options = []): Channel`	异步调用，返回 Channel
`asyncBatchCall(array $calls): Channel`	异步批量调用，返回 Channel

ServerInterface

方法	说明
`start(string $host, int $port, array $options = []): void`	启动服务
`addService(string $name, object $service): self`	手动注册服务
`addInstance(object $service): self`	按类名自动注册
`setServiceContainer(ServiceContainer $c): self`	共享外部 ServiceContainer
`getServiceContainer(): ServiceContainer`	获取当前 ServiceContainer
`withPacker(PackerInterface $packer): self`	设置打包器
`withMiddleware(MiddlewareInterface\\|callable $m): self`	追加中间件

ServiceScanner

方法	说明
`scan(array $namespaceMaps): int`	扫描目录，返回注册数量
`scanDirectory(string $namespace, string $dir): int`	扫描单个目录

`#[RpcService]` 注解参数

参数	类型	默认值	说明
`name`	string\|null	null	服务名（null 时自动推断）
`middleware`	array	`[]`	服务级中间件

TcpServer 启动选项

选项	类型	默认值	说明
`worker_num`	int	4	Worker 进程数
`name`	string	`fiberphp-jsonrpc-tcp`	进程名
`context`	array	`[]`	SSL 上下文选项
`on_worker_stop`	callable	null	Worker 停止回调

完整示例

项目内已包含完整的示例项目：

项目	角色	端口
`goods-service/`	RPC 服务端（商品服务）	9501
`stock-service/`	RPC 服务端（库存服务）	9502
`http/`	RPC 客户端（HTTP API）	8787

测试端点

# 同步调用
curl http://127.0.0.1:8787/rpc/sync

# 异步调用
curl http://127.0.0.1:8787/rpc/async

# 批量调用
curl http://127.0.0.1:8787/rpc/batch

# 混合调用
curl http://127.0.0.1:8787/rpc/mixed

启动服务

# 启动 RPC 服务端
cd goods-service && php start.php start
cd stock-service && php start.php start

# 启动 HTTP 客户端
cd http && php start.php start

fiberphp / jsonrpc

Maintainers

Package info

Statistics

Security