实时流JavaScript教程

使用 Server-Sent Events 实时推送金属价格

用 Server-Sent Events 实时推送金银价格：详解 /stream 端点与事件格式、EventSource 与 Node.js 客户端、断线重连、配额消耗及实时行情条的实现。

作者：MetalAPI Team·发布于 2026年10月25日·阅读约 10 分钟

流式推送还是轮询？

轮询是按定时器主动请求价格；流式推送则保持一个 HTTP 连接，由服务器在新价格发布时主动推送。MetalAPI 的实时流采用 Server-Sent Events（SSE）：一个内容类型为 text/event-stream 的普通 HTTP 响应，连接保持打开并持续发送具名事件。它不需要 WebSocket 库，能顺利穿过大多数代理，浏览器也通过 EventSource 原生支持。

流式推送是否划算，取决于您套餐的更新间隔。无论价格变化多频繁，一个数据流在建立连接时计 1 次请求，保持连接期间每分钟再计 1 次请求；而轮询每调用一次就计 1 次请求。以单个客户端在 31 天的月份里全天候监控价格为例：

套餐	更新间隔	按该间隔轮询	一个常开的数据流
Pro	1 分钟	44,640 次请求	约 45,400 次请求
Ultra、Business	15 秒	178,560 次请求	约 45,400 次请求
Enterprise	5 秒	535,680 次请求	约 45,400 次请求

数据流的数字已包含每小时重连一次所消耗的请求（详见下文）。在 Pro 套餐中，流式推送与每分钟轮询的成本大致相同；数据流的优势在于新价格一经发布即可送达，无需反复调整轮询逻辑。在 Ultra、Business 和 Enterprise 套餐中，数据流还比全速轮询便宜得多。

以下场景不适合使用流式推送：价格只需在每次页面浏览时准确一次、夜间批处理任务，以及阈值检查。前两种请使用带缓存的 /latest 调用，第三种请使用价格提醒与 Webhook。

/stream 端点

http

GET https://metalapi.com/api/v1/stream?base=USD&currencies=XAU,XAG,XPT,XPD

数据流接受与 /latest 相同的 base、currencies（或 symbols）和 unit 参数（参见 latest 端点文档），身份验证方式也相同：查询字符串中的 api_key、Authorization: Bearer 请求头或 X-API-Key 请求头。参数会在数据流打开之前完成校验，因此未知的品种或单位会返回普通的 JSON 错误，而不是一个空的数据流。

套餐	数据流价格更新频率	每月请求数
Pro	每 60 秒	100,000
Ultra	每 15 秒	500,000
Business	每 15 秒	500,000
Enterprise	每 5 秒	1,500,000

较低套餐会收到错误 106（HTTP 403），提示实时流仅适用于 Pro 及以上套餐。每个连接还受以下限制：

每个 API 密钥最多 5 个并发数据流。 第 6 个连接会被拒绝并返回 104（HTTP 429）。
每个连接最长持续 60 分钟。 一小时后，服务器会发送 reconnect 事件并关闭数据流，客户端随后重新连接。
每分钟重新校验一次密钥。 如果密钥已被吊销或每月配额已用完，服务器会发送 error 事件并关闭数据流。
建立连接同样计入每分钟速率限制，与其他请求无异，因此陷入快速重连循环的客户端会开始收到 104 错误。

服务器发送的内容

以下是一个数据流开头部分的原始文本（价格仅作示意）：

text/event-stream

retry: 5000
event: prices
data: {"timestamp":1792834892,"date":"2026-10-24","base":"USD","rates":{"XAU":0.0002501319,"USDXAU":3997.89,"XAG":0.0185392857,"USDXAG":53.94}}

: ping 1792834907000

event: prices
data: {"timestamp":1792834952,"date":"2026-10-24","base":"USD","rates":{"XAU":0.0002500812764,"USDXAU":3998.7,"XAG":0.01852194851,"USDXAG":53.99}}

事件	数据	处理方式
`prices`	完整快照：`timestamp`、`date`、`base`、`rates`，不以金衡盎司计价时还包含 `unit`	替换当前显示的价格
`reconnect`	`{"reason":"max connection time reached"}`	无需处理；数据流关闭后重新连接即可
`error`	`code`、`type` 和 `info`，错误代码与 REST API 相同	根据错误代码停止或退避重试
`: ping …`	每 15 秒发送一行注释	解析器会忽略它；可将其视为连接存活的信号

有三个特性决定了客户端代码的写法。每个 prices 事件都是所请求品种的完整快照，而不是增量，因此客户端重连后只要收到第一个事件就能恢复到最新状态。第一个事件在建立连接后立即送达。此后只有在汇率变化时才会发送事件：服务器按您套餐的间隔检查，没有变化时保持静默，这在周末的金属市场很常见。有心跳的静默是正常的；连心跳都没有的静默则意味着连接已经断开。

事件不带 id，因此 Last-Event-ID 不起作用。也不存在需要重放的内容，因为每个快照都会替换之前的全部数据。rates 的键名遵循 REST 格式：USDXAU 是 1 金衡盎司黄金的美元价格，XAU 是它的倒数。

在浏览器中使用 EventSource

EventSource 会替您处理连接、解析和重连。它无法发送自定义请求头，因此密钥只能放在查询字符串中。请使用限制为您自己域名的密钥：浏览器会随请求发送 Origin 请求头，受限密钥会以错误 107 拒绝从其他网站打开的数据流。

浏览器

// 请使用限制为您自己域名的密钥：任何打开该页面的人都能看到它。
const url = "https://metalapi.com/api/v1/stream?api_key=YOUR_RESTRICTED_KEY&base=EUR&currencies=XAU,XAG&unit=gram";
const FATAL = [101, 102, 103, 105, 106, 107];
const es = new EventSource(url);

es.addEventListener("prices", (e) => {
  const { timestamp, base, rates } = JSON.parse(e.data);
  document.querySelector("#gold").textContent = rates[`${base}XAU`].toFixed(2);
  document.querySelector("#silver").textContent = rates[`${base}XAG`].toFixed(3);
  document.querySelector("#updated").textContent = new Date(timestamp * 1000).toLocaleTimeString();
});

es.addEventListener("error", (e) => {
  if (e instanceof MessageEvent && e.data) {
    // MetalAPI 在关闭数据流之前发送：{"code":105,"type":"quota_exceeded",...}
    const err = JSON.parse(e.data);
    console.warn("MetalAPI stream error", err.code, err.info);
    if (FATAL.includes(err.code)) es.close(); // 重连也会再次失败
  } else if (es.readyState === EventSource.CLOSED) {
    console.warn("Stream refused; check the key and plan with /usage");
  }
  // 否则就是连接中断，浏览器会自行重连。
});

error 监听器会收到两类不同的事件。连接问题以普通 Event 的形式到达，EventSource 会在服务器 retry 行设定的 5 秒后自行重连。MetalAPI 发送的错误则以 MessageEvent 的形式到达，其 data 中是 JSON 格式的错误，服务器随即关闭数据流。对于重试也无法解决的错误代码，请调用 close()；否则浏览器会不断重连，也会不断被拒绝。

如果第一个响应就是 HTTP 错误，例如因套餐不包含实时流而返回 106，EventSource 既不会暴露响应正文，也不会重试：readyState 会直接变为 CLOSED。需要排查原因时，可通过免费的用量端点查看该密钥所属的套餐。

在浏览器中直接连接数据流，适合只有少数用户的内部看板。每个打开的标签页都是一个独立的数据流：每分钟消耗 1 次请求，并计入每个密钥 5 个并发数据流的上限。对于公开页面，请在服务器上运行一个数据流，再分发给所有访问者，就像下文的行情条那样。

在 Node.js 中使用 fetch

在服务器端，密钥可以放在请求头中，重连也完全由您掌控。Node.js 18 及更高版本内置了支持流式响应正文的 fetch，因此一个完整的客户端只需一个模块，无需任何依赖：

metal-stream.ts

// metal-stream.ts — Node.js 18+，无任何依赖
export interface Prices {
  timestamp: number;
  date: string;
  base: string;
  unit?: string;
  rates: Record<string, number>;
}

const STREAM_URL = "https://metalapi.com/api/v1/stream?base=USD&currencies=XAU,XAG,XPT,XPD";
const FATAL = new Set([101, 102, 103, 105, 106, 107]); // 重试也无济于事
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));

class FatalStreamError extends Error {}

export async function streamPrices(onPrices: (p: Prices) => void): Promise<never> {
  let retryMs = 5_000; // 服务器发送 `retry: 5000`
  let failures = 0;
  for (;;) {
    const ctrl = new AbortController();
    let watchdog: ReturnType<typeof setTimeout> | undefined;
    const kick = () => {
      clearTimeout(watchdog);
      watchdog = setTimeout(() => ctrl.abort(), 45_000); // 错过 3 次心跳
    };
    try {
      const res = await fetch(STREAM_URL, {
        headers: { Authorization: `Bearer ${process.env.METALAPI_KEY}`, Accept: "text/event-stream" },
        signal: ctrl.signal,
      });
      if (!res.ok || !res.body) {
        const body = await res.json().catch(() => null);
        const code = body?.error?.code;
        throw FATAL.has(code) ? new FatalStreamError(`${code}: ${body.error.info}`) : new Error(`HTTP ${res.status}`);
      }
      kick();
      const reader = res.body.getReader();
      const decoder = new TextDecoder();
      let buffer = "";
      for (;;) {
        const { value, done } = await reader.read();
        if (done) break;
        kick();
        buffer += decoder.decode(value, { stream: true }).replace(/\r\n?/g, "\n");
        let end: number;
        while ((end = buffer.indexOf("\n\n")) !== -1) {
          const block = buffer.slice(0, end);
          buffer = buffer.slice(end + 2);
          let event = "message";
          const data: string[] = [];
          for (const line of block.split("\n")) {
            if (!line || line.startsWith(":")) continue; // 心跳注释
            const i = line.indexOf(":");
            const field = i === -1 ? line : line.slice(0, i);
            const val = i === -1 ? "" : line.slice(i + 1).replace(/^ /, "");
            if (field === "event") event = val;
            else if (field === "data") data.push(val);
            else if (field === "retry" && /^\d+$/.test(val)) retryMs = Number(val);
          }
          if (event === "prices") {
            failures = 0;
            onPrices(JSON.parse(data.join("\n")));
          } else if (event === "error") {
            const err = JSON.parse(data.join("\n"));
            if (FATAL.has(err.code)) throw new FatalStreamError(`${err.code}: ${err.info}`);
          }
          // `reconnect`：服务器接下来会关闭连接，由外层循环重新连接。
        }
      }
    } catch (err) {
      if (err instanceof FatalStreamError) throw err;
      failures++;
      console.warn("stream interrupted:", (err as Error).message);
    } finally {
      clearTimeout(watchdog);
    }
    // 正常关闭后等待 5 秒；持续失败时逐步退避，最长间隔 1 分钟。
    const backoff = Math.min(60_000, retryMs * 2 ** Math.min(failures, 4));
    await sleep(failures ? backoff + Math.random() * 1_000 : retryMs);
  }
}

可在任意脚本或服务中使用它：

watch.ts

import { streamPrices } from "./metal-stream";

streamPrices((p) => {
  console.log(new Date(p.timestamp * 1000).toISOString(), "gold", p.rates.USDXAU, "silver", p.rates.USDXAG);
}).catch((err) => {
  console.error("Stream stopped:", err.message); // FATAL 错误代码：请检查密钥、套餐或配额
  process.exit(1);
});

该解析器遵循 SSE 格式：事件之间以空行分隔，event: 指定事件名称，多行 data: 以换行符拼接，以冒号开头的行是注释。任何允许设置请求头的 SSE 客户端库同样适用，只要它遵循下一节中的重连规则。

重连与故障处理

上面的模块遵循一组简短的规则，任何数据流客户端都应具备：

情况	信号	应对方式
每小时轮换	收到 `reconnect` 事件，随后数据流结束	按服务器的 `retry` 延迟（5 秒）重新连接
网络中断	读取出错，或数据流在没有 `reconnect` 的情况下结束	使用带随机抖动的指数退避重连，间隔最长 60 秒
静默连接	45 秒内既无数据也无心跳	中止并重连：已错过三次心跳
临时拒绝	`104`、`500`、`503`	退避后重试
永久拒绝	`101`、`102`、`103`、`105`、`106`、`107`	停止重连并通知相关人员

看门狗的作用比看上去更重要。一个没有收到 TCP 重置就已失效的连接（例如位于 NAT 网关或负载均衡器之后），在操作系统察觉之前可能会在几分钟内看起来仍处于打开状态。有了每 15 秒一次的心跳，客户端可以在 45 秒内发现问题。

当心重连循环：每次成功建立连接都会消耗 1 次请求，每次尝试都会计入每分钟速率限制。当某个环节持续失败时，指数退避可以把损失降到最低。

配额用尽（105）值得单独设置告警。数据流会在下一次每分钟检查时以 error 事件报告这一情况；在配额于每月 1 日（UTC）重置或套餐升级之前，重连都无济于事。

构建实时行情条

公开的价格行情条不应为每位访问者都打开一个 MetalAPI 数据流。更好的做法是：由一个小型中继服务在服务器上保持一个数据流，再通过它自己的 SSE 端点把每个快照推送给所有浏览器。这样您的用量只取决于中继的数量，与访问者数量无关，密钥也永远不会离开服务器：

relay.ts

// relay.ts — 服务器上只保持一个 MetalAPI 数据流，服务任意数量的浏览器。
import { readFileSync } from "node:fs";
import http from "node:http";
import { streamPrices } from "./metal-stream";

const page = readFileSync(new URL("./ticker.html", import.meta.url));
const clients = new Set<http.ServerResponse>();
let latest: string | null = null;

const broadcast = (chunk: string) => {
  for (const res of clients) res.write(chunk);
};

streamPrices((p) => {
  latest = JSON.stringify({ timestamp: p.timestamp, base: p.base, rates: p.rates });
  broadcast(`event: prices\ndata: ${latest}\n\n`);
}).catch((err) => {
  console.error("upstream stream stopped:", err.message);
  process.exit(1);
});

setInterval(() => broadcast(": ping\n\n"), 15_000);

http.createServer((req, res) => {
  if (req.url === "/") return void res.writeHead(200, { "Content-Type": "text/html" }).end(page);
  if (req.url !== "/prices") return void res.writeHead(404).end();
  res.writeHead(200, {
    "Content-Type": "text/event-stream",
    "Cache-Control": "no-store",
    "X-Accel-Buffering": "no",
  });
  res.write("retry: 3000\n\n");
  if (latest) res.write(`event: prices\ndata: ${latest}\n\n`); // 新访问者立即获得价格
  clients.add(res);
  req.on("close", () => clients.delete(res));
}).listen(3000, () => console.log("Ticker on http://localhost:3000"));

中继会立即向每位新访问者发送最新快照，只转发时间戳、基础货币和汇率，并自行发送心跳。页面只需几行代码即可渲染行情条，并根据与上一次更新相比的涨跌方向为每个价格着色：

ticker.html

<!doctype html>
<meta charset="utf-8">
<title>Live metal prices</title>
<style>
  .ticker { display: flex; flex-wrap: wrap; gap: 1.5rem; font: 15px/1.4 system-ui, sans-serif; font-variant-numeric: tabular-nums; }
  .up { color: #15803d; }
  .down { color: #dc2626; }
</style>

<div id="ticker" class="ticker">Loading prices…</div>

<script>
  const NAMES = { XAU: "Gold", XAG: "Silver", XPT: "Platinum", XPD: "Palladium" };
  const previous = {};
  const ticker = document.getElementById("ticker");

  new EventSource("/prices").addEventListener("prices", (e) => {
    const { base, rates, timestamp } = JSON.parse(e.data);
    const money = new Intl.NumberFormat(undefined, { style: "currency", currency: base });
    ticker.replaceChildren(
      ...Object.entries(NAMES).map(([code, name]) => {
        const price = rates[base + code];
        const el = document.createElement("span");
        if (previous[code] !== undefined && price !== previous[code]) {
          el.className = price > previous[code] ? "up" : "down";
        }
        previous[code] = price;
        el.textContent = `${name} ${money.format(price)}`;
        return el;
      }),
    );
    ticker.title = `Prices as of ${new Date(timestamp * 1000).toLocaleString()}`;
  });
</script>

使用 METALAPI_KEY=mk_live_... npx tsx relay.ts 启动，然后打开 http://localhost:3000。行情条会将每个价格与上一个事件进行比较。如果想改为显示当日涨跌，可从 OHLC 端点获取一次今日开盘价，再将每个新价格与之比较。

生产环境注意事项

关闭代理缓冲。 MetalAPI 会发送 X-Accel-Buffering: no 和 Cache-Control: no-transform。如果您的中继前面还有 nginx 或其他代理，请为 SSE 路由关闭响应缓冲（nginx 中为 proxy_buffering off;），否则事件会成批到达。
不要压缩事件流。 压缩中间件可能会扣留小数据块，直到其缓冲区填满为止。请将 text/event-stream 响应排除在压缩之外。
将中继作为常驻进程运行。 Serverless 函数通常有执行时间限制，不适合全天候保持数据流连接。
显示数据时间。 显示事件中的 timestamp，而不是浏览器收到事件的时间，这样访问者才能区分周末的平静行情和停滞的数据源。
避免干扰屏幕阅读器。 不要将快速更新的行情条放在 aria-live 区域中，逐条播报每次更新只会造成干扰。
规划好配额。 一个常开的数据流每月约消耗 45,400 次请求，因此 Pro 套餐可以容纳两个数据流，外加常规 API 调用。可通过用量端点免费查看用量。

API 文档列出了数据流的全部参数，定价页面则列出了包含该功能的套餐。

用您自己的数据试一试

创建免费账户，两分钟内即可发起第一次请求。每月 100 次请求，无需信用卡。

免费获取 API 密钥阅读文档