Cloudflare K2：无服务器事件流

# 发布 Cloudflare K2：无服务器事件流 来源：https://blog.cloudflare.com/cloudflare-k2-streams/ 在传统的远程过程调用（RPC）架构中，存在一个核心难题：生产者与消费者必须在规模和时间上保持对齐。如果你的生产者发送的数据量超出了消费者的处理能力，或者消费者或下游服务不可用，事件就会被丢弃。当有多个消费者需要各自独立地处理数据时，这个问题会更加突出。例如，电商后端在交易完成时会发出事件，这些事件需要被分析系统和欺诈检测服务同时读取。 我们可以通过*解耦*生产者和消费者来解决这个问题——在中间插入一个服务，吸收写入量，同时允许独立的读者按自己的节奏消费数据。 今天，我们正式发布 Cloudflare K2 公测版来解决这一问题。K2 是开发者平台上的一个持久化事件流原语。你将事件发送到 K2 流中，它会将这些事件存储为有序的日志。消费者可以通过多种方式读取事件，例如将读取任务分配给一组消费者，或将所有消息分发给每个消费者。它是完全无服务器的，可以扩展到海量数据量，并支持长期保留，因此即使消费者长时间宕机也不会丢失数据。 在底层，K2 在 R2 对象存储之上实现了分区的持久化日志，这使它能够扩展到巨大的存储规模。 如果你已经准备好开始，可以按照[这里](https://developers.cloudflare.com/k2/get-started/)的指南，在几秒内创建你的第一个流。 ## 边缘上的事件流 我们最初开发 K2，是因为*我们*需要一个位于边缘的持久化缓冲区，最初是作为 [Basin Pipelines](https://developers.cloudflare.com/pipelines/) 的接入层。Pipelines 由一个[流处理引擎](https://www.arroyo.dev/)驱动，该引擎采用基于拉取的模型，这意味着必须由其他系统在事件被读取、转换并写入 R2 之前先存储这些事件。由于我们承诺一旦事件被接受进入 Pipelines 流就绝不会丢弃它们，因此这种存储必须是持久的——即在可能很长的时间内不会丢失数据。 这就是大多数公司会部署 Apache Kafka 的场景。然而，Pipelines 运行在 Cloudflare 边缘网络上，该网络跨越了分布在超过 335 个城市中的大量服务器。我们独特的架构意味着我们往往无法运行像 Kafka 这样的传统分布式系统软件，需要重新思考这些系统的构建和运行方式。 对于有状态服务而言，Cloudflare 的全球基础设施尤其带来一些挑战：我们只能获得相对较小的机器资源切片，这些机器相对短暂，网络连接通常走公共互联网。但我们的基础设施也有几个超能力：无论用户身处世界何处，我们都能与他们保持近距离，并且拥有令人难以置信的水平扩展能力。 在设计最终成为 K2 的持久化缓冲系统时，我们决定依赖我们已有的强大状态原语：R2。像 R2 这样的对象存储系统将极其持久的存储（11 个 9！https://developers.cloudflare.com/r2/reference/durability/）与强一致性 API 结合在一起。将复制和共识委托给存储层，可以使应用层（这里是 K2）变得极其简单、成本更低、性能更高。另一个好处是将计算与存储分离，这意味着两者可以独立扩展。这使我们能够以低成本存储海量历史数据。 我们如何在对象存储之上构建日志？一个直接的问题是，R2 —— 与其他对象存储一样 —— 不支持追加操作，而这是日志的标准操作。相反，我们必须写入完整的文件或段（segment），这些文件需要足够大以摊薄每次写入和读取的成本。我们的做法是：首先在一个边缘服务上将写入累积在内存中。等待一小段时间让数据到达后，我们将所有事件写入为一个段文件。我们利用 R2 的原子操作来实现顺序性和严格递增的偏移量，无需单独的协调服务。 虽然基于 R2 构建有许多优势，但也有一个缺点：产生延迟更高。写入对象存储比写入本地磁盘慢，而且我们必须等待本地批处理累积完成才能开始写入。在 K2 的初始版本中，这会在第 99 百分位响应时间上增加约 1 秒的产生延迟。 我们将在即将发布的技术深度解析文章中分享更多关于 K2 设计的细节。 ## 流、队列还是管道？ Cloudflare 已有多个现有的异步投递原语，包括 [Queues](https://developers.cloudflare.com/queues/) 和 [Basin Pipelines](https://developers.cloudflare.com/basin-pipelines/)。在什么情况下应该选择 K2 而不是这些现有产品呢？ Queues 和 K2 流之间有一些表面相似之处：两者都接收事件、持久存储事件并将它们投递给消费者。Queues 的设计初衷是跟踪需要异步完成的昂贵或耗时的单项工作。例如，图像处理应用可以将用户请求入队，交由实际的图像处理服务处理。它们支持在单个工作项粒度上的复杂逻辑，如重试、延迟和死信队列（用于失败的尝试）。 相比之下，K2 是为大规模数据移动、长期保留和扇出消费而设计的。消息以批量形式生产和消费——通过牺牲消息级重试来实现高效处理。这种批处理也会带来比队列更高的产生延迟。 Basin Pipelines 是一个无服务器接入服务。你可以发送 Pipeline JSON 事件，这些事件可以被转换并写入 R2 或 Basin Catalog。当最终结果是将事件写入对象存储或 Iceberg 表时，我们推荐使用 Pipelines；当需要自定义处理或写入其他目标时，推荐使用 K2。 ## 快速上手 使用 K2 需要先创建一个流。你可以在账户中创建多个流，用于不同的用例或事件类型。流可以通过 [cf](https://blog.cloudflare.com/cloudflare-cf-cli-launch/)、Wrangler、控制台或 API 创建。 以收集和处理产品分析数据为例。首先，我们用 cf 创建一个流： ``` $ cf k2 streams create --name app_events --http-enabled { "id": "d78b09ee1f50430e9ec92a8af92b0231", "name": "app_events", "retention_seconds": 604800, "endpoint": "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com", "http": { "enabled": true, "authentication": false }, "worker_binding": { "enabled": true }, "created_at": "2026-09-28T15:14:39.053Z", "modified_at": "2026-09-28T15:14:39.053Z" } ``` 有了流之后，我们就可以通过 HTTP API 或 Worker 绑定向它写入数据。例如，从 Worker 中： ``` const result = await env.EVENTS.send([ { content: new TextEncoder().encode( JSON.stringify({ event: "page_view", path: new URL(request.url).pathname, timestamp: Date.now(), }), ), headers: { "content-type": "application/json" }, }, ]); if (!result.success) { console.error(`Produce failed: ${result.error.message}`); return new Response("Failed to record event", { status: result.error.retryable ? 503 : 500, }); } ``` K2 以字节形式表示数据，因此你可以使用适合你应用的任何格式或编码方式。 现在事件已经存在于流中，我们可以创建一个订阅。订阅在消费者之间分配工作，实现*读取并行性*——扩展到多个读取器，以处理单台服务器无法承受的更多负载。 我们可以通过 HTTP API 创建一个订阅： ``` $ curl -X POST "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com/subscriptions" \ -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \ -H "Content-Type: application/json" \ --data '{ "name": "analytics_processor", "start_at": { "type": "earliest" } }' ``` ``` { "result": { "id": "ee13f761783d3823a447a47b572ebf76" }, "success": true, "errors": [], "messages": [] } ``` 订阅创建好之后，我们就可以从每个消费者轮询它： ``` $ curl -X POST "https://4d8f5394e3e733debdeca9c65c5b7439.k2.cloudflarestorage.com/subscriptions/ee13f761783d3823a447a47b572ebf76/consume" \ -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \ -H "Content-Type: application/json" \ --data '{ "worker_id": "analytics-1", "max_records": 100 }' ``` ``` { "result": { "batch_id": "b7e4c9210a3f468d95c2e1068fdb734a", "leased_until_ms": 1790633929437, "records": [ { "timestamp_ms": 1790633629168, "content": "eyJldmVudCI6InBhZ2VfdmlldyJ9", "headers": { "content-type": "application/json" } }, ... ] }, "success": true, "errors": [], "messages": [] } ``` 当客户端调用 `consume` 时，它会获得该批事件时长为 5 分钟的*租约（lease）*。客户端可以执行以下三种操作之一： - *确认（ack）*该批次，标记其已处理完毕，确保不会被重新投递 - *否认（nack）*（negative ack），表示处理失败，希望该批次被重新投递 - *延长（extend）*其租约，以防处理需要更多时间 ``` $ curl -X POST "https://d78b09ee1f50430e9ec92a8af92b0231.k2.cloudflarestorage.com/subscriptions/ee13f761783d3823a447a47b572ebf76/batches/b7e4c9210a3f468d95c2e1068fdb734a/ack" \ -H "Authorization: Bearer ${CLOUDFLARE_API_TOKEN}" \ -H "Content-Type: application/json" \ --data '{ "worker_id": "analytics-1" }' ``` 这是从 K2 消费的一种方式：将工作分配给多个消费者，使每个消费者获取数据的一部分。另一种读取方式是为每个消费者创建独立的订阅——即发布/订阅（pub/sub）模式——在这种情况下，每个消费者都会看到所有消息。你也可以在两种方式之间灵活组合，维护多个独立的消费者池。 完整 API 详情请参见 [K2 文档](https://developers.cloudflare.com/k2/)。 ## 定价与可用性 K2 今天面向拥有 Workers Paid 订阅的账户在公开测试版中上线，限额如下： - 最高 10GB 存储使用量 - 每个流最高 30 MB/s 的产生速率 如果你需要更高的限额，请通过 Discord 联系我们的团队，或填写[限额提升申请表](https://docs.google.com/forms/d/e/1FAIpQLSf1_eGOxXFGsh4O72fqdvDbYBjzpV3Q6Vf5X3fqpoTHObGFIA/viewform)。 在测试期间，K2 的使用不会计费。我们开始计费后，预计定价如下： 定价 | 项目 | 价格 | | --- | --- | | 产生的数据 | $0.04 / GB | | 消费的数据 | $0.04 / GB | | 保留的数据 | $0.02 / GB / 月 | ## 未来规划 未来几个月，我们对 K2 有一个激动人心的路线图，包括： - 更高的写入并行性，流吞吐量可达数 GB/s - 消息键（message keys）和基于键的顺序保证 - 基于推送的 Worker 消费者 - Express 层级，提供更低的产生延迟和端到端延迟 - 对 Apache Kafka 客户端的即插即用支持 我们非常期待看到你在 K2 上构建的精彩作品！欢迎在 [Cloudflare Discord](https://discord.com/invite/cloudflaredev) 上分享你的反馈。