> ## Documentation Index
> Fetch the complete documentation index at: https://docs.blockdb.chaintable.com/llms.txt
> Use this file to discover all available pages before exploring further.

# 配合 BlockX 高并发计算

> 订阅上游区块表 → 用 BlockX 函数计算 → 写入目标 block 表，实时订阅与历史回填共用一份配置。

# 配合 BlockX 高并发计算

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

## 最典型的场景

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

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

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

## 用 Pipeline 串联整条链路

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

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

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

```python theme={null}
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`（提交任务）：

```python theme={null}
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()
```

<Info>
  完整 API 与配置项请参考 [BlockX 文档](https://www.blockx.example.com)。这里只介绍与 BlockDB 配合的核心路径。
</Info>

## 相关阅读

* [条件筛选与算子管道](/zh/guides/filter-operator)
* [多表区块高度对齐订阅](/zh/guides/multi-table-alignment)
* [实时订阅](/zh/guides/subscribe)


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.