如何使用链索引器框架创建区块链索引器

JarrodWatts
发布于 2023-12-06 14:18
阅读 39

本文介绍了如何为以太坊、Polygon等EVM兼容区块链构建索引器。

本指南将向你展示如何为与 EVM 兼容的区块链（如以太坊、Polygon Proof of Stake 等）构建你自己的索引器！

最后，你将拥有一个运行 [Kafka](https://kafka.apache.org/) 的 [Docker](https://www.docker.com/) 容器，该容器捕获你指定的所有区块链事件，以及有关如何在应用程序内部转换和使用此数据的参考！

让我们开始吧！

### 工作原理

链索引器框架的架构可以分为三个部分：

1. **区块生产者**：扫描区块链并将原始区块数据发布到 Kafka 中。

2. **转换器**：将来自 Kafka 的原始数据转换成有意义的事件。

3. **消费者**：读取转换后的数据，用于各种用例，例如来自前端应用程序或 API 端点。

我们将首先在 Docker 容器内设置我们的 Kafka 实例，然后为这三个组件中的每一个运行一个用 TypeScript 编写的脚本。

#### 架构图

生产者从指定的区块链读取数据，并将其发布到在 Docker 容器中运行的 Kafka 实例中。 同时，它还会在 Mongo DB 中保留最新的区块，以防 [reorgs](https://learnblockchain.cn/article/15431)。

一旦存储在 Kafka 中，我们就能够将其转换为更有意义的数据，并使其可供任何消费者读取；例如应用程序。

![](https://img.learnblockchain.cn/2025/08/17/85905703_image.webp)

### 设置后端基础设施

在我们创建三个服务（生产者、转换器、消费者）中的每一个之前，我们需要设置连接到区块链和存储数据所需的基础设施。 在本节中，我们将设置以下服务：

1. 用于连接到区块链的 RPC

2. 一个运行 Kafka 以存储原始数据的 Docker 容器

3. 用于存储最新区块的 Mongo 数据库。

首先，让我们创建一个新目录来包含我们将要做的所有工作：

复制

```lang-bash
## 创建我们的基本目录
mkdir chain-indexer-framework-guide

## 更改到这个新目录
cd chain-indexer-framework-guide

## 在 VS Code 中打开
code .

```

#### 创建 Docker Compose 文件

现在我们在这里，让我们通过创建一个 [Docker Compose 文件](https://docs.docker.com/compose/compose-file/03-compose-file/)来设置我们的 Docker 容器。

💡

你将需要安装 [Docker](https://docs.docker.com/get-docker/) 才能完成本指南的这一部分。 按照 [Get Docker](https://docs.docker.com/get-docker/) 页面上的说明进行操作，并通过在终端中运行“docker --version”来验证你的安装。

Docker Compose 文件允许我们定义应用程序的环境。 要创建一个，让我们在这个当前目录中创建一个 `docker-compose.yml` 文件。

💡

如果你使用的是 [VS Code](https://code.visualstudio.com/)，建议安装 [Docker Extension](https://code.visualstudio.com/docs/containers/overview)。

在 `docker-compose.yml` 中，设置以下 compose 文件：

复制

```lang-yaml
---
version: "3"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.0
    hostname: zookeeper
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

broker:
    image: confluentinc/cp-kafka:7.0.0
    container_name: broker
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://broker:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

```

我们不会在本指南中介绍所有字段（[了解有关 compose 文件的更多信息](https://docs.docker.com/compose/compose-file/)），但此文件将允许我们运行一个 [Docker 容器](https://www.docker.com/resources/what-container/)，其中包含用于维护配置信息的 [Apache ZooKeeper](https://zookeeper.apache.org/) 和用于存储原始数据的 [Apache Kafka](https://kafka.apache.org/)。

#### 运行 Docker 容器

现在我们已经定义了 compose 文件，让我们运行 Docker 容器。

使用 [Docker Compose](https://docs.docker.com/compose/) 的“[up](https://docs.docker.com/engine/reference/commandline/compose_up/)”命令，并使用 `-d` 标志在分离模式下运行容器。

复制

```lang-bash
docker-compose up -d

```

💡

**第一次使用 Docker**？ 确保你已[完成 Docker 安装](https://stackoverflow.com/questions/40459280/docker-cannot-start-on-windows#:~:text=First%2C%20verify%20that%20Docker%20Desktop%20application%20is%20running.%20If%20not%2C%20launch%20it%3A%20that%20will%20run%20the%20docker%20daemon%20(just%20wait%20few%20minutes).)。

运行 `docker ps` 命令以查看所有容器的状态：

![](https://img.learnblockchain.cn/2025/08/17/53513917_image.webp)

你也可以运行 [Docker Desktop](https://www.docker.com/products/docker-desktop/) 来查看正在运行的容器：

![](https://img.learnblockchain.cn/2025/08/17/71999953_image.webp)

现在我们的 Kafka 实例已经启动并运行，我们准备好存储来自区块链的原始数据。 但是，要读取区块链数据，我们需要一种连接到链本身的方法；所以接下来让我们创建一个新的 RPC 节点。

#### 创建 RPC 节点

有 [许多解决方案提供商提供与区块链的 RPC 连接](https://ecosystem.polygon.technology/spn/explore/?search=&competency=RPC+Provider&chain=)。 在本指南中，我们将使用流行的 RPC 提供商 [Alchemy](https://docs.alchemy.com/) 作为我们的 RPC 提供商，因为它们提供了一个慷慨的免费层。

首先，转到你的 Alchemy 仪表板的 [应用程序](https://dashboard.alchemy.com/apps) 页面，然后单击**创建新应用**：

![](https://img.learnblockchain.cn/2025/08/17/29617790_image.webp)

选择你要从中读取数据的链和网络，然后单击**创建应用**：

![](https://img.learnblockchain.cn/2025/08/17/16177756_image.webp)

稍后当我们编写生产者时，我们将返回此处获取我们的 RPC URL。

#### 创建 Mongo DB

接下来，让我们设置一个数据库来执行 TODO。

在本指南中，我们将使用 [Mongo DB](https://www.mongodb.com/)，因为链索引器框架的 **Producer** 具有一个我们可以轻松将此数据库插入的 `mongoUrl` 属性。

如果你尚未这样做，请[**注册 Mongo DB**](https://www.mongodb.com/)，然后部署一个新数据库。

以下是我选择的选项；但你可以根据自己的喜好进行自定义：

![](https://img.learnblockchain.cn/2025/08/17/64459180_image.webp)

配置好数据库后，完成设置过程。

首先，添加一种身份验证方法；在本指南中，我们使用一个简单的用户名和密码，该用户名和密码是为我们自动生成的。

![](https://img.learnblockchain.cn/2025/08/17/68242726_image.webp)

其次，配置谁可以访问你的数据库。 在本指南中，我们只是添加我们当前机器的 IP 地址。 最后，单击**完成并关闭**。

![](https://img.learnblockchain.cn/2025/08/17/19807053_image.webp)

最后，基础设施设置完成！ 🎉

### 设置 Node.js + TypeScript

现在我们准备好设置链索引器框架的三个组件。

通过运行以下命令来设置一个简单的 TypeScript + Node.js 项目：

复制

```lang-bash
npm init -y

```

要添加 TypeScript，请创建一个包含以下内容的 `tsconfig.json` 文件：

复制

```lang-json
{
  "compilerOptions": {
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "esModuleInterop": true,
    "target": "ES2020",
    "sourceMap": true,
    "strict": true,
    "outDir": "dist",
    "declaration": true
  },
  "include": ["src/**/*"]
}

```

将 `package.json` 文件设置为以下内容：

复制

```lang-bash
{
  "name": "producer",
  "version": "1.0.0",
  "type": "module",
  "main": "index.js",
  "scripts": {
    "build": "tsc",
    "start-producer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/producer.js",
    "start-transformer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/transformer.js",
    "start-consumer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/consumer.js"
  },
}

```

安装所需的 TypeScript 依赖项：

复制

```lang-bash
npm install -D typescript @types/node

```

在安装更多依赖项之前，你需要按照 [`node-gyp`](https://github.com/nodejs/node-gyp) 的安装说明进行操作，作为安装 [`@maticnetwork/chain-indexer-framework`](https://github.com/0xPolygon/chain-indexer-framework) 包的先决条件。 如果你使用的是 Windows，这包括在 Visual Studio 中安装当前版本的 Python 和 Visual C++ Build Environment。

接下来，通过运行以下命令安装 [`@maticnetwork/chain-indexer-framework`](https://github.com/0xPolygon/chain-indexer-framework) 包：

复制

```lang-bash
npm install @maticnetwork/chain-indexer-framework

```

最后，创建一个 `src` 文件夹，其中包含三个文件：

- `producer.ts`

- `transformer.ts`

- `consumer.ts`

现在我们准备开始了。 让我们从生产者开始。

### 创建生产者

在 `producer.ts` 文件中，我们将连接到你在 RPC 步骤中选择的区块链，并将原始数据写入 Kafka。

以下是在 Kafka 中开始存储原始数据的基本代码。 Mongo DB 用于在 Mongo DB 成为 Kafka 中的永久数据之前存储最新数量的区块（在 `maxReOrgDepth` 字段中指定）。

复制

```lang-typescript
import { BlockPollerProducer } from "@maticnetwork/chain-indexer-framework/block_producers/block_polling_producer";
import { produce } from "@maticnetwork/chain-indexer-framework/kafka/producer/produce";

const producer = produce<BlockPollerProducer>({
  startBlock: 50689834, // 选择你想要的任何起始区块
  rpcWsEndpoints: ["<your-alchemy-rpc-url-here>",],
  blockPollingTimeout: 20000,
  topic: "polygon.1442.blocks",
  maxReOrgDepth: 256, // Polygon 上的最大 reorg 深度为 256
  maxRetries: 5,
  mongoUrl: "mongodb+srv://<your-mongo-username>:<your-mongo-password>@chain-indexer.0ymaemb.mongodb.net/",
  "bootstrap.servers": "localhost:9092",
  "security.protocol": "plaintext",
  type: "blocks:polling",
});

producer.on("blockProducer.fatalError", (error: any) => {
  process.exit(1); //发生致命错误时退出进程。 进程管理器需要重新启动进程。
});

```

重要的是，你需要：

- 在 `rpcsWsEndpoints` 字段中添加你的 HTTPS Alchemy RPC URL。

- 将你的 Mongo URL 连接字符串添加为 `mongoUrl`。

- （可选）根据你的偏好配置其他属性。

准备好后，你可以运行以下命令来启动生产者：

复制

```lang-typescript
npm run start-producer

```

如果一切正常，你应该开始在 Mongo DB 中看到一些数据，你可以在 **Database > Collections** 部分中找到：

![](https://img.learnblockchain.cn/2025/08/17/35069598_image.webp)

#### 添加日志

链索引器框架还具有日志记录功能，可以查看幕后发生的事情。 要将其添加到你的生产者，请首先导入 Logger：

复制

```lang-typescript
import { Logger } from "@maticnetwork/chain-indexer-framework/logger";

```

并且可以选择将日志添加到生产者发出的任何事件，例如：

复制

```lang-typescript
producer.on("blockProducer.fatalError", (error: any) => {
  Logger.error(`Block producer exited. ${error.message}`);
  process.exit(1); //发生致命错误时退出进程。 进程管理器需要重新启动进程。
});

```

### 转换和使用数据

现在我们已经在 Kafka 实例中提供了原始数据！ 你已准备好开始在应用程序中转换和使用数据。

尽管开始这些步骤需要大量的代码，但我们已经为所有内容奠定了基础，因此它应该相对简单！

与其转储所有代码供你复制和粘贴，不如我们提供了一些开源 [示例](https://github.com/0xPolygon/chain-indexer-framework/tree/main/examples)，说明如何在下面链接的 GitHub 上设置不同类型的转换器和生产者。

- [MATIC 转账示例](https://github.com/0xPolygon/chain-indexer-framework/tree/main/examples/matic_transfer)

- [NFT 余额示例](https://github.com/0xPolygon/chain-indexer-framework/tree/main/examples/nft_balancer)

>- 原文链接： [blog.jarrodwatts.com/cre...](https://blog.jarrodwatts.com/create-a-blockchain-indexer-with-chain-indexer-framework)
>- 登链社区 AI 助手，为大家转译优秀英文文章，如有翻译不通的地方，还请包涵～

本指南将向你展示如何为与 EVM 兼容的区块链（如以太坊、Polygon Proof of Stake 等）构建你自己的索引器！

最后，你将拥有一个运行 Kafka 的 Docker 容器，该容器捕获你指定的所有区块链事件，以及有关如何在应用程序内部转换和使用此数据的参考！

让我们开始吧！

工作原理

链索引器框架的架构可以分为三个部分：

区块生产者：扫描区块链并将原始区块数据发布到 Kafka 中。
转换器：将来自 Kafka 的原始数据转换成有意义的事件。
消费者：读取转换后的数据，用于各种用例，例如来自前端应用程序或 API 端点。

我们将首先在 Docker 容器内设置我们的 Kafka 实例，然后为这三个组件中的每一个运行一个用 TypeScript 编写的脚本。

架构图

生产者从指定的区块链读取数据，并将其发布到在 Docker 容器中运行的 Kafka 实例中。同时，它还会在 Mongo DB 中保留最新的区块，以防 reorgs。

一旦存储在 Kafka 中，我们就能够将其转换为更有意义的数据，并使其可供任何消费者读取；例如应用程序。

设置后端基础设施

用于连接到区块链的 RPC
一个运行 Kafka 以存储原始数据的 Docker 容器
用于存储最新区块的 Mongo 数据库。

首先，让我们创建一个新目录来包含我们将要做的所有工作：

复制

## 创建我们的基本目录
mkdir chain-indexer-framework-guide

## 更改到这个新目录
cd chain-indexer-framework-guide

## 在 VS Code 中打开
code .

创建 Docker Compose 文件

现在我们在这里，让我们通过创建一个 Docker Compose 文件来设置我们的 Docker 容器。

💡

你将需要安装 Docker 才能完成本指南的这一部分。按照 Get Docker 页面上的说明进行操作，并通过在终端中运行“docker --version”来验证你的安装。

Docker Compose 文件允许我们定义应用程序的环境。要创建一个，让我们在这个当前目录中创建一个 docker-compose.yml 文件。

💡

如果你使用的是 VS Code，建议安装 Docker Extension。

在 docker-compose.yml 中，设置以下 compose 文件：

复制

---
version: "3"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.0.0
    hostname: zookeeper
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-kafka:7.0.0
    container_name: broker
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_INTERNAL:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,PLAINTEXT_INTERNAL://broker:29092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1

我们不会在本指南中介绍所有字段（了解有关 compose 文件的更多信息），但此文件将允许我们运行一个 Docker 容器，其中包含用于维护配置信息的 Apache ZooKeeper 和用于存储原始数据的 Apache Kafka。

运行 Docker 容器

现在我们已经定义了 compose 文件，让我们运行 Docker 容器。

使用 Docker Compose 的“up”命令，并使用 -d 标志在分离模式下运行容器。

复制

docker-compose up -d

💡

第一次使用 Docker？确保你已完成 Docker 安装。

运行 docker ps 命令以查看所有容器的状态：

你也可以运行 Docker Desktop 来查看正在运行的容器：

现在我们的 Kafka 实例已经启动并运行，我们准备好存储来自区块链的原始数据。但是，要读取区块链数据，我们需要一种连接到链本身的方法；所以接下来让我们创建一个新的 RPC 节点。

创建 RPC 节点

有许多解决方案提供商提供与区块链的 RPC 连接。在本指南中，我们将使用流行的 RPC 提供商 Alchemy 作为我们的 RPC 提供商，因为它们提供了一个慷慨的免费层。

首先，转到你的 Alchemy 仪表板的应用程序页面，然后单击创建新应用：

选择你要从中读取数据的链和网络，然后单击创建应用：

稍后当我们编写生产者时，我们将返回此处获取我们的 RPC URL。

创建 Mongo DB

接下来，让我们设置一个数据库来执行 TODO。

在本指南中，我们将使用 Mongo DB，因为链索引器框架的 Producer 具有一个我们可以轻松将此数据库插入的 mongoUrl 属性。

如果你尚未这样做，请注册 Mongo DB，然后部署一个新数据库。

以下是我选择的选项；但你可以根据自己的喜好进行自定义：

配置好数据库后，完成设置过程。

首先，添加一种身份验证方法；在本指南中，我们使用一个简单的用户名和密码，该用户名和密码是为我们自动生成的。

其次，配置谁可以访问你的数据库。在本指南中，我们只是添加我们当前机器的 IP 地址。最后，单击完成并关闭。

最后，基础设施设置完成！ 🎉

设置 Node.js + TypeScript

现在我们准备好设置链索引器框架的三个组件。

通过运行以下命令来设置一个简单的 TypeScript + Node.js 项目：

复制

npm init -y

要添加 TypeScript，请创建一个包含以下内容的 tsconfig.json 文件：

复制

{
  "compilerOptions": {
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "esModuleInterop": true,
    "target": "ES2020",
    "sourceMap": true,
    "strict": true,
    "outDir": "dist",
    "declaration": true
  },
  "include": ["src/**/*"]
}

将 package.json 文件设置为以下内容：

复制

{
  "name": "producer",
  "version": "1.0.0",
  "type": "module",
  "main": "index.js",
  "scripts": {
    "build": "tsc",
    "start-producer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/producer.js",
    "start-transformer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/transformer.js",
    "start-consumer": "npm run build && node --experimental-import-meta-resolve --trace-warnings dist/consumer.js"
  },
}

安装所需的 TypeScript 依赖项：

复制

npm install -D typescript @types/node

在安装更多依赖项之前，你需要按照 node-gyp 的安装说明进行操作，作为安装 @maticnetwork/chain-indexer-framework 包的先决条件。如果你使用的是 Windows，这包括在 Visual Studio 中安装当前版本的 Python 和 Visual C++ Build Environment。

接下来，通过运行以下命令安装 @maticnetwork/chain-indexer-framework 包：

复制

npm install @maticnetwork/chain-indexer-framework

最后，创建一个 src 文件夹，其中包含三个文件：

producer.ts
transformer.ts
consumer.ts

现在我们准备开始了。让我们从生产者开始。

创建生产者

在 producer.ts 文件中，我们将连接到你在 RPC 步骤中选择的区块链，并将原始数据写入 Kafka。

以下是在 Kafka 中开始存储原始数据的基本代码。 Mongo DB 用于在 Mongo DB 成为 Kafka 中的永久数据之前存储最新数量的区块（在 maxReOrgDepth 字段中指定）。

复制

import { BlockPollerProducer } from "@maticnetwork/chain-indexer-framework/block_producers/block_polling_producer";
import { produce } from "@maticnetwork/chain-indexer-framework/kafka/producer/produce";

const producer = produce&lt;BlockPollerProducer>({
  startBlock: 50689834, // 选择你想要的任何起始区块
  rpcWsEndpoints: ["&lt;your-alchemy-rpc-url-here>",],
  blockPollingTimeout: 20000,
  topic: "polygon.1442.blocks",
  maxReOrgDepth: 256, // Polygon 上的最大 reorg 深度为 256
  maxRetries: 5,
  mongoUrl: "mongodb+srv://&lt;your-mongo-username>:&lt;your-mongo-password>@chain-indexer.0ymaemb.mongodb.net/",
  "bootstrap.servers": "localhost:9092",
  "security.protocol": "plaintext",
  type: "blocks:polling",
});

producer.on("blockProducer.fatalError", (error: any) => {
  process.exit(1); //发生致命错误时退出进程。 进程管理器需要重新启动进程。
});

重要的是，你需要：

在 rpcsWsEndpoints 字段中添加你的 HTTPS Alchemy RPC URL。
将你的 Mongo URL 连接字符串添加为 mongoUrl。
（可选）根据你的偏好配置其他属性。

准备好后，你可以运行以下命令来启动生产者：

复制

npm run start-producer

如果一切正常，你应该开始在 Mongo DB 中看到一些数据，你可以在 Database > Collections 部分中找到：

添加日志

链索引器框架还具有日志记录功能，可以查看幕后发生的事情。要将其添加到你的生产者，请首先导入 Logger：

复制

import { Logger } from "@maticnetwork/chain-indexer-framework/logger";

并且可以选择将日志添加到生产者发出的任何事件，例如：

复制

producer.on("blockProducer.fatalError", (error: any) => {
  Logger.error(`Block producer exited. ${error.message}`);
  process.exit(1); //发生致命错误时退出进程。 进程管理器需要重新启动进程。
});

转换和使用数据

现在我们已经在 Kafka 实例中提供了原始数据！你已准备好开始在应用程序中转换和使用数据。