← Cloudflare Workers / workers / runtime-apis

Удалённый вызов процедур (RPC)

Workers предоставляют встроенный, нативный для JavaScript RPC (удалённый вызов процедуры) ↗ систему, которая позволяет:

• Определяйте публичные методы вашего Worker, которые могут вызывать другие Workers в том же аккаунте Cloudflare, через Service Bindings
• Определяйте публичные методы для Durable Objects которую могут вызывать другие workers в том же аккаунте Cloudflare, если они объявляют привязку к ней.

Система RPC спроектирована так, чтобы вызов ощущался максимально похожим на вызов функции JavaScript в том же Worker. В большинстве случаев код можно писать так же, как если бы все находилось в одном 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:

{
	"$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 возможность вызывать add() метод из Worker B:

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 на самом деле не возвращают Promises, но они возвращают тип, который ведет себя как Promise. Этот тип является «custom thenable», то есть реализует метод then(). JavaScript поддерживает ожидание любого типа «thenable», поэтому в большинстве случаев возвращаемое значение можно рассматривать как Promise.

(Чуть позже мы разберём, почему этот тип на самом деле не является Promise.)

Почти все типы, которые Структурированное клонирование ↗ можно использовать как параметр или возвращаемое значение метода RPC. Сюда входят практически все базовые типы значений JavaScript, включая объекты, массивы, строки и числа.

В отличие от Structured Clone, классы, определённые приложением (или объекты с пользовательскими прототипами), нельзя передавать через RPC, за исключением случаев, описанных ниже.

Система RPC также поддерживает ряд типов, не являющихся Structured Cloneable, включая:

• Functions, которые заменяются заглушками, обращающимися обратно к отправителю.
• Классы, определённые приложением, которые расширяют RpcTarget, которые точно так же заменяются заглушками.
• ReadableStream и WriteableStream, с автоматическим контролем потока при потоковой передаче.
• Запрос и Ответ, для удобного представления HTTP-сообщений.
• сами заглушки RPC, даже если заглушка была получена от стороннего Worker.

Функцию можно передать по RPC. При этом функция заменяется «заглушкой» (stub). Получатель может вызывать заглушку как обычную функцию, однако такой вызов инициирует новый RPC-запрос обратно в то место, откуда функция была передана изначально.

Рассмотрим следующие два Worker, соединенных через Service Binding. Сервис счётчика предоставляет 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;
		};
	}
}

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        return Response("Hello from counter-service")

    async def new_counter(self):
        value = 0
        def increment(amount=0):
            nonlocal value
            value += amount
            return value
        return increment

Затем эту функцию можно вызвать из клиентского 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);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()
        await counter(2)  # returns 2
        await counter(1)  # returns 3
        count = await counter(-5)  # returns -2

        return Response(str(count))

Как это возможно? Система не сериализует саму функцию. Когда функция, возвращённая CounterService вызывается, он выполняется в CounterService : даже если он вызван другим Worker.

На самом деле вызывающая сторона обращается не напрямую к самой функции, а к так называемой «заглушке». Заглушка представляет собой Прокси ↗ объект, который позволяет клиенту вызывать удалённый сервис так, как если бы он был локальным и выполнялся в том же Worker. За кулисами он обращается обратно к Worker, который реализует CounterService и просит его выполнить замыкание функции, возвращенное ранее.

Также в параметрах RPC можно передать функцию. Это позволяет «серверу» вызывать «клиента» в обратном направлении, меняя направление взаимодействия.

Поэтому термины «клиент» и «сервер» могут быть неоднозначными применительно к RPC. «Сервером» является Durable Object или WorkerEntrypoint, а «клиентом» является Worker, вызвавший сервер через привязку. При этом RPC может выполняться в обе стороны между ними. Когда речь идёт об отдельном вызове RPC, рекомендуем использовать термины «вызывающая сторона» и «вызываемая сторона».

Чтобы использовать экземпляр определённого вами класса в качестве параметра или возвращаемого значения метода 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);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        counter = await self.env.COUNTER_SERVICE.new_counter()

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

        count = await counter.value  # returns -2

        return Response(str(count))

Классы, которые расширяют RpcTarget работают во многом как функции: сам объект не сериализуется, а заменяется заглушкой. В этом случае саму заглушку нельзя вызвать напрямую, а вот ее методы можно. Вызов любого метода заглушки на самом деле выполняет RPC обратно к исходному объекту, там, где он был создан.

Как показано выше, вы также можете обращаться к свойствам классов. Свойства ведут себя как методы RPC, не принимающие аргументов: чтобы асинхронно получить текущее значение свойства, его нужно дождаться (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();

# Two round trips.
counter = await self.env.COUNTER_SERVICE.get_counter()
await counter.increment()

Однако рассмотрим случай, когда служба Worker, которую вы вызываете, может находиться далеко в сети, как, например, в случае с Smart Placement или Durable Objects. Код выше выполняет два обращения туда и обратно: один раз при вызове getCounter(), и снова при вызове .increment(). Нам хотелось бы избежать этого.

В большинстве RPC-систем избежать этой проблемы можно, только объединив два вызова в один пакетный вызов («batch»), который можно было бы назвать 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();

# Only one round trip! Note the missing await.
promise_for_counter = self.env.COUNTER_SERVICE.get_counter()
await promise_for_counter.increment()

В этом коде getCounter() возвращает promise для счетчика. Обычно с promise можно сделать только одно: await его. Однако RPC promise в Workers особенные: они также позволяют инициировать спекулятивные вызовы к будущему результату promise. Эти вызовы отправляются на сервер немедленно, не дожидаясь завершения первого вызова. Таким образом, несколько последовательных вызовов могут быть выполнены за одно обращение к серверу.

Как это работает? Промис, возвращаемый RPC-вызовом, не является настоящим JavaScript Promise. Вместо этого это специальный «Thenable» ↗. У него есть .then() метод, например Promise, что позволяет использовать его везде, где вы бы использовали обычный Promise. Например, вы можете await его. Но, помимо этого, RPC promise также ведёт себя как заглушка. Вызов любого метода на 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",
			},
		};
	}
}

from workers import WorkerEntrypoint

class MyService(WorkerEntrypoint):
    async def foo(self):
        return {
            "bar": {
                "baz": lambda: "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);
	},
};

from workers import WorkerEntrypoint, Response

class Default(WorkerEntrypoint):
    async def fetch(self, request):
        foo = self.env.MY_SERVICE.foo()
        baz = await foo.bar.baz()
        return Response(baz)

Если первоначальный вызов RPC завершается исключением, все конвейеризированные вызовы также завершаются тем же исключением

Можно отправлять и получать ReadableStream, WriteableStream, Request, а также Response с помощью методов RPC. При этом байты тела автоматически передаются потоком с соответствующим управлением потоком данных. Это позволяет отправлять по RPC сообщения, размер которых превышает стандартное ограничение в 32 MiB.

Только байт-ориентированные потоки ↗ (потоки, у которых базовый источник байтов type: "bytes") поддерживаются.

Во всех случаях владение потоком переходит к получателю. После отправки отправитель больше не может читать поток или писать в него. Если отправителю нужно сохранить собственную копию, он может использовать tee() метод из ReadableStream ↗ или clone() метод из Request или Response ↗. Учтите, что это может заставить систему буферизовать байты и свести на нет преимущества управления потоком.

Stub, полученный по RPC от одного Worker, можно передать по 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);

counter = self.env.COUNTER_SERVICE.get_counter()
await self.env.ANOTHER_SERVICE.use_counter(counter)

Здесь задействованы три разных Worker:

1. Вызывающий Worker (далее будем называть его "introducer")
2. COUNTER_SERVICE
3. ANOTHER_SERVICE

Когда ANOTHER_SERVICE вызывает метод у counter который ему передан, этот вызов будет автоматически проксирован через introducer и далее к RpcTarget класс, реализованный COUNTER_SERVICE.

Таким образом, Worker-посредник может соединить два Worker, у которых иначе не было бы возможности установить прямое соединение друг с другом.

В настоящее время такое проксирование действует только до завершения контекста выполнения Workers. Сохранить проксируемое соединение для последующего использования нельзя.

В этом видео мы разбираем, как Cloudflare Workers поддерживают удалённый вызов процедур (RPC) для упрощения взаимодействия между Workers. Вы узнаете, как реализовать RPC в приложениях на JavaScript и без труда создавать бессерверные решения. Управляете ли вы микросервисами или оптимизируете веб-архитектуру, это руководство покажет, как быстро настроить и использовать Cloudflare Workers для вызовов RPC. К концу видео вы поймёте, как вызывать функции между Workers, передавать функции в качестве аргументов и реализовывать аутентификацию пользователей в Cloudflare Workers.

• Жизненный цикл
• Зарезервированные методы
• Модель видимости и безопасности
• TypeScript
• Обработка ошибок

• Smart Placement в настоящее время игнорируется при выполнении RPC-вызовов. Если Smart Placement включён для Worker A, а Worker B объявляет Service Binding к нему, при вызове Worker A из Worker B через RPC 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();
  			},
  		});
  	}
  }