WebSocket与QuickNode Streams - Quicknode

概述

在区块链开发中，实时数据对于构建响应式、交互式应用程序至关重要。传统上，开发者使用 WebSocket 来处理实时通信，尽管 WebSocket 在许多场景下有效，但在应用程序发展壮大时，它存在的局限可能会使得可扩展性和可靠性挑战重重。这就是 Streams 的用武之地。

QuickNode 的 Streams 是一个专门设计的解决方案，旨在应对大规模处理实时和历史区块链数据的挑战，同时提供附加功能以简化你的开发过程。

在本指南中，我们将探索 WebSocket 与 Streams 之间的相似点和差异，并帮助你理解 Streams 如何解决开发者在使用 WebSocket 时常遇到的一些问题。到最后，你将掌握做出明智决策所需的知识，以选择最适合你的应用程序实时数据需求的解决方案。

什么是 WebSocket？

WebSocket 是一种通信协议，提供客户端和服务器之间的实时双向数据传输。与传统的 HTTP 请求不同，WebSocket 维护持久连接，允许无需持续发起请求就能进行连续数据交换。

WebSocket 在区块链中的工作原理

在区块链环境中，WebSocket 通常用于订阅实时事件，如新块、交易或智能合约事件。它们能够实现低延迟通信，非常适合需要快速响应区块链事件的应用程序。

以下是简化的过程：

1. 客户端建立与区块链节点的 WebSocket 连接。
2. 客户端发送特定事件的订阅请求。
3. 节点在事件发生时向客户端发送实时更新。
4. 客户端实时处理传入数据。

什么是 Streams？

Streams 是一个区块链数据流和 ETL（提取、转换、加载）服务，旨在使实时和历史区块链数据对开发者易于访问。它允许你可靠地将实时区块链数据流或回填历史数据到你想要的目的地，无论它是 API（例如，Webhook）、兼容 S3 的存储桶还是数据库（例如，PostgreSQL、Snowflake）。

Streams 旨在简化区块链数据的管理，无论是实时还是历史，允许开发者专注于其应用逻辑。

Streams 的关键功能

• 实时数据流：实时接收区块链数据，无需持续轮询 JSON-RPC 端点。
• 历史数据回填：使用 预构建模板 回填历史区块链数据，并事先提供清晰的成本和完成时间估算。
• 可定制数据过滤器：通过在你的 Stream 上应用 自定义过滤器，仅获取对你重要的数据。这允许你使用 JavaScript 代码（ECMAScript）个性化从你的 Stream 传送的数据，从而制定规则或检测模式，例如跟踪事件、交易、解码数据等。
• 成本效率：Streams 的定价基于处理的数据量，为高流量应用提供可预测的成本。
• 多链支持：Streams 跨多个区块链网络工作，简化多链开发。
• 数据一致性：通过内置的可靠性机制，如重组处理，确保数据一致性，即使在网络重组的情况下也不会错过任何区块或交易。更多信息请见 此处。
• 压缩：通过压缩优化数据传输，尽量减少带宽使用。
• 函数集成：将 Streams 与 Functions 结合，创建强大的无服务器 API 或实时数据处理管道，而无需管理任何基础设施。

带过滤器的 Streams

带过滤器的 Streams 让你仅接收所需的区块链数据，减少数据量和成本。通过 JavaScript 过滤器，你可以在数据到达目的地之前对其进行修改。过滤器还包括对 关键值存储 的访问，允许你存储来自 Stream 的数据并通过 REST API 访问。

示例过滤函数

function main(stream) {
    const data = stream.data
    var numberDecimal = parseInt(data[0].number, 16);
    var filteredData = {
        hash: data[0].hash,
        number: numberDecimal
    };
    return filteredData;
}

此过滤器确保你仅接收区块编号和区块哈希，而不是整个区块数据集。你可以在 此处 找到其他多个过滤器示例（例如，ERC-20 转账、跟踪特定地址等）。

带函数的 Streams

虽然 Streams 提供实时和历史区块链数据，将其与 Functions 结合可以使你动态处理和增强这些数据。无论你是在创建 API 还是在区块链数据上执行复杂操作，Functions 都能增加另一层灵活性和自定义性，无需管理服务器。

以下是 Functions 与 Streams 的完美互补原因：

• API 准备就绪：Functions 会自动曝光为 API，使你可以轻松将其集成到前端或其他服务中。你可以将实时区块链数据转换为立即可用的 API。

• 无服务器灵活性：使用 Functions，你可以在无需担心管理服务器或扩展的情况下，在区块链数据上执行复杂操作。Functions 随着需求自我扩展，这意味着你只需支付所使用的部分。

• 外部库支持：Functions 使你可以在运行时环境中访问外部库，包括与 web3 相关的和实用工具相关的库。查看 此处 的完整支持库列表。

• 过滤和转换数据：Functions 允许你从 Streams 中过滤特定数据，如交易或事件，并将编码的区块链数据转换为适合你应用需求的格式。

• 增强数据：需要在将数据路由到目标之前增强数据吗？Functions 使你能够进行额外的 API 调用或检索补充数据，在将区块链数据传递之前提供额外的上下文。

• 关键值存储：访问 关键值存储，使你能根据 Functions 的操作存储和检索数据。

示例 Streams 和 Functions 用例

以下是 Streams 和 Functions 如何协同工作的示例。有关详细的 Functions 库，请单击 此处。

• Defi 或 NFT 平台的自定义 API：通过结合 Streams 和 Functions，你可以创建提供实时或聚合数据的自定义 API，例如 DeFi 平台的最新交易统计数据或 NFT 交易的实时更新。

• 自动通知：设置一个 Stream 来监控重要的区块链事件，并使用一个 Function 来实时触发通知，将其路由到如 PagerDuty、Discord 或 Google Sheets 等服务。这些用例在 Functions 库 中得到了很好的体现，在那里你会找到类似 Stream 到 Discord 或 Stream 到 Google Sheets 的示例。

• 数据分析：通过分析 Stream 事件实时处理区块链数据，将结果存储在关键值存储中，并在你需要进一步处理或通过 REST API 提供服务时检索它们。此工作流非常适合跟踪和提供指标及交易或构建自定义数据增强的 API 解决方案。

Streams 的工作原理

Streams 基于发布-订阅模型运作，但具有额外的可靠性和功能：

1. 数据摄取：Streams 不断从节点捕获区块链数据，包括块、交易、日志和跟踪。
2. 过滤和处理：应用自定义的 JavaScript 过滤器实时处理和转换数据，使你只提取所需的信息。
3. 交付机制：处理后的数据几乎实时或按可配置批次推送到你指定的目的地（例如，Webhook、兼容 S3 的存储桶或如 PostgreSQL 和 Snowflake 的数据库）。
4. 重组处理：Streams 自动检测并管理区块链重组，确保数据的一致性和准确性。

WebSocket 与 Streams

在本节中，我们将逐个主题比较 WebSocket 和 Streams，逐一细分它们的关键功能和使用场景，重点关注 WebSocket 的常见问题。

特征比较

性能考虑

使用场景

WebSocket 的最常见问题

尽管 WebSocket 是实时通信的强大工具，但它也存在一些挑战：

1. 处理断开连接

由于网络问题或服务器故障，WebSocket 可能会意外断开。重新连接并确保在停机期间没有数据丢失需要手动实现，例如重连逻辑和排队错过的事件。Streams 提供内置的可靠性，有可配置的超时设置（例如：30 秒阈值）、可自定义的重试次数及在指定重试次数后暂停机制。

建议

如果你需要处理这种断开连接，请查看 处理 WebSocket 断开连接 文章。

2. 带宽和数据管理

由于 WebSocket 通常交付超过所需的更多数据，可能会导致带宽使用过度。开发者必须在客户端过滤这些数据，这在高流量应用中特别低效。Streams 提供 服务器端过滤。

3. 同步实时与历史数据

WebSocket 通常仅提供实时数据。如果你需要将实时数据与历史数据同步（例如，在区块链浏览器中），那么为历史和实时数据管理单独的数据管道会增加你的架构复杂性。Streams 提供 实时和历史数据。

现在我们已经讨论了一些开发者在使用 WebSocket 时面临的常见挑战，接下来我们将快速探讨如何实现 WebSocket 和 Streams。

实现示例

下面的代码是代码示例，旨在为你快速概述每种设置的工作方式。请注意，这些并不是完整的、生产就绪的解决方案，而是帮助你快速启动 WebSocket 和 Streams 的起点。

确保将占位符 YOUR-QUICKNODE-WSS-URL 和 YOUR-QUICKNODE-API-KEY 替换为你自己的。

使用 Ethers.js 的 WebSocket

此示例演示如何使用 ethers.js 设置简单的 WebSocket 连接，以监听以太坊网络上的新块。你需要你的 QuickNode WebSocket 端点，可以从 QuickNode 仪表板 获取。

const { WebSocketProvider } = require('ethers')
const wsProvider = new WebSocketProvider('YOUR-QUICKNODE-WSS-URL')
wsProvider.on('block', blockNumber => {
    console.log('新块:', blockNumber)
})
// 处理断开连接
wsProvider.websocket.on('close', () => {
    console.log('WebSocket 断开连接')
    // 在此实现重连逻辑
})

此基本示例展示了如何实时监听新块。如果 WebSocket 连接断开，你需要实现自定义的重连逻辑，因为 WebSocket 不会自动处理这一点。

使用 API 创建一个 Stream

在本示例中，我们将使用 QuickNode Streams API 设置一个 Stream。该 Stream 监听以太坊主网并将数据发送到指定的 webhook。你需要配置你的 API 密钥、webhook URL 和用于流式传输的区块范围。

在下面的代码中，我们首先定义一个 main() 函数，这是我们的过滤函数，处理数据并将区块编号从十六进制转换为十进制格式。逻辑的核心围绕通过 QuickNode 的 API 创建一个 Stream，但在创建 Stream 之前，我们使用 testFilterFunction() 进行预检查。

testFilterFunction() 向 QuickNode API 发送过滤函数的 base64 编码版本（在这种情况下是 main() 函数），以验证其行为。它检查过滤器与示例区块数据的匹配，以确保它正常工作。

如果测试通过，setupQuickNodeStream() 函数将继续创建带有已验证过滤器的新 Stream。如果测试失败，将不会创建 Stream，并记录错误。

点击这里以展开代码示例

const axios = require("axios");

const QUICKNODE_API_KEY = "YOUR-QUICKNODE-API-KEY"; // 替换为你的实际 API 密钥
const WEBHOOK_URL = "YOUR-WEBHOOK-URL"; // 替换为你的 ngrok URL

function main(stream) {
  try {
    const data = stream.data
    const numberDecimal = parseInt(data[0].number, 16);
    const filteredData = {
      hash: data[0].hash,
      number: numberDecimal,
    };
    return filteredData;
  } catch (error) {
    return { error: error.message, stack: error.stack };
  }
}

async function testFilterFunction(base64FilterFunction) {
  const hash =
    "0xb72704063570e4b5a5f972f380fad5e43e1e8c9a1b0e36f204b9282c89adc677"; // 测试区块的哈希
  const number = "17811625"; // 测试区块的编号

  let data = JSON.stringify({
    network: "ethereum-mainnet",
    dataset: "block",
    filter_function: base64FilterFunction,
    block: number,
  });

  try {
    const response = await axios.post(
      "https://api.quicknode.com/streams/rest/v1/streams/test_filter",
      data,
      {
        headers: {
          accept: "application/json",
          "Content-Type": "application/json",
          "x-api-key": QUICKNODE_API_KEY,
        },
      }
    );

    if (
      response.status === 201 &&
      response.data.hash == hash &&
      response.data.number == number
    ) {
      console.log("过滤函数测试成功!");
      return true;
    } else {
      console.error("测试过滤函数时出错:", response.status);
      return false;
    }
  } catch (error) {
    console.error(
      "测试过滤函数时出错:",
      error.response ? error.response.data : error.message
    );
    throw error;
  }
}

async function setupQuickNodeStream(startSlot, endSlot) {
  const filterFunctionString = main.toString();
  const base64FilterFunction =
    Buffer.from(filterFunctionString).toString("base64");

  // 首先测试过滤函数
  const testResult = await testFilterFunction(base64FilterFunction);

  if (!testResult) {
    console.error("过滤函数失败。Stream 未创建。");
    return;
  }

  console.log("过滤函数通过。开始创建 Stream。");

  const streamConfig = {
    name: "通过 API 测试的 Streams",
    network: "ethereum-mainnet",
    dataset: "block",
    filter_function: base64FilterFunction,
    region: "usa_east",
    start_range: startSlot,
    end_range: endSlot,
    dataset_batch_size: 1,
    include_stream_metadata: "header",
    destination: "webhook",
    fix_block_reorgs: 0,
    keep_distance_from_tip: 0,
    destination_attributes: {
      url: WEBHOOK_URL,
      compression: "none",
      headers: {
        "Content-Type": "application/json",
      },
      max_retry: 3,
      retry_interval_sec: 1,
      post_timeout_sec: 10,
    },
    status: "active",
  };

  try {
    const response = await axios.post(
      "https://api.quicknode.com/streams/rest/v1/streams",
      streamConfig,
      {
        headers: {
          accept: "application/json",
          "Content-Type": "application/json",
          "x-api-key": QUICKNODE_API_KEY,
        },
      }
    );

    console.log("Stream 创建成功:", response.data);
    return response.data.id;
  } catch (error) {
    console.error(
      "创建 Stream 时出错:",
      error.response ? error.response.data : error.message
    );
    throw error;
  }
}

setupQuickNodeStream(1, 5).then((res) => console.log(res));

此示例展示了如何通过编程创建一个 Stream。你可以通过调整过滤器和目的地设置进一步自定义。

在前端创建一个 Stream

你也可以直接从 QuickNode 仪表板创建和管理 Streams，而无需编写代码。以下是在仪表板上设置 Stream 的方法：

1. 转到你的 QuickNode 仪表板。
2. 点击 创建 Stream 或选择可用的 Stream 模板之一。
3. 配置 Stream 的网络、数据集和过滤器。
4. 设置数据的目的地，例如 webhook 或 S3 桶。

这种方法允许你快速部署一个 Stream，而无需任何代码。

结论

尽管 WebSocket 在某些情况下可能适用于实时区块链数据，但其通常需要高额的维护劳动，尤其是在处理重连逻辑和断开连接时。另一方面，Streams 提供了一个更健壮、可扩展且功能丰富的替代方案，特别适合复杂或高流量的应用。

通过理解两者的优缺点，你可以对哪个解决方案最适合你的项目做出明智决定。无论你是选择 WebSocket 以简化开发，还是选择 Streams 以增强可靠性和可扩展性，QuickNode 都具备支持你所有需求的工具和基础设施。

如有任何问题、建议或想法，请直接 联系我。同时，通过关注我们的 Twitter 并加入我们的 Discord 和 Telegram 公告频道 来了解最新动态。

我们❤️反馈！

告知我们你是否有任何反馈或新的主题请求。我们期待你的来信。

进一步资源

• 文档：QuickNode Streams
• 技术指南：如何使用 Streams 构建区块链索引器
• 技术指南：如何回填以太坊 ERC-20 代币转移数据
• 技术指南：如何使用 TypeScript 创建对 Solana 区块链的 WebSocket 订阅
• 技术指南：如何管理与以太坊节点端点的 WebSocket 连接
• 其他与 Streams 相关的指南

• 原文链接： quicknode.com/guides/qui...
• 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～