远程过程调用（RPC）

Workers 提供内置的、JavaScript 原生的 RPC（远程过程调用） ↗ 系统，使你可以：

• 在 Worker 上定义公共方法，供同一 Cloudflare 账户下的其他 Worker 通过服务绑定（Service Bindings）调用
• 在 Durable Objects 上定义公共方法，供同一 Cloudflare 账户下声明了相应绑定的其他 Worker 调用。

RPC 系统的设计目标是尽可能接近在同一 Worker 内调用 JavaScript 函数的体验。在大多数情况下，你可以像在单个 Worker 中一样编写代码。

例如，如果 Worker B 实现公共方法 add(a, b)：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_b",
	"main": "./src/workerB.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_b"
main = "./src/workerB.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a, b) {
		return a + b;
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from Worker B");
	}

	add(a: number, b: number) {
		return a + b;
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from Worker B")

    def add(self, a: int, b: int) -> int:
        return a + b

Worker A 可以声明对 Worker B 的绑定（binding）：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "worker_a",
	"main": "./src/workerA.js",
	"services": [
		{
			"binding": "WORKER_B",
			"service": "worker_b"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "worker_a"
main = "./src/workerA.js"

[[services]]
binding = "WORKER_B"
service = "worker_b"

这使得 Worker A 可以调用 Worker B 的 add() 方法：

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

export default {
	async fetch(request, env) {
		const result = await env.WORKER_B.add(1, 2);
		return new Response(result);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        result = await self.env.WORKER_B.add(1, 2)
		return Response(f"Result: {result}")

在本例中，客户端（Worker A）调用 Worker B，并使用客户端提供的特定参数指示其执行特定过程。这通过标准 JavaScript 类实现。

无论你在服务端声明的方法是否为异步，在客户端侧都会表现为异步。你必须 await 结果。

请注意，RPC 调用实际上并不返回 Promise，而是返回行为类似 Promise 的类型。该类型是一种「自定义 thenable」，因为它实现了 then() 方法。JavaScript 支持 await 任何「thenable」类型，因此在大多数情况下，你可以像对待 Promise 一样处理返回值。

（稍后我们会解释为什么该类型实际上不是 Promise。）

几乎所有结构化可克隆（Structured Cloneable） ↗类型都可以作为 RPC 方法的参数或返回值。这包括 JavaScript 中大多数基本「值」类型，如对象、数组、字符串和数字。

作为 Structured Clone 的例外，应用程序定义的类（或具有自定义原型的对象）无法通过 RPC 传递，下文另有说明的情况除外。

RPC 系统还支持多种非 Structured Cloneable 的类型，包括：

• 函数，会被替换为回调至发送方的 stub。
• 继承 RpcTarget 的应用程序定义类，同样会被替换为 stub。
• ReadableStream 和 WriteableStream，具有自动流式流量控制。
• Request 和 Response，便于表示 HTTP 消息。
• RPC stub 本身，即使该 stub 来自第三方 Worker。

你可以通过 RPC 发送函数。发送时，函数会被替换为「stub」。接收方可以像调用函数一样调用 stub，但这样做会向函数来源处发起新的 RPC。

考虑以下两个通过 Service Binding 连接的 Worker。counter 服务提供 RPC 方法 newCounter()，该方法返回一个函数：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter-service",
	"main": "./src/counterService.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter-service"
main = "./src/counterService.js"

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export default class extends WorkerEntrypoint {
	async fetch() {
		return new Response("Hello from counter-service");
	}

	async newCounter() {
		let value = 0;
		return (increment = 0) => {
			value += increment;
			return value;
		};
	}
}

然后客户端 Worker 可以调用此函数：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client_worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter-service"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client_worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter-service"

export default {
	async fetch(request, env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using f = await env.COUNTER_SERVICE.newCounter();
		await f(2); // returns 2
		await f(1); // returns 3
		const count = await f(-5); // returns -2

		return new Response(count);
	},
};

这是如何实现的？系统不会序列化函数本身。当 CounterService 返回的函数被调用时，它在 CounterService 内运行——即使它是由另一个 Worker 调用的。

在底层，调用方并不是真正直接调用函数本身，而是调用所谓的"stub"。"stub" 是一个 Proxy ↗ 对象，允许客户端调用远程服务，就像它在本地运行、在同一 Worker 中一样。在幕后，它会回调实现 CounterService 的 Worker，并要求它执行之前返回的函数闭包。

你也可以在 RPC 的参数中发送函数。这使「服务端」可以回调「客户端」，从而反转两者之间的调用方向。

因此，在讨论 RPC 时，「客户端」和「服务端」可能含义模糊。「服务端」是 Durable Object 或 WorkerEntrypoint，「客户端」是通过绑定调用服务端的 Worker。但 RPC 可以在两者之间双向流动。讨论单个 RPC 时，我们建议使用「调用方（caller）」和「被调用方（callee）」。

要将你定义的类实例用作 RPC 方法的参数或返回值，必须继承内置的 RpcTarget 类。

考虑以下示例：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "counter",
	"main": "./src/counter.js"
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "counter"
main = "./src/counter.js"

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

import { WorkerEntrypoint, RpcTarget } from "cloudflare:workers";

class Counter extends RpcTarget {
	#value = 0;

	increment(amount: number) {
		this.#value += amount;
		return this.#value;
	}

	get value() {
		return this.#value;
	}
}

export class CounterService extends WorkerEntrypoint {
	async newCounter() {
		return new Counter();
	}
}

export default {
	fetch() {
		return new Response("ok");
	},
};

客户端可以直接调用 increment 方法，也可以访问公共属性 value：

{
	"$schema": "./node_modules/wrangler/config-schema.json",
	"name": "client-worker",
	"main": "./src/clientWorker.js",
	"services": [
		{
			"binding": "COUNTER_SERVICE",
			"service": "counter",
			"entrypoint": "CounterService"
		}
	]
}

"$schema" = "./node_modules/wrangler/config-schema.json"
name = "client-worker"
main = "./src/clientWorker.js"

[[services]]
binding = "COUNTER_SERVICE"
service = "counter"
entrypoint = "CounterService"

export default {
	async fetch(request, env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

export default {
	async fetch(request: Request, env: Env) {
		using counter = await env.COUNTER_SERVICE.newCounter();

		await counter.increment(2); // returns 2
		await counter.increment(1); // returns 3
		await counter.increment(-5); // returns -2

		const count = await counter.value; // returns -2

		return new Response(count);
	},
};

继承 RpcTarget 的类与函数类似：对象本身不会被序列化，而是被替换为 stub。在此情况下，stub 本身不可调用，但其方法可调用。调用 stub 上的任何方法实际上都会向创建该对象的原始位置发起 RPC。

如上所示，你还可以访问类的属性。属性的行为类似不带参数的 RPC 方法——你 await 属性以异步获取其当前值。请注意，await 属性（在幕后调用其 .then()）才会触发属性获取。如果访问属性时不使用 await，则不会获取该属性。

调用 RPC 方法并获取对象后，通常会立即调用该对象上的方法：

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

// Two round trips.
using counter = await env.COUNTER_SERVICE.getCounter();
await counter.increment();

但考虑你所调用的 Worker 服务可能位于网络远端的情况，例如智能放置（Smart Placement）或 Durable Objects。上述代码会进行两次往返：一次在调用 getCounter() 时，另一次在调用 .increment() 时。我们希望避免这种情况。

在大多数 RPC 系统中，避免此问题的唯一方式是将两次调用合并为单个「批处理」调用，例如 getCounterAndIncrement()。但这会使接口变差——你不会这样设计本地接口。

Workers RPC 允许另一种方式：你可以省略第一个 await：

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

// Only one round trip! Note the missing `await`.
using promiseForCounter = env.COUNTER_SERVICE.getCounter();
await promiseForCounter.increment();

在此代码中，getCounter() 返回一个 counter 的 promise。通常，对 promise 唯一能做的就是 await 它。然而，Workers RPC 的 promise 很特殊：它们还允许你对 promise 的未来结果发起推测性调用。这些调用会立即发送到服务端，无需等待初始调用完成。因此，多个链式调用可以在一次往返中完成。

这是如何实现的？RPC 返回的 promise 不是真正的 JavaScript Promise。相反，它是一种自定义的 "Thenable" ↗。它像 Promise 一样具有 .then() 方法，因此可以在所有使用普通 Promise 的地方使用，例如可以 await 它。此外，RPC promise 还充当 stub。在 promise 上调用任何方法名都会对 promise 的最终结果形成推测性调用。这称为「promise pipelining」。

这也适用于调用 RPC 方法返回的对象的属性。例如：

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

import { WorkerEntrypoint } from "cloudflare:workers";

export class MyService extends WorkerEntrypoint {
	async foo() {
		return {
			bar: {
				baz: () => "qux",
			},
		};
	}
}

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

export default {
	async fetch(request, env) {
		using foo = env.MY_SERVICE.foo();
		let baz = await foo.bar.baz();
		return new Response(baz);
	},
};

如果初始 RPC 最终抛出异常，则所有流水线调用也会因相同异常而失败。

你可以使用 RPC 方法发送和接收 ReadableStream、WriteableStream、Request 和 Response。发送时，正文中的字节会自动以适当的流量控制进行流式传输。这使你可以通过 RPC 发送大于典型 32 MiB 限制的消息。

仅支持面向字节的流（byte-oriented streams） ↗（底层字节源为 type: "bytes" 的流）。

在所有情况下，流的所有权都会转移给接收方。发送后，发送方无法再读取/写入该流。如果发送方希望保留自己的副本，可以使用 ReadableStream 的 tee() 方法 ↗或 Request 或 Response 的 clone() 方法 ↗。请注意，这样做可能迫使系统缓冲字节并失去流量控制的优势。

从一个 Worker 通过 RPC 接收的 stub 可以再次通过 RPC 转发给另一个 Worker。

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

using counter = env.COUNTER_SERVICE.getCounter();
await env.ANOTHER_SERVICE.useCounter(counter);

此处涉及三个不同的 Worker：

1. 调用方 Worker（我们称之为「引入方（introducer）」）
2. COUNTER_SERVICE
3. ANOTHER_SERVICE

当 ANOTHER_SERVICE 调用传入的 counter 上的方法时，该调用会自动通过引入方代理到 COUNTER_SERVICE 实现的 RpcTarget 类。

通过这种方式，引入方 Worker 可以连接两个原本无法直接建立连接的其他 Worker。

目前，这种代理仅持续到 Worker 执行上下文结束。代理连接无法持久化以供后续使用。

在本视频中，我们探讨 Cloudflare Workers 如何支持远程过程调用（RPC）以简化 Worker 之间的通信。学习如何在你的 JavaScript 应用程序中实现 RPC，并轻松构建无服务器解决方案。无论你是在管理微服务还是优化 Web 架构，本教程都将展示如何快速设置和使用 Cloudflare Workers 进行 RPC 调用。视频结束时，你将了解如何在 Worker 之间调用函数、将函数作为参数传递，以及使用 Cloudflare Workers 实现用户身份验证。

• 生命周期
• 保留方法
• 可见性与安全模型
• TypeScript
• 错误处理

• 进行 RPC 调用时，当前会忽略智能放置（Smart Placement）。如果 Worker A 启用了 Smart Placement，且 Worker B 声明了指向它的服务绑定（Service Binding），当 Worker B 通过 RPC 调用 Worker A 时，Worker A 将在本地、同一台机器上运行。

• 最大序列化 RPC 限制为 32 MiB。返回更多数据时，请考虑使用 ReadableStream。

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }

  export class MyService extends WorkerEntrypoint {
  	async foo() {
  		// Although this works, it puts a lot of memory pressure on the isolate.
  		// If possible, streaming the data from its original source is much preferred and would yield better performance.
  		// If you must buffer the data into memory, consider chunking it into smaller pieces if possible.
  
  		const sizeInBytes = 33 * 1024 * 1024; // 33 MiB
  		const arr = new Uint8Array(sizeInBytes);
  
  		return new ReadableStream({
  			start(controller) {
  				controller.enqueue(arr);
  				controller.close();
  			},
  		});
  	}
  }