配合 BlockX 高并发计算

BlockDB 管存取，BlockX 管计算。当上游单块数据量大、出块频率高时，单进程顺序处理追不上出块节奏（例如 BSC 链需在几秒内处理完一个区块），此时用 BlockX 把每个区块的行全并发执行，让业务逻辑始终与出块频率同步。

最典型的场景

大多数链上数据加工都是同一条链路：

订阅上游区块表（如 chain.trace.eth）
        │
        ▼
对每个区块的每一行执行你的函数
        │
        ▼
把函数返回的结果写入你自己的目标 block 表

上游最典型的是 chain.trace.eth 这类区块 trace 表——一个区块里有成百上千条 trace，天然适合全并发计算。而这条链路实时订阅和历史区块回填用的是同一份配置、同一个函数。

用 Pipeline 串联整条链路

BlockX 的上层封装 Pipeline 把「源表 → 函数 → 目标表」包装成一个对象，只暴露两个动词：

backfill(block_start, block_end)：历史区块的批量回填。
update()：先 catch-up 把历史缺口补齐，追平后自动转入实时订阅。

两者复用同一份 triggers、同一个函数、同一个 target_table。

from blockx import Pipeline, Trigger, SOURCE_ROW

# 1) 写一个无状态函数：源表一行 dict 进来，返回 dict 就写入目标表，返回 None 就丢弃
#    （形参顺序和 params 里的位置一一对应，源表行用 SOURCE_ROW 占位）
def trace_to_token_transfer(chain_id, record):
    value = float(record.get("value") or 0)
    if value <= 0 or not record.get("tx_id"):
        return None              # 不合格的行直接过滤掉
    return {
        "id":        record["id"],
        "from_addr": record["from_addr"],
        "to_addr":   record["to_addr"],
        "value":     value,
        "tx_id":     record["tx_id"],
    }

# 2) 组装 Pipeline：订阅 chain.trace.eth，写进 token.token_transfer.eth
pipe = Pipeline(
    triggers=[
        Trigger(
            table="chain.trace.eth",
            func=trace_to_token_transfer,
            params=["eth", SOURCE_ROW],   # chain_id="eth"，第二个参数是源表行
        ),
    ],
    target_table="token.token_transfer.eth",
)

# 3) 历史回填：从 18000000 起批量回填到上游当前共识高度
backfill_result = pipe.backfill(block_start=18000000)

# 4) 实时：先补缺口、追平后转订阅，持续消费新区块
pipe.update()

实时与回填共用一份配置

backfill 和 update 走的是同一套配置：

	`backfill`	`update`
用途	历史高度的批量回填	实时消费 + 自动补历史缺口
数据来源	高度区间 `[block_start, block_end]`	订阅流 + catch-up 补齐
函数 / Trigger / 目标表	相同	相同
结束条件	跑完区间返回汇总	常驻，直到被 stop

因此你只需维护一份函数和 Pipeline 配置，历史要回填多少、实时要追到哪，都由同一个对象统一处理，天然不会出现”回填和实时逻辑不一致”的问题。

底层接口（进阶）

Pipeline 是上层封装。若你需要更细粒度控制，可绕开它，直接组合底层的 BlockTableCallConfig（算什么）、BlockWriteHandler（结果写哪）、TaskBuilder（提交任务）：

from blockx import BlockTableCallConfig, BlockWriteHandler, TaskBuilder
from blockdb import Block

block = Block(id="0x...", height=18000000, timestamp="2024-06-08T00:00:00Z")

config = BlockTableCallConfig(block=block, triggerSources=[
    {"table": "chain.trace.eth", "func": trace_to_token_transfer},
])
handler = BlockWriteHandler(targetTable="token.token_transfer.eth", block=block)

task = TaskBuilder.build(call_config=config, handler=handler)
result = task.submit()

完整 API 与配置项请参考 BlockX 文档。这里只介绍与 BlockDB 配合的核心路径。

开始使用

核心概念

架构介绍

使用指南

参考

配合 BlockX 高并发计算

配合 BlockX 高并发计算

最典型的场景

用 Pipeline 串联整条链路

实时与回填共用一份配置

底层接口（进阶）

相关阅读

​配合 BlockX 高并发计算

​最典型的场景

​用 Pipeline 串联整条链路

​实时与回填共用一份配置

​底层接口（进阶）

​相关阅读

配合 BlockX 高并发计算

最典型的场景

用 Pipeline 串联整条链路

实时与回填共用一份配置

底层接口（进阶）

相关阅读