Yar RPC 框架

Yar（Yet Another RPC Framework）是鸟哥（Laruence）开发的 PHP RPC 框架，专为高性能 PHP 服务间通信设计。它使用 JSON 或 MSGPACK 作为序列化格式，基于 HTTP 协议传输，支持并行调用、持久连接等特性，比 XML-RPC 和 SOAP 更加高效。

前置知识

阅读本节前，建议先了解：cURL 详解、JSON 编解码

基础概念

什么是 Yar

Yar 是一个轻量级、高性能的 PHP RPC 框架，具有以下特点：

高性能：使用 JSON/MSGPACK 序列化，传输体积小、解析快
并行调用：支持同时调用多个远程服务
持久连接：支持 HTTP Keep-Alive 长连接
自定义协议：可扩展传输协议和打包器
简单易用：只需定义接口，无需编写通信代码

Yar 架构

Client                          Server
  |                               |
  |--- HTTP POST (JSON/MSGPACK) -->|
  |                               |--- 反序列化请求
  |                               |--- 调用本地方法
  |                               |--- 序列化响应
  |<-- HTTP Response (JSON/MSGPACK) -|
  |--- 反序列化响应               |

安装

bash

# PECL 安装
pecl install yar

# 编译安装
git clone https://github.com/laruence/yar.git
cd yar
phpize
./configure
make && make install

# php.ini 添加
extension=yar.so

Yar 客户端

基本调用

php

<?php
declare(strict_types=1);

/**
 * Yar 客户端基本用法
 */

// 方式一：直接调用
$client = new Yar_Client('http://api.example.com/service.php');
$result = $client->getUserInfo(123);

// 方式二：使用魔术方法
$client = new Yar_Client('http://api.example.com/service.php');
$result = $client->getUserInfo(['userId' => 123]);

// 方式三：使用 call 方法
$client = new Yar_Client('http://api.example.com/service.php');
$result = $client->call('getUserInfo', [123]);

配置选项

php

<?php
declare(strict_types=1);

$client = new Yar_Client('http://api.example.com/service.php');

// 设置超时（毫秒）
$client->SetOpt(YAR_OPT_CONNECT_TIMEOUT, 5000);  // 连接超时 5 秒
$client->SetOpt(YAR_OPT_TIMEOUT, 10000);         // 执行超时 10 秒

// 设置打包器（JSON 或 MSGPACK）
$client->SetOpt(YAR_OPT_PACKAGER, 'json');         // 默认 json
// $client->SetOpt(YAR_OPT_PACKAGER, 'msgpack');  // 需要 yar 扩展支持

// 设置持久连接
$client->SetOpt(YAR_OPT_PERSISTENT, true);

// 设置自定义 Header
$client->SetOpt(YAR_OPT_HEADER, [
    'X-Api-Key: your-api-key-here',
    'X-Request-Id: ' . uniqid(),
]);

// 设置 Token 用于认证
$client->SetOpt(YAR_OPT_TOKEN, 'your-secret-token');

// 调用
$result = $client->getUserInfo(123);

错误处理

php

<?php
declare(strict_types=1);

try {
    $client = new Yar_Client('http://api.example.com/service.php');
    $result = $client->getUserInfo(123);

    if ($result === false) {
        throw new RuntimeException("调用失败");
    }

    echo "用户名: {$result['name']}" . PHP_EOL;

} catch (Yar_Client_Exception $e) {
    // Yar 特有的异常
    echo "Yar 错误: " . $e->getMessage() . PHP_EOL;
    echo "错误号: " . $e->getCode() . PHP_EOL;
} catch (Yar_Server_Exception $e) {
    // 服务端异常
    echo "服务端错误: " . $e->getMessage() . PHP_EOL;
} catch (Yar_Transport_Exception $e) {
    // 传输层异常
    echo "传输错误: " . $e->getMessage() . PHP_EOL;
}

并行调用

php

<?php
declare(strict_types=1);

/**
 * Yar 并行调用（Concurrent）
 * 使用 Yar_Client::call() 的回调模式
 */
function concurrentCalls(): void
{
    $callback = function (
        $sequence,
        $response,
        $request = null
    ): void {
        echo "请求 #{$sequence} 完成:" . PHP_EOL;
        if ($request instanceof Yar_Client_Request) {
            echo "  方法: " . $request->getMethodName() . PHP_EOL;
        }
        if ($response instanceof Yar_Client_Response) {
            echo "  状态: " . $response->getStatus() . PHP_EOL;
            echo "  结果: " . json_encode($response->getBody(), JSON_UNESCAPED_UNICODE) . PHP_EOL;
        } else {
            echo "  错误: " . $response->getMessage() . PHP_EOL;
        }
    };

    // 并行调用多个服务方法
    Yar_Client::call(
        'http://api.example.com/service.php',
        'getUserInfo',
        [123],
        $callback
    );

    Yar_Client::call(
        'http://api.example.com/service.php',
        'getOrderInfo',
        [456],
        $callback
    );

    // 或者使用数组方式批量并行调用
    $calls = [
        ['http://api.example.com/service.php', 'getUserInfo', [123]],
        ['http://api.example.com/service.php', 'getOrderInfo', [456]],
        ['http://api.example.com/service.php', 'getProductList', [0, 10]],
    ];

    foreach ($calls as $call) {
        Yar_Client::call($call[0], $call[1], $call[2], $callback);
    }

    // 等待所有调用完成（需要事件循环支持）
    Yar_Client::loop(); // 如果使用 Yar_Concurrent_Client
}

并行调用（传统方式）

php

<?php
declare(strict_types=1);

/**
 * Yar_Concurrent_Client 并行调用
 */
function parallelCall(): array
{
    $results = [];
    $sequences = [];

    // 设置回调
    Yar_Concurrent_Client::call(
        'http://api.example.com/user.php',
        'getUser',
        [1],
        function ($sequence, $response) use (&$results): void {
            $results[$sequence] = $response;
        }
    );
    $sequences[] = 0;

    Yar_Concurrent_Client::call(
        'http://api.example.com/user.php',
        'getUser',
        [2],
        function ($sequence, $response) use (&$results): void {
            $results[$sequence] = $response;
        }
    );
    $sequences[] = 1;

    Yar_Concurrent_Client::call(
        'http://api.example.com/user.php',
        'getUser',
        [3],
        function ($sequence, $response) use (&$results): void {
            $results[$sequence] = $response;
        }
    );
    $sequences[] = 2;

    // 发送所有并行请求
    Yar_Concurrent_Client::loop();

    return $results;
}

$results = parallelCall();
print_r($results);

Yar 服务器

创建 Yar 服务

php

<?php
declare(strict_types=1);

/**
 * 用户服务
 */
class UserService
{
    /**
     * 获取用户信息
     */
    public function getUser(int $userId): array
    {
        // 模拟数据库
        $users = [
            1 => ['id' => 1, 'name' => '张三', 'email' => 'zhangsan@example.com'],
            2 => ['id' => 2, 'name' => '李四', 'email' => 'lisi@example.com'],
        ];

        if (isset($users[$userId])) {
            return $users[$userId];
        }

        throw new RuntimeException("用户不存在: {$userId}", 404);
    }

    /**
     * 创建用户
     */
    public function createUser(string $name, string $email): array
    {
        // 验证
        if (empty($name)) {
            throw new RuntimeException("用户名不能为空", 400);
        }

        if (!filter_var($email, FILTER_VALIDATE_EMAIL)) {
            throw new RuntimeException("邮箱格式无效", 400);
        }

        // 模拟插入
        return [
            'id'        => random_int(100, 999),
            'name'      => $name,
            'email'     => $email,
            'created_at' => date('c'),
        ];
    }

    /**
     * 获取用户列表
     */
    public function listUsers(int $page = 1, int $perPage = 10): array
    {
        return [
            'page'     => $page,
            'per_page' => $perPage,
            'total'    => 100,
            'data'     => [
                ['id' => 1, 'name' => '张三'],
                ['id' => 2, 'name' => '李四'],
            ],
        ];
    }
}

// 启动 Yar 服务
$service = new Yar_Server(new UserService());
$service->handle();

服务端配置

php

<?php
declare(strict_types=1);

$service = new Yar_Server(new UserService());

// 设置打包器
$service->setPackager('json');

// 设置 Header
$service->setHeader([
    'Access-Control-Allow-Origin: *',
    'Access-Control-Allow-Methods: POST',
]);

// 处理请求
$service->handle();

服务端错误处理

php

<?php
declare(strict_types=1);

class ProductService
{
    public function getProduct(int $id): array
    {
        $product = $this->findProduct($id);

        if (!$product) {
            // Yar 会自动将 RuntimeException 转换为服务端错误
            throw new RuntimeException("产品不存在", 404);
        }

        return $product;
    }

    // 对于业务级错误，返回标准格式
    public function purchase(int $productId, int $quantity): array
    {
        $product = $this->findProduct($productId);

        if (!$product) {
            return [
                'success' => false,
                'code'    => 404,
                'message'  => '产品不存在',
            ];
        }

        if ($product['stock'] < $quantity) {
            return [
                'success' => false,
                'code'    => 400,
                'message'  => '库存不足',
            ];
        }

        return [
            'success' => true,
            'code'    => 0,
            'message'  => '购买成功',
            'data'    => [
                'order_id' => uniqid('ORD'),
            ],
        ];
    }

    private function findProduct(int $id): ?array
    {
        $products = [
            1 => ['id' => 1, 'name' => '商品A', 'price' => 99.9, 'stock' => 100],
        ];
        return $products[$id] ?? null;
    }
}

详细说明

Yar 协议

Yar 请求/响应的格式：

json

{
    "i": 1234567890,
    "m": "getUserInfo",
    "p": [123]
}

i：请求 ID（时间戳）
m：方法名
p：参数数组

自定义打包器

php

<?php
declare(strict_types=1);

/**
 * 自定义 JSON 打包器
 */
class CustomJsonPackager implements Yar_Packager
{
    public function pack(mixed $data): string|false
    {
        return json_encode($data, JSON_UNESCAPED_UNICODE);
    }

    public function unpack(string $data): mixed
    {
        return json_decode($data, true);
    }

    public function getName(): string
    {
        return 'CustomJson';
    }

    public function getHeader(): string
    {
        return 'Content-Type: application/json';
    }
}

自定义传输协议

php

<?php
declare(strict_types=1);

/**
 * 自定义传输协议（基于 TCP）
 */
class TcpTransport implements Yar_Transport
{
    public function open(string $uri, array $options): bool
    {
        // 建立 TCP 连接
        return true;
    }

    public function send(string $uri, mixed $data): mixed
    {
        // 发送数据
        return true;
    }

    public function exec(string $uri, array $header, mixed $body, array $options): mixed
    {
        // 执行请求
        return $body;
    }

    public function setCookie(string $name, string $value): void {}
    public function getCookie(): array { return []; }
    public function setHeader(string $name, string $value): void {}
    public function getHeader(): array { return []; }

    public function close(): void
    {
        // 关闭连接
    }
}

实战示例

服务发现与负载均衡

php

<?php
declare(strict_types=1);

/**
 * Yar 客户端 + 服务发现
 */
class ServiceDiscovery
{
    private array $services = [];
    private array $currentIndex = [];

    public function register(string $name, string $url): void
    {
        $this->services[$name][] = $url;
    }

    /**
     * 随机选择一个服务实例
     */
    public function getRandomInstance(string $name): ?string
    {
        if (empty($this->services[$name])) {
            return null;
        }

        $index = array_rand($this->services[$name]);
        return $this->services[$name][$index];
    }

    /**
     * 轮询选择
     */
    public function getRoundRobinInstance(string $name): ?string
    {
        if (empty($this->services[$name])) {
            return null;
        }

        if (!isset($this->currentIndex[$name])) {
            $this->currentIndex[$name] = 0;
        }

        $index = $this->currentIndex[$name] % count($this->services[$name]);
        $this->currentIndex[$name]++;

        return $this->services[$name][$index];
    }
}

/**
 * 负载均衡 RPC 客户端
 */
class LoadBalancedRpcClient
{
    public function __construct(
        private readonly ServiceDiscovery $discovery,
        private readonly string $serviceName,
        private readonly int $timeout = 5000
    ) {}

    public function call(string $method, array $params = []): mixed
    {
        $url = $this->discovery->getRoundRobinInstance($this->serviceName);

        if ($url === null) {
            throw new RuntimeException("服务不可用: {$this->serviceName}");
        }

        $client = new Yar_Client($url);
        $client->SetOpt(YAR_OPT_TIMEOUT, $this->timeout);
        $client->SetOpt(YAR_OPT_CONNECT_TIMEOUT, 3000);

        return $client->call($method, $params);
    }
}

// 使用示例
$discovery = new ServiceDiscovery();
$discovery->register('UserService', 'http://192.168.1.1:8080/user.php');
$discovery->register('UserService', 'http://192.168.1.2:8080/user.php');
$discovery->register('UserService', 'http://192.168.1.3:8080/user.php');

$client = new LoadBalancedRpcClient($discovery, 'UserService');
$user = $client->call('getUser', [1]);

Yar 中间件（认证）

php

<?php
declare(strict_types=1);

/**
 * 带 Token 认证的 Yar 服务
 */
class AuthMiddleware
{
    private string $expectedToken;

    public function __construct(string $token)
    {
        $this->expectedToken = $token;
    }

    public function before(\Yar_Server $server): bool
    {
        // 从 Header 获取 Token
        $headers = getallheaders();
        $token = $headers['X-Auth-Token'] ?? '';

        if ($token !== $this->expectedToken) {
            header('HTTP/1.1 401 Unauthorized');
            echo json_encode(['error' => '认证失败']);
            return false;
        }

        return true;
    }
}

注意事项

安全性

安全提示

在生产环境中，Yar 服务应部署在内网
使用 Token 或 IP 白名单进行访问控制
对所有输入参数进行验证
敏感数据应使用 HTTPS 传输

性能对比

指标	Yar (JSON)	Yar (MSGPACK)	SOAP (XML)	XML-RPC
序列化速度	快	非常快	慢	慢
传输体积	小	更小	大	大
连接复用	支持	支持	有限	有限
并行调用	支持	支持	不支持	不支持

最佳实践

使用 MSGPACK 打包器：比 JSON 更快的序列化速度和更小的传输体积
启用持久连接：减少 TCP 握手开销
合理设置超时：避免长时间阻塞
使用并行调用：对无依赖的调用使用 Yar_Concurrent_Client
服务发现：结合服务发现实现负载均衡
监控日志：记录所有 RPC 调用的耗时和状态

继续学习：DOM 操作

Yar RPC 框架 ​

基础概念 ​

什么是 Yar ​

Yar 架构 ​

安装 ​

Yar 客户端 ​

基本调用 ​

配置选项 ​

错误处理 ​

并行调用 ​

并行调用（传统方式） ​

Yar 服务器 ​

创建 Yar 服务 ​

服务端配置 ​

服务端错误处理 ​

详细说明 ​

Yar 协议 ​

自定义打包器 ​

自定义传输协议 ​

实战示例 ​

服务发现与负载均衡 ​

Yar 中间件（认证） ​

注意事项 ​

安全性 ​

性能对比 ​

最佳实践 ​

参考链接 ​