Streamline：用 Cloudflare Stream 和 Workers 构建自定义视频管线-未完纪

原文标题：Streamline: custom video pipelines with Cloudflare Stream and Workers
作者：Willi Geiger、Taylor Smith
来源：Cloudflare Blog，2026 年 10 月 2 日
原文：https://blog.cloudflare.com/streamline/

Cloudflare Stream 可以满足许多常规视频托管与直播需求；但如果希望在直播画面上动态叠加标注，或把字幕直接烧录进已有视频，就需要一条能够自行控制处理过程的视频管线。Cloudflare 在这篇文章中介绍了 Streamline：一个由 Workers、Containers 和 Durable Objects 组合而成的开发示例，用于处理视频并把结果输出为直播或新的托管视频。

为什么把媒体处理与请求分开

视频处理可能持续几分钟甚至数小时。若把处理过程绑定到发起它的单个 HTTP 请求，客户端一旦断开，任务的生命周期就很难管理。一个实用的管线应当允许应用启动任务、提供输入、查看状态和预览，并在需要时停止任务，同时让媒体处理在控制端暂时离线时继续运行。

Streamline 把职责分给三种运行时：

Container 运行长时间媒体任务，承载媒体引擎及其所需的编译程序。
Durable Object 协调会话、容器生命周期和预览中继。
Worker 提供控制信号、监控接口，并承载用户界面或其他调用方。

媒体引擎在容器中运行。示例里的控制器是 Go 编写的 HTTP 服务，负责把请求转换成媒体引擎操作；当前处理器使用 FFmpeg，但原文明确说这是内部实现细节，不属于面向使用者的 API。这个模块化边界也让媒体引擎将来可以换成专用编码服务。Worker 与容器的连接即使中断，已经启动的处理仍可继续。

在部署架构中，控制应用可以是浏览器全栈应用、代理程序或嵌入式设备。用户界面负责客户端逻辑、身份和访问策略；Durable Object 负责协调任务。容器可以从 Stream Live 输入拉取 RTMPS、向另一个 Stream Live 输入推送 RTMPS，也能读取 Cloudflare Stream 的 HLS 清单和分片、接受控制应用上传的视频，并通过 WebSocket 中继发送预览。

本地开发时，容器可由本地 Docker 实例代替，不需要 Durable Object；示例按单用户方式运行，控制应用没有访问授权，预览直接连接 localhost。部署到 Cloudflare 后，访问 Worker 的获准用户可以创建会话，系统按需启动 Streamline 容器，并在 Stream 与处理管线之间转发媒体。这个本地模式与部署模式的授权和隔离能力不同，不能把本地无访问控制的设置直接当作线上安全配置。

会话生命周期与客户端 API

控制端创建长任务后可以断开并重新连接，而容器会持续处理，直到控制应用停止会话。示例还为会话设置最长运行时间，避免失去外部控制后无限运行。同一容器实例在某个会话运行时不能被其他应用同时占用。

Cloudflare Container 平时会在一段时间没有请求后休眠；持续处理的媒体任务却不能因为控制端断开而停掉。示例通过覆盖 onActivityExpired() 回调来处理：在锁内读取当前中继会话，若仍有未到期的 expiresAt 就续期活动超时；否则销毁容器。

async onActivityExpired() {
  await this.withControlLock(async () => {
    const session = await this.getRelaySessionLocked()

    if (session?.expiresAt) {
      this.renewActivityTimeout()
      return
    }

    await this.destroy()
  })
}

Streamline 导出两类包：@cloudflare/streamline/client 提供高层会话 API；@cloudflare/streamline/ 暴露与容器关联的 Durable Object 基类，负责路由 API 请求、预览中继，并提供安全和访问策略的扩展点。远程部署中的 Worker 可以导入 @streamline/cloudflare，定义应用专用的 Durable Object 子类和存储逻辑。本地模式没有 Durable Object，前端用薄适配层维持相同的会话 API，再直接连接本地 Docker。

一次最简单的调用是创建客户端、创建会话，再启动配置好的管线：

const streamline = createStreamline({ baseUrl: 'https://media.example' })
const session = await streamline.sessions.create()
const result = await session.start(config)

// 管线已启动，除非启动失败。
console.log(result)

客户端 API 还提供 streamline.sessions.resume(id) 重新连接既有会话，以及 session.ingest(chunk) 上传摄像头输入、session.annotation(png) 更新透明标注、session.metrics() 读取会话指标和 session.stop() 停止处理。

用 JSON 配置输入、操作与输出

session.start() 接受一个 JSON 配置，分成输入、处理操作和输出。以下例子从 RTMP 接入直播，在右上角叠加透明图片，再以 H.264 编码后推送到另一个 RTMP 配置：

const session = await streamline.sessions.create()
const result = await session.start({
  input: { type: 'rtmp', profile: 'primary-input' },
  pipeline: [
    {
      op: 'overlay',
      params: { image: '/app/assets/cf-logo.png', position: 'top-right' },
    },
    {
      op: 'encode',
      params: {
        codec: 'h264',
        preset: 'fast',
        bitrate: '1500k',
        resolution: '1280x720',
        fps: 30,
      },
    },
  ],
  output: { mode: 'rtmp', profile: 'primary-output' },
})

点播视频也可通过 HLS 作为输入。示例读取 Cloudflare Stream 的视频清单，使用自动字幕来源把字幕烧录到画面，再编码并输出到 RTMP：

const streamVideoId = 'your-cloudflare-stream-video-id'
const session = await streamline.sessions.create()

await session.start({
  input: {
    type: 'hls',
    url: `https://videodelivery.net/${streamVideoId}/manifest/video.m3u8`,
  },
  pipeline: [
    { op: 'subtitle', params: { source: 'auto' } },
    {
      op: 'encode',
      params: {
        codec: 'h264',
        preset: 'fast',
        bitrate: '1500k',
        resolution: '1280x720',
        fps: 30,
      },
    },
  ],
  output: { mode: 'rtmp', profile: 'default' },
})

从浏览器摄像头发送视频并叠加动态标注

另一种输入方式是让控制应用把视频分块送进管线。浏览器可以用 getUserMedia() 取得摄像头和麦克风流，再用 MediaRecorder 每 250 毫秒产生一块 WebM 数据，并通过 session.ingest() 顺序上传：

const stream = await navigator.mediaDevices.getUserMedia({
  video: true,
  audio: true,
})
const recorder = new MediaRecorder(stream, {
  mimeType: 'video/webm;codecs=vp8,opus',
})
let uploadTail = Promise.resolve()

recorder.addEventListener('dataavailable', (event) => {
  if (event.data.size === 0) return
  uploadTail = uploadTail
    .then(() => session.ingest(event.data))
    .catch((error) => reportUploadFailure(error))
})

recorder.start(250)

摄像头输入的管线示例先创建会话和预览连接，再启动处理：亮度增加 0.1，翻转画面，使用名为 annotation 的叠加层，最后以 H.264、1280x720、每秒 30 帧编码为 fMP4 WebSocket 输出。编码示例还设置 gop: 60。

标注图层可在处理过程中更新。应用可以把 Canvas 转成 PNG Blob，再交给 session.annotation()；持续更新时，实际频率会受 PNG 大小、可用带宽和处理能力限制。

function canvasPng(canvas: HTMLCanvasElement): Promise<Blob> {
  return new Promise((resolve, reject) => {
    canvas.toBlob((blob) => {
      if (blob) resolve(blob)
      else reject(new Error('Canvas could not produce a PNG'))
    }, 'image/png')
  })
}

const png = await canvasPng(overlayCanvas)
await session.annotation(png)

Streamline 的 WebSocket 预览由容器把 fMP4 片段发送给 Durable Object，再由 Durable Object 转发到应用源站下的 /relay/view。浏览器必须在启动发布端之前连上中继，否则发布端会被拒绝。连接地址使用 wss: 或 ws:，并以 session_id 查询参数标识会话：

function openViewer(origin: string, sessionId: string): Promise<WebSocket> {
  const url = new URL('/relay/view', origin)
  url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:'
  url.searchParams.set('session_id', sessionId)

  return new Promise((resolve, reject) => {
    const socket = new WebSocket(url)
    socket.binaryType = 'arraybuffer'
    socket.addEventListener('open', () => resolve(socket), { once: true })
    socket.addEventListener('error', () => reject(new Error('Preview relay failed')), {
      once: true,
    })
  })
}

收到二进制消息后，浏览器可把 ArrayBuffer 片段追加到播放器的 SourceBuffer；字符串消息中的 {"type":"eos"} 表示流结束。生产环境的 MediaSource 播放器要在 SourceBuffer.updating 为真时排队等待，不能直接并发追加片段。

当前支持的处理操作

示例的 pipeline 数组只包含媒体引擎支持的操作。需要留意：当前执行顺序由引擎固定，数组中操作的排列顺序并不决定执行顺序。 原文列出的操作是：

filter：应用模糊、饱和度等滤镜。
overlay：叠加 URL 图片，或由 annotation() 单独发送的二进制 PNG。
subtitle：把字幕烧录到画面。
encode：设置输出编码参数。

安全设计与示例边界

安全设计需要同时保护会话控制、租户隔离、Stream Live 密钥和资源用量。文章里的所有者部署通过 Workers Access 保持私有：Worker 在接受控制请求前验证 Access 会话，并将活动会话绑定到已验证的身份。示例一次只运行一个会话；另一个身份不能停止或替换当前会话。该配置是私有、单例路由，不是面向公众的多用户服务安全模型。

Stream Live 输入和输出密钥保存在 Worker secrets 或 Durable Object 中只写的共享覆盖项里。设置 API 不会返回密钥，浏览器存储也不会保存密钥；控制应用只传命名配置，Worker 在联系容器前解析配置。预览流另有两种不同用途的凭据：Cloudflare Access 服务令牌用于容器工作负载向发布端认证；随机的会话级 capability 只授权当前活动中继发布。服务令牌由容器的出站 Worker 注入，不进入容器内存。初始部署曾使用临时的、限定路径的 Access Bypass，同时仍要求会话 capability；部署并通过 smoke test 后，再改为 Service Auth。

文章同时发布了开源实现与示例应用：媒体引擎和包导出位于 cloudflare/streamline，Worker 与 Astro 前端示例、部署配置及 Access 工具位于 cloudflare/streamline-demo。文章称代码以开源方式发布，但该页面没有单独列出文章文本的转载许可或代码仓库的具体许可证。公开 playground 地址为 playground.streamline-video.workers.dev。

这套实现当前使用容器 CPU 处理媒体，在高画质或高帧率下会形成瓶颈。作者提出的后续方向包括计算机视觉管线、硬件加速媒体处理、WebRTC 和 MoQ 等新协议，以及在 Workers 中原生提供视频编码与解码能力。上述内容描述的是原文发布时的实现与计划，不代表任何运行性能已在本稿中实测。

文章版权归作者所有，未经允许请勿转载。

THE END