Cloudflare开发工具基础设施

Cloudflare 发布 K2：无服务器事件流

K2 把生产者与消费者在扩展和时间上解耦：边缘持久缓冲、多消费者各自独立读取、按 key 保序，且兼容 Kafka 客户端。文章讲它跟 Queues、Pipelines 的分工与适用边界。

Cloudflare Blog6 分钟阅读

中文
中文日本語
复制
Cloudflare 风格的插画：橙色放大镜里是一座雪山，四周有数据流线条、彩色方块与节点

在传统的远程过程调用（RPC）架构中，有一个核心难题：生产者和消费者必须在规模和节奏上保持一致。如果生产者发送的数据超出消费者的处理能力，或者消费者及下游服务不可用，事件就会丢失。当多个消费者需要各自独立处理数据时，这个问题会更加严重。例如，一个电商后端会在交易完成时发出事件，这些事件需要被分析系统和欺诈检测服务读取。解决办法是解耦生产者和消费者——在中间插入一个服务，由它承接写入，同时让各个读取方按自己的节奏独立消费。今天我们推出 Cloudflare K2 公测版，正是为了解决这个问题。K2 是 Developer Platform 上的持久化事件流原语。你把事件发送到 K2 流，它会将这些事件存储为有序日志。消费者可以用多种方式读取，例如把读取任务拆分到一组消费者上，或者把全部消息投递给所有消费者。它完全无服务器，可扩展到海量数据，并支持长期保留，因此即使消费者长时间停机也不会丢失数据。在底层，K2 在 R2 对象存储之上实现了分区、持久化的日志，从而能够扩展到巨大的存储规模。如果你准备开始使用，可以按照这里的指南在几秒内创建你的第一个流。

边缘上的流

我们最初做 K2，是因为我们需要一个边缘上的持久缓冲，一开始是为了给 Basin Pipelines 做摄取层。Pipelines 由一个流处理引擎驱动，采用拉取模型，也就是说必须有另一个系统先把事件存起来，之后才能读取、转换并写入 R2。而一旦事件被 Pipelines Stream 接收，我们就承诺绝不丢弃，所以这个存储必须持久——也就是不能丢数据——而且要能撑过可能很长的时间。大多数公司到这一步会部署 Apache Kafka。但 Pipelines 跑在 Cloudflare 边缘上，横跨 335 多个城市的大量服务器。我们这套独特的架构意味着往往跑不了 Kafka 这类传统的分布式系统软件，必须重新思考这类系统该怎么构建和运维。对有状态服务来说尤其如此，Cloudflare 的全球基础设施带来了一些挑战：我们拿到的机器切片相对较小，这些机器相对短命，网络还常常走公网。但我们的基础设施也有几项超能力：无论用户身在何处都离他们很近，并且有惊人的横向扩展能力。在设计后来成为 K2 的这套持久缓冲系统时，我们决定依赖已有的强大状态原语：R2。像 R2 这样的对象存储系统，把极高持久性的存储（11 个 9！）和强一致的 API 结合在一起。把复制和共识下推给存储层，能让应用层（这里是 K2）变得简单得多、便宜得多、性能也高得多。附带的好处是计算与存储分离，两者可以各自独立扩展。这让我们能以低成本存下海量历史数据。怎么在对象存储之上构建日志？一个直接的问题是，R2 和其他对象存储一样不支持 append，而 append 是日志的标准操作。我们只能写完整的文件，也就是 segment，而且要足够大，才能摊薄每个文件读写带来的开销。做法是先在边缘服务的内存里累积写入。等一小段时间让数据到齐，再把所有事件写成一个 segment 文件。我们借助 R2 的原子操作实现有序和严格递增的 offset，不需要额外的协调服务。基于 R2 构建虽然好处很多，但有一个缺点：写入延迟更高。写对象存储比写本地磁盘慢，而且必须先等本地批次攒够才能开始写。在 K2 的首个版本里，这加起来大约是 99 分位响应时间下 1 秒的写入延迟。关于 K2 的设计细节，我们会在后续的技术深潜文章里分享更多。

Streams、Queues 还是 Pipelines？

Cloudflare 已有若干异步投递原语，包括 Queues 和 Basin Pipelines。什么时候该用 K2，而不是这些现成的产品？Queues 和 K2 Streams 表面上有些相似：都接收事件、持久化存储，再投递给消费者。Queues 的设计围绕追踪单个任务项——那些昂贵或耗时、需要异步完成的工作。比如图像处理应用可以把用户请求入队，交给真正的图像处理服务去处理。它支持针对单个工作项的复杂逻辑，比如重试、延迟，以及失败后的死信队列。K2 则不同，它面向的是大规模数据搬运、长期保留和扇出消费。消息按批次生产和消费——换来高效处理，代价是没法做消息级重试。这种批处理也让生产者的延迟高于 Queues。Basin Pipelines 是无服务器摄取服务。你把 Pipeline 的 JSON 事件发过去，经过转换后写入 R2 或 Basin Catalog。如果最终目的是把事件写进对象存储或 Iceberg 表，我们推荐 Pipelines；如果要做自定义处理，或写到其他目的地，就用 K2。

上手

用 K2 得先创建 stream。一个账号下可以有多个 stream，对应不同的用例或事件类型。创建方式有 cf、Wrangler、控制台或 API。以收集和处理产品分析数据为例。先用 cf 创建一个 stream：

$ 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"
}

有了 stream，就可以往里生产数据了，通过 HTTP API 或 Worker binding 都行。比如在 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 把数据表示为字节，所以格式和编码随你的应用需要来定。

现在 stream 里有了事件，就可以创建 subscription 了。Subscription 把工作分配给多个消费者，实现 读并行——横向扩展到多个读取者，承受单台服务器扛不住的负载。

我们可以通过 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 时，会拿到这批事件的一份 lease，有效期 5 分钟。客户端可以做三件事之一：

  • ack，把这批事件标记为已处理，确保不会被重新投递
  • nack（negative ack），表示处理失败，希望重新投递
  • extend，延长 lease，用于处理时间不够的情况
$ 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 文档。

定价与可用性

K2 今天起进入公开测试，拥有 Workers Paid 订阅的账户即可使用，限制如下：

  • 存储用量上限 10GB
  • 每个 stream 写入 30 MB/s

如果需要更高的限额，请在 Discord 上联系团队，或填写限额提升表单。

测试期间 K2 的使用不收费。开始计费后，我们预计的定价如下：

定价
写入数据$0.04 / GB
读取数据$0.04 / GB
数据保留$0.02 / GB / 月

下一步

未来几个月 K2 有一份令人期待的路线图，包括：

  • 更高的写入并行度，stream 可达多 GB/s
  • 消息 key 以及基于 key 的顺序保证
  • 基于 push 的 worker 消费者
  • Express 层，更低的写入延迟和端到端延迟
  • 直接兼容 Apache Kafka 客户端

期待看到你在 K2 上做出什么！欢迎在 Cloudflare Discord 上分享反馈。

来源: Cloudflare Blog← 返回首页