PHP 异步双向 JSON-RPC 2.0 对等通信库

概述

这是一种极简的、异步的、双向的 JSON-RPC 2.0 协议实现，它通过换行符分隔的 JSON 数据流进行通信。该协议基于 amphp 字节流协议，并利用了 Revolt 事件循环机制来加速处理流程。

这是一个基于 JSON-RPC 的库：它提供了一种长寿命的连接方式，双方可以通过该连接发送请求和通知，响应传入的请求，并按任意顺序处理返回的结果。这种机制是如 Language Server 协议和 Model Context 协议等 stdio 协议的基础。

区别于传统 PHP JSON-RPC 库（单向 HTTP、区分客户端/服务端、同步阻塞），它是对等 Peer 模型：单条长连接两端可互相发起调用、推送通知，响应可乱序返回，专为 stdio 进程双向通信设计，是 LSP（语言服务协议）、MCP（模型上下文协议）底层基础组件。

核心差异化特性

1. 对等双向通信（Peer）不分客户端/服务端，同一连接两端均可主动发请求、发通知；请求并发执行，响应不强制按发送顺序返回。
2. 持久双工流传输基于行分隔 JSON，兼容任意 amphp 读写流：标准输入输出(stdio)、TCP Socket、内存测试流，无 HTTP 限制。
3. 全异步协程模型每个入站请求独立协程运行，长任务阻塞不影响其他消息处理；内置 Amp Cancellation 协作式取消机制。
4. 完整 JSON-RPC 2.0 规范兼容支持单次请求、无响应通知、批量混合消息；内置标准错误码自动处理。
5. 进程间协议原生适配原生适配 LSP、MCP 这类基于 stdio 的双向 RPC 协议，内置标准化取消通知绑定能力。

安装

composer require fabpot/json-rpc-peer

使用方式

use Amp\ByteStream;
use Fabpot\JsonRpc\JsonRpcDispatcher;
use Fabpot\JsonRpc\JsonRpcPeer;

$input = ByteStream\getStdin();
$output = ByteStream\getStdout();

$peer = new JsonRpcPeer($input, $output);
$dispatcher = new JsonRpcDispatcher($peer);

处理请求和通知

根据方法名称来注册处理程序。当请求处理程序返回结果时，调度器会将其作为 JSON-RPC 响应发送出去。

$dispatcher->onRequest('sum', function (array $params): array {
    return ['total' => array_sum($params['values'])];
});

通知处理程序不返回任何内容，因为通知本身并没有对应的响应机制：

$dispatcher->onNotification('log', function (array $params): void {
    fwrite(\STDERR, $params['message']."\n");
});

运行对等节点

在注册了处理程序之后，调用 listen() 。它会读取并分发消息，直到输入流到达末尾或被关闭。之后，它会请求取消正在运行的处理程序，并等待它们完成工作后再返回。

$peer->listen();

核心模块与基础使用流程

1. 核心类

• JsonRpcPeer：底层流读写、消息收发、连接管理
• JsonRpcDispatcher：方法路由、请求/通知处理器注册、取消管理
• JsonRpcError：JSON-RPC 2.0 标准错误码常量
• JsonRpcException：RPC 业务异常封装
• PsrTrafficLogger：流量日志（自动脱敏密钥、token、密码等敏感字段）

2. 最简使用流程

1. 绑定输入输出流（典型 stdin/stdout）
2. 实例化 Peer + Dispatcher
3. 注册请求处理器（有返回值）、通知处理器（无返回）
4. 可选绑定取消通知规则（如 LSP $/cancelRequest）
5. 异步启动 listen() 持续监听消息
6. 外部可主动发起请求/推送通知

关键功能详解

1. 请求 & 通知处理

• onRequest(method, handler)：处理远端调用，返回结果自动封装 Response；支持接收 Cancellation 实现任务取消。
• onNotification(method, handler)：无响应推送（如进度日志），无返回值。
• 远端主动调用：peer->request() 返回 Amp Future，await 等待结果；远端推送通知：

2. 错误体系（标准 JSON-RPC 2.0 错误码）

库自动抛出前3类错误；业务参数错误手动抛 JsonRpcException；未捕获异常统一转为内部错误，不泄露原始异常信息。

3. 长任务与请求取消（核心亮点）

1. 每个入站请求自动分配 Cancellation 对象，处理器可接收第二个参数使用。
2. 支持协作式取消：调用 $cancellation->throwIfRequested() 中断任务；Amp 延迟/IO 原生支持取消。
3. 绑定取消通知：dispatcher->onCancel('通知名', 'id字段')，例如 LSP 标准
4. 流关闭时自动取消所有活跃请求，抛出连接关闭异常。

4. 批量消息 Batch

支持混合请求+通知批量发送：

• BatchRequest：会返回 Future，可 await 获取结果
• BatchNotification：无响应，无返回值 批量响应顺序与处理完成顺序一致，不匹配发送顺序；通知不会出现在响应数组。

5. 流量日志

PsrTrafficLogger 对接任意 PSR-3 日志，自动脱敏 token/password/authorization 等敏感 key，支持自定义扩展敏感字段，记录收发原始 JSON 行，用于调试协议交互。

生命周期与连接管理

1. $peer->listen()：阻塞监听流，直到流 EOF/关闭；关闭前等待所有活跃协程执行完毕。
2. 异步启动：Amp\async(fn() => $peer->listen())，主线程可同时向外发请求。
3. 流关闭行为：
   • 本地 $input->close() 或远端关闭输出，触发 listen 退出；
   • 所有未完成出站请求抛出 ConnectionClosedException；
   • 关闭后新调用 request() 直接抛连接异常。

适用场景

1. 语言服务 LSP：PHP 开发 VSCode 插件、自定义语言服务器（stdio 双向通信）
2. MCP 大模型上下文协议：本地进程与 AI 工具双向调用交互
3. 本地进程间 IPC：父子进程、多程序长连接双向互调
4. Socket 长连接 RPC：TCP 双工通信，服务端主动推送消息
5. 异步测试：内存流替代真实 stdio，单元测试 RPC 交互

对比传统 PHP JSON-RPC 库优势