iii 渐进式改造指南零大爆炸重写将存量 HTTP 服务分片迁移到 iii【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii导读本文是 iii 官方教程增量采纳Incremental Adoption系列的总览篇对应仓库文档 docs/tutorials/incremental-adoption/overview.mdx.skill.md核心回答一个问题如何在不做 big-bang 全量重写、不需要一次性割接的前提下把一套已经在跑的存量系统逐步搬到 iii 上。你将依次掌握三条可独立执行、可随时回滚的迁移路径——用 iii 函数包住现有 HTTP 服务、把慢调用转移到队列、把持久化切片迁入 state 原语——最终让系统跑在 iii 上却从未经历完整切换。读完本文你既能照做三个子教程也能理解每条路径背后的引擎与 SDK 实现原理。为什么需要增量采纳iii 的架构决定了迁移可以切片进行iii 的引擎本质是一个路由器官方文档原话The engine is just a router它接收请求、把请求路由到对应的 worker、再把响应路由回来见 docs/using-iii/engine.mdx。这套架构天然带来两个对渐进式改造极其友好的特性函数是第一公民的寻址单元任何能力都以service::name形式的function_id注册到引擎调用方只需要知道这个 ID而不需要知道实现它的是哪个进程、跑在哪台机器上见 docs/using-iii/functions.mdx 与 skills/iii-core-primitives/SKILL.md。worker 只是连上引擎的进程worker 不需要与引擎同机部署只要有一条指向引擎实例的连接串即可docs/using-iii/engine.mdx 中的 Note 明确说明了这一点。这意味着你的存量服务可以先原封不动只在其前方加一个 iii 形状的入口流量、数据、依赖都可以一小片一小片地迁每一片都是独立可逆的。这正是总览文档给出的方法论一个切片slice一次随时可暂停、可回滚不会把系统其余部分拖下水。前置条件一个跑起来的引擎 一个能被 HTTP 访问的服务按总览文档动手前你需要准备好三样东西一个正在运行的 iii 引擎先完成安装然后在一个 scratch 目录里运行iii让它自动创建config.yaml再用iii worker add把每一步需要的 worker 加进来。关于引擎首次运行的行为与默认配置见 Default configuration。一个你想要迁移的存量服务要求它能通过 HTTP 访问到这是第 1 步包 API的前提。对应语言的 SDKTypeScript、Python 或 Rust 三选一取决于你的服务用什么语言写的。关于引擎默认配置你需要知道的三件事结合 docs/using-iii/engine.mdx 的说明首次运行有几个容易踩的点在没有config.yaml的目录里运行iii它会主动询问是否创建该文件非交互式会话如容器、CI 中则直接创建、不询问生成的文件以一个空的workers:列表起步——引擎自带内部服务SDK 连接的 WebSocket 监听器、configuration worker、可观测性其余能力全部靠iii worker add name按需引入且引擎会实时监听该文件、把新增 worker 热加载进来。引擎只读config.yaml不读iii.locklockfile 是 worker 安装层面的东西由iii worker sync/update/verify写入和消费用来保证安装可复现详见 Workers / The lockfile (iii.lock)。config.yaml中的值支持${VAR:default}环境变量展开适合按环境切换端口、URL 和 feature flag而无需 fork 配置文件。一个典型的最小config.yaml长这样来自 docs/using-iii/engine.mdxworkers: - name: http config: port: 3111 host: 127.0.0.1 - name: state config: adapter: name: kv config: store_method: file_based file_path: ./data/state_store.db注意每个 worker 下的config:块是一次性的first-boot seed——worker 首次启动时读取它来初始化自身在 configuration worker 里的条目此后设置会持久化到./config/下每个 worker 一个文件支持从磁盘、console 或configuration::set实时编辑而引擎会把config.yaml里已消费的块替换成注释。三步迁移路线总览总览文档给出三条子教程路径顺序执行即可完成系统整体迁移步骤子教程目标风险暴露1Wrap an existing API用 iii 函数给存量 HTTP 服务加上iii 形状的入口零流量迁移只新增寻址层2Offload work to a queue把耗时的长调用放到队列 worker 后面调用方立即返回重试交给队列3Migrate persistence一次一个切片地把持久化迁入 state worker系统其余部分保持不动直到你准备好迁下一片下面结合仓库源码与配套文档把每一步能落地的细节展开。第 1 步包装现有 API——给存量服务一个 iii 寻址入口这一步的目标是让现有的 HTTP 服务可以被系统内任意 worker 通过function_id寻址到。注意这一步不迁移任何流量——你只是新增一个 iii 形状的入口点。在 iii 中完成这件事的标准做法是注册一个 HTTP-invokable 函数HTTP 可调用函数引擎在函数被调用时替你做 HTTP 调用你的 worker 只需要声明这个端点不需要改动存量服务的一行代码。官方文档明确说这非常适合委托给现有的 API Gateway、webhook、serverless 平台Lambda、Azure Functions、Google Cloud Functions或任何你想以常规 iii 函数形式暴露的第三方 API见 docs/creating-workers/functions.mdx 的 HTTP-invokable functions 一节。以 Node / TypeScript 为例把外部 webhook 注册为notifications::sendimport { registerWorker } from iii-sdk; const url process.env.III_URL; if (!url) throw new Error(III_URL must be set); const worker registerWorker(url); worker.registerFunction( notifications::send, { url: https://hooks.provider.example.com/notify, method: POST, timeout_ms: 5000, headers: { X-Service: iii-worker }, auth: { type: bearer, token_key: PROVIDER_API_TOKEN }, }, { description: POST a notification to the provider webhook }, );Python 版本使用HttpInvocationConfigimport os from iii import HttpInvocationConfig, InitOptions, register_worker from iii.iii_types import HttpAuthBearer worker register_worker( os.environ.get(III_URL), InitOptions(worker_namenotifications-worker), ) worker.register_function( notifications::send, HttpInvocationConfig( urlhttps://hooks.provider.example.com/notify, methodPOST, timeout_ms5000, headers{X-Service: iii-worker}, authHttpAuthBearer(token_keyPROVIDER_API_TOKEN), ), descriptionPOST a notification to the provider webhook, )Rust 版本use std::collections::HashMap; use iii_sdk::{ HttpAuthConfig, HttpInvocationConfig, HttpMethod, InitOptions, RegisterFunctionMessage, register_worker, }; let url std::env::var(III_URL).expect(III_URL must be set); let worker register_worker(url, InitOptions::default()); let mut headers HashMap::new(); headers.insert(X-Service.into(), iii-worker.into()); worker.register_function(( RegisterFunctionMessage::with_id(notifications::send.into()) .with_description(POST a notification to the provider webhook.into()), HttpInvocationConfig { url: https://hooks.provider.example.com/notify.into(), method: HttpMethod::Post, timeout_ms: Some(5000), headers, auth: Some(HttpAuthConfig::Bearer { token_key: PROVIDER_API_TOKEN.into(), }), }, ));HttpInvocationConfig 字段说明普通函数接受id handler而 HTTP 可调用函数接受idHttpInvocationConfigdocs/creating-workers/functions.mdx 中的字段表字段类型默认值说明urlstring必填函数被调用时引擎去请求的端点methodGET \| POST \| PUT \| PATCH \| DELETEPOSTHTTP 方法timeout_msnumber30000单次请求超时毫秒headersRecordstring, string{}每次调用都会附加的请求头authHttpAuthConfig无认证配置支持bearer、hmac或api_key配合token_key、secret_key或value_key一个值得注意的安全设计auth里的token_key/secret_key/value_key指定的是环境变量的名字而不是密钥本身。引擎在注册时从自己的进程环境里解析这些变量因此密钥只存在于引擎宿主机上永远不会通过 SDK 的 WebSocket 连接传输。注册之后它就是一个普通函数包装完成后这个函数可以像任何其他 iii 函数一样被触发worker.trigger、iii trigger以及任何绑定型 triggerqueue、cron、state、http都直接可用无需其他改动。引擎把调用 payload 作为 JSON 请求体发出任何非 2xx 响应或网络错误都会作为调用失败传回调用方。HTTP 可调用函数同样出现在engine::functions::list里并像进程内 handler 一样在 console 中可发现见 docs/creating-workers/functions.mdx 的 HTTP error handling 一节。这一步对存量系统是零侵入的没有流量移动、没有代码改写只是多了一个寻址层。对应本系列第一个子教程 wrap-existing-api。第 2 步把慢活卸载到队列——调用方立即返回重试交给队列第 1 步解决的是寻址第 2 步解决的是慢调用拖垮调用方。总览文档的目标是把耗时的长调用放到队列 worker 后面让包装函数立即返回重试交给队列处理。添加 queue workerqueueworker 把生产者和消费者解耦函数向一个命名 topic 发布消息后立即返回任何订阅了该 topic 的函数在后台处理消息自带重试和死信队列DLQ见 docs/creating-workers/queues.mdx。iii worker add queue定义命名队列队列在queueworker 的queue_configs下定义- name: queue config: queue_configs: email-jobs: max_retries: 3 concurrency: 10 type: standard每个queue_configs条目支持的字段来自 docs/creating-workers/queues.mdx字段默认值说明max_retries3消息进入 DLQ 之前的投递尝试次数concurrency10并行处理的任务数必须 1schema 会拒绝0因此不能用它暂停队列fifo队列强制为1typestandardstandard并发或fifo按消息组内有序message_group_field无fifo必填决定排序组的 payload 字段backoff_ms1000重试基础延迟毫秒指数退避poll_interval_ms100worker 轮询间隔毫秒用 TriggerAction.Enqueue 入队入队函数和普通worker.trigger的注册方式完全一样唯一区别是给 trigger 提供一个TriggerAction.Enqueueactionimport { TriggerAction, type EnqueueResult } from iii-sdk; const { messageReceiptId } await worker.triggerunknown, EnqueueResult({ function_id: email::send, payload: { to: ab.com, subject: hi }, action: TriggerAction.Enqueue({ queue: email-jobs }), // queue 指定队列名 }); // messageReceiptId 标识已入队的任务Pythonfrom iii import TriggerAction receipt worker.trigger({ function_id: email::send, payload: {to: ab.com, subject: hi}, action: TriggerAction.Enqueue(queueemail-jobs), # queue 指定队列名 }) # receipt[messageReceiptId] 标识已入队的任务Rustuse iii_sdk::TriggerAction; use iii_sdk::protocol::TriggerRequest; use serde_json::json; let receipt worker .trigger(TriggerRequest { function_id: email::send.to_string(), payload: json!({ to: ab.com, subject: hi }), action: Some(TriggerAction::Enqueue { queue: email-jobs.to_string() }), timeout_ms: None, }) .await?; // receipt[messageReceiptId] 标识已入队的任务三种调用模式的选择依据从 skills/iii-core-primitives/SKILL.md 可以看到 iii 明确区分三种调用模式这在迁移时是重要的决策点模式形状适用场景Synctrigger({ function_id, payload })调用方需要同步拿到结果VoidTriggerAction.Void()可选的副作用不需要结果EnqueueTriggerAction.Enqueue({ queue })需要重试保证的可靠异步任务官方建议必须完成的重活带重试用 enqueue分析、通知等非关键副作用用 void。这正好对应迁移的第 2 步——把原来同步阻塞的慢调用改成 enqueue 后包装函数立即返回调用方不再被长耗时拖住。死信与可观测性迁移后的运维兜底队列迁移的最大顾虑是消息失败了怎么办。iii 的 queue worker 提供完整的 DLQ 生命周期迁移后可以直接用 CLI 巡检# 列出有死信消息的 topic iii trigger engine::queue::dlq_topics # 浏览死信消息含失败原因、重试次数 iii trigger engine::queue::dlq_messages topicemails # 修好代码后把整个 topic 的死信消息重新投递 iii trigger iii::queue::redrive topicemails # 或只处理单条消息 iii trigger iii::queue::redrive_message topicemails message_idid iii trigger iii::queue::discard_message topicemails message_idid消息只有在订阅函数耗尽重试次数后才会进入 DLQ所以 DLQ 函数平时返回空需要验证时可以让 handler 故意抛错观察消息经历 3 次指数退避重试1 秒、2 秒后落进死信队列。此外 pub/sub 场景下还可以用iii trigger engine::queue::list_topics和iii trigger engine::queue::topic_stats topicemails查看订阅数与积压深度depth/dlq_depth。这些能力让把慢活搬到队列后面在运维上是可以被观测、被干预的——这正是渐进式迁移敢于切流的前提。对应子教程 offload-to-queue。第 3 步迁移持久化——一次一个切片地搬进 state 原语第 3 步处理的是系统里最难大爆炸的部分数据。总览文档的原则是一次只把一个状态切片slice迁入 state worker系统其余部分保持不动直到你准备好迁移下一个切片。iii 的 state 模型iii 的状态是一个分布式 KV 存储以scope分组key条目 ID寻址。任何 worker 都可以通过触发state::get、state::set、state::delete、state::list来读写状态见 docs/0-11-0/examples/state-management.mdx。state worker 的存储后端由配置决定默认示例使用kv适配器 file_based文件存储见前文config.yaml示例。切片写入示例以用户状态更新为例把某个用户的状态切片写入 stateimport { registerWorker, TriggerAction } from iii-sdk; const iii registerWorker(process.env.III_URL ?? ws://localhost:49134); iii.registerFunction( { id: users::update_status, description: Update user status in state }, async (req: ApiRequest{ status: string }) { const userId req.path_params?.id; if (!userId) return { status_code: 400, body: { error: Missing user ID } }; const { status active } req.body ?? {}; // 写入 state —— 用 scope key 寻址 await iii.trigger({ function_id: state::set, payload: { scope: users, key: userId, value: { status, updatedAt: new Date().toISOString() }, }, action: TriggerAction.Void(), }); return { status_code: 200, body: { userId, status } }; }, ); iii.registerTrigger({ type: http, function_id: users::update_status, config: { api_path: /users/:id/status, http_method: POST }, });Python 版本from datetime import datetime, timezone from iii import register_worker, InitOptions, ApiRequest, ApiResponse, Logger, TriggerAction iii register_worker(addressws://localhost:49134, optionsInitOptions(worker_namestate-worker)) def update_user_status(data) - ApiResponse: req ApiRequest(**data) if isinstance(data, dict) else data user_id req.path_params.get(id) if req.path_params else None if not user_id: return ApiResponse(status_code400, body{error: Missing user ID}) new_status (req.body or {}).get(status, active) # 写入 state —— 用 scope key 寻址 iii.trigger({ function_id: state::set, payload: {scope: users, key: user_id, value: {status: new_status, updatedAt: datetime.now(timezone.utc).isoformat()}}, action: TriggerAction.Void(), }) return ApiResponse(status_code200, body{userId: user_id, status: new_status}) iii.register_function(users::update_status, update_user_status) iii.register_trigger({ type: http, function_id: users::update_status, config: {api_path: /users/:id/status, http_method: POST}, })让切片的其余部分感知变化迁移持久化时一个关键问题是其他仍然留在旧系统的部分如何感知新状态的变化。iii 的statetrigger 为此提供了内置的事件流注册statetrigger 后handler 会收到{ event_type, scope, key, old_value, new_value }形式的变更事件见 skills/iii-core-primitives/SKILL.md 的内置 trigger 形状表。也就是说你迁入 state 的切片既可以主动被读也可以被动广播变化旧系统可以在不改写核心逻辑的前提下订阅这些事件完成联动。如果希望 handler 只在满足条件时才执行还可以给 trigger 配置加condition_function_id先跑一个小函数、返回 false 就跳过 handler——这为灰度放量提供了一个天然的开关参见 docs/how-to/use-trigger-conditions.mdx 的规划与 skills/iii-core-primitives/SKILL.md。对应子教程 migrate-persistence。把三步串起来切片化迁移的完整路径综合三步一个典型的存量系统迁移是这样的加入口用 HTTP-invokable 函数把现有服务暴露为your-service::*函数——其他 worker、CLI、trigger 都能用function_id找到它但流量还在老路上。加队列把最耗时的调用发邮件、生成报表、调用外部 API 等改为TriggerAction.Enqueue调用方立即返回max_retries/backoff_ms/ DLQ 保证可靠投递engine::queue::*系列函数保证可观测、可干预。切片搬数据按业务域scope把一个一个状态切片迁入 state worker用state::set/state::get读写用statetrigger 向仍然留在旧系统的部分广播变化。每一步都是一个独立可逆的切片可以在任何边界暂停也可以随时回滚而不会把系统其余部分拖垮——这正是总览文档结论部分的承诺Each slice is independently reversible, so you can pause or roll back at any boundary without taking the rest of the system down.迁移期间的工具支撑整个过程都跑在标准 iii CLI 之上你可以随时用以下命令确认每一步的状态命令语义见 docs/using-iii/cli.mdxiii --help # 查看全部子命令 iii worker add name # 按需添加 worker引擎热加载 iii trigger fn # 触发一个已注册函数验证入口是否就绪引擎侧的可发现性函数engine::workers::list、engine::functions::list、engine::triggers::list可以随时查看当前系统里已经 iii 化的部分skills/iii-core-primitives/SKILL.md它们本身就是迁移进度的仪表盘。结论iii 的增量采纳路径证明了上 iii不等于重写系统用 HTTP-invokable 函数给存量服务一个 iii 寻址入口零流量、零改写用队列把慢活异步化调用方立即返回、重试与 DLQ 兜底用 state 原语按 scope 切片迁移持久化每片独立可逆、可广播变更。三个子教程按顺序走完你的系统就在从未做过一次全量割接的情况下整体跑在了 iii 上。每一个边界的暂停与回滚代价都被限制在单个切片之内。【免费下载链接】iiiEffortlessly compose, extend, and observe every service in real-time for the first time ever.项目地址: https://gitcode.com/GitHub_Trending/mo/iii创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考