指南API 参考手册社区Discord博客FAQ漏洞赏金公告中心
指南
English(英文版)

Kafka 事件插件

通过 Kafka 中继插件订阅 TRON 链上事件——支持历史回放的持久流。最适合数据分析流水线、多端业务订阅及高吞吐的归档服务。

📘

前置阅读

Kafka 插件负责将 TRON 链上产生的事件发送至 Apache Kafka 消息系统。相比于内置的 ZeroMQ 发布器，Kafka 提供数据持久化（事件写入后按 Topic 的保留策略存储，不取决于消费者是否确认）、数据可回播（消费者可以通过调整偏移量 Offset 重播历史事件）以及高并发扇出（支持多个独立的消费者按照各自的步调读取同一个事件流）。

因此，如果业务涉及数据索引服务、数据分析流水线，或后端需要在宕机后恢复并从已知检查点（Checkpoint）重新拉取数据，Kafka 是更合适的选择。

本文说明如何编译插件、运行 Kafka 代理服务（Broker）、配置全节点，以及验证整套端到端的消息推送链路。

推荐硬件（全节点与插件主机）

由于 Kafka 插件在全节点进程内部加载并运行，其 CPU 和内存开销属于全节点总工作负载的一部分。Kafka Broker 可以与全节点部署在同一台服务器上，也可以部署在独立服务器上。生产环境建议使用独立服务器运行 Kafka，避免 Kafka 的磁盘、内存或 GC 压力间接影响全节点的出块共识。

资源项推荐硬件规格
CPU / 内存16 核 / 32 GB 以上
SSD 磁盘全节点数据至少 3 TB；Kafka 容量按事件量、保留周期和副本数另行规划
运行环境Linux / macOS

1. 编译 Kafka 事件插件

首先，从 GitHub 克隆插件项目的源码并使用 Gradle 进行打包：

git clone https://github.com/tronprotocol/event-plugin.git
cd event-plugin
./gradlew build

编译成功后，当前版本生成的插件压缩包路径为：event-plugin/build/plugins/plugin-kafka-3.0.0.zip。插件版本升级后文件名可能变化，请以 build/plugins 目录中的实际文件名为准，并在后续配置中使用同一文件名。

2. 部署 Kafka 代理服务 (Broker)

📘

版本说明

以下步骤以 Kafka 4.3.1 的 KRaft 单节点模式为例。Kafka Broker 需要 Java 17 或更高版本，该要求仅适用于 Kafka 进程；java-tron 的 JDK 选择请参阅部署节点。

# 本示例使用 Kafka 4.3.1
KAFKA_SCALA=2.13
KAFKA_VERSION=4.3.1
KAFKA_BASE_URL=https://downloads.apache.org/kafka/${KAFKA_VERSION}

cd /usr/local
wget "${KAFKA_BASE_URL}/kafka_${KAFKA_SCALA}-${KAFKA_VERSION}.tgz"
wget "${KAFKA_BASE_URL}/kafka_${KAFKA_SCALA}-${KAFKA_VERSION}.tgz.sha512"
⚠️

下载校验

解压前，从相同的 Apache Kafka 下载目录获取与二进制包对应的 .sha512 或 .asc 文件，并按 Apache 下载验证指南校验 SHA-512 或 OpenPGP 签名。校验失败时不得解压或启动该二进制包。

校验通过后再解压并进入 Kafka 目录：

tar -xzf "kafka_${KAFKA_SCALA}-${KAFKA_VERSION}.tgz"
cd "kafka_${KAFKA_SCALA}-${KAFKA_VERSION}"

初始化 KRaft 存储目录并启动 Kafka Broker。以下单节点命令仅适合开发和测试环境：

# 生成集群 ID 并格式化存储目录
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"
bin/kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties

# 启动 Kafka Broker
bin/kafka-server-start.sh config/server.properties &

生产环境应部署多 Broker KRaft 集群，使用进程管理器管理 Kafka 进程，并根据可用性和存储需求配置主题副本数与日志保留周期。

3. 配置全节点

在全节点的 config.conf 配置文件中添加 event.subscribe 配置块：

event.subscribe = {
  enable = true         // 启用事件订阅
  version = 1            // 1 表示启用 V2.0 事件框架（支持历史回填）；0 表示启用 V1.0 框架（默认值）
  startSyncBlockNum = 0  // 仅在 V2.0 框架下有效——具体配置参见下方说明

  native = {
    useNativeQueue = false   // 必须为 false 以便将事件中继路由至外部插件
    bindport = 5555
    sendqueuelength = 1000
  }

  path = "/path/to/plugin-kafka-3.0.0.zip"   // 插件压缩包的绝对路径
  server = "127.0.0.1:9092"                  // Kafka 代理服务的连接地址
  dbconfig = ""                              // 仅用于 MongoDB 插件；Kafka 场景下保留为空
  contractParse = true
}

各配置项含义说明：

配置项键名核心含义
enable设为 true 以启用事件订阅。
version指定事件订阅框架版本。1 代表 V2.0（支持历史区块回填）；0 代表 V1.0（默认值）。版本区别详见事件服务框架介绍。
startSyncBlockNum仅在 V2.0 框架下生效。设为 0 或负值时禁用历史回填；设为正数高度值时，节点会从该区块高度开始向前重播事件。在启用回填前请确保插件使用的是最新编译版本。
native.useNativeQueue必须配置为 false 以将数据分发给外部 Kafka 插件。（若设为 true 则意味着开启了内置的 ZeroMQ 发布通道。）
path指向本地 plugin-kafka-3.0.0.zip 文件的绝对路径。
serverKafka Broker 的连接地址，格式为 IP:Port。Kafka 的默认通信端口为 9092。Broker 与全节点分开部署时，应填写全节点可以访问的 Broker 地址，并正确配置 Kafka 的 listeners 和 advertised.listeners。
dbconfig仅供 MongoDB 插件进行账号鉴权时使用；在 Kafka 模式下必须留空。
contractParse设为 true 时，智能合约事件在被发送至消息队列前，会由全节点先根据合约的 ABI 自动完成解码。

选择订阅的事件类型

全节点支持分发七种不同的触发器类型——包括区块、交易、合约事件、原始日志，以及对应的固化区块版本。触发器的详细载荷与过滤选择，请参阅事件类型指南。请不要订阅与业务无关的触发器类型，以免为全节点造成无谓的 CPU 与内存负荷。

在配置文件的 topics 数组中添加对应的订阅项。其中 topic 字段为中继后存入 Kafka 的主题名称（该值需与 Kafka 中预先创建的主题一致）：

topics = [
  {
    triggerName = "block"   // 节点底层定义的触发器类型，不可修改
    enable = true           // 设为 false 表示禁用该条订阅
    topic = "block"         // 投递至 Kafka 的目标主题名称，支持自定义
  }
]

配置过滤器 (可选)

可以通过配置 filter 块，按特定的区块高度区间、智能合约地址或事件哈希来缩小过滤流范围（注意：该过滤仅适用于合约事件与原始日志，区块与交易流无法被过滤）：

filter = {
  fromblock = ""        // "", "earliest", 或特定的区块高度
  toblock = ""          // "", "latest", 或特定的区块高度
  contractAddress = [
    ""                  // 合约账户地址；保留为空时匹配所有合约
  ]
  contractTopic = [
    ""                  // 事件 Topic 哈希；保留为空时匹配所有合约事件
  ]
}

4. 创建 Kafka 主题

建议在启动全节点前，为 config.conf 的 topics 数组中每个启用的事件类型预先创建对应主题：

bin/kafka-topics.sh --create --topic block --bootstrap-server localhost:9092

这样可以避免事件投递行为依赖 Broker 的自动创建主题配置。请为每个启用的事件主题重复执行该创建步骤。

5. 启动全节点并进行验证

事件订阅模块默认关闭。请在 config.conf 中设置 event.subscribe.enable = true，然后启动全节点：

java -jar FullNode.jar -c config.conf

请通过检查日志来确认插件是否已加载：

grep -i eventplugin logs/tron.log

如果加载成功，会看到类似如下的日志行：

[o.t.c.l.EventPluginLoader] '/path/to/plugin-kafka-3.0.0.zip' loaded

如果加载失败，请检查全节点日志中的报错原因——这通常是由于插件包路径配置有误、包文件无读写权限，或配置的 Kafka 连接地址不可达导致的。

6. 使用客户端消费事件

可以使用 Kafka 自带的命令行控制台消费端（或任何支持 Kafka 消费的 SDK 客户端）连接并读取消息：

bin/kafka-console-consumer.sh --topic block --from-beginning --bootstrap-server localhost:9092

接收到的事件为标准的 JSON 载荷格式，样例如下：

{
  "timeStamp": 1539973125000,
  "triggerName": "blockTrigger",
  "blockNumber": 3341315,
  "blockHash": "000000000032fc03440362c3d42eb05e79e8a1aef77fe31c7879d23a750f2a31",
  "transactionSize": 16,
  "latestSolidifiedBlockNumber": 3341297,
  "transactionList": [
    "8757f846e541b51b5692a2370327f4b8031125f4557f8ad4b1037d4452616d39",
    "f6adab7814b34e5e756170f93a31a0c3393c5d99eff11e30271916375adc7467",
    "89bcbcd063a48ef4a5678a033acf5edbb6b17419a3c91eb0479a3c8598774b43"
  ]
}

事件进入 Kafka 消息通道后，应用后端可以使用任何标准 Kafka 消费 SDK（如 Python 的 confluent-kafka-python、Go 的 sarama、Java 的 Kafka client 等）构建高可靠的事件消费管道。

相关资源

Updated 8 days ago