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

Kafka 事件插件

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

📘

前置阅读

Kafka 插件负责将 TRON 链上产生的事件发送至 Apache Kafka 消息系统。相比于内置的 ZeroMQ 发布器，配合 Kafka 的消息中继机制具备三大核心优势：高可靠持久化（在被消费者确认前，数据安全保留在 Kafka 磁盘中）、数据可回播（消费者可以通过调整偏移量 Offset 重播历史事件）以及高并发扇出（支持多个独立的消费者按照各自的步调读取同一个事件流）。

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

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

推荐硬件（插件服务主机）

由于 Kafka 插件在全节点进程内部加载并运行，其计算开销包含在全节点主机的日常开销中。Kafka 代理服务可以与全节点部署在同一台服务器上，也可以部署在独立服务器上。生产环境建议使用独立服务器运行 Kafka，避免 Kafka 的磁盘、内存或 GC 压力间接影响全节点的出块共识。

资源项推荐硬件规格
CPU / 内存16 核 / 32 GB 以上
SSD 磁盘2.5 TB 以上
运行环境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-1.0.0.zip。

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

📘

版本说明

以下步骤以下载当前的稳定版 Kafka 3.x 为例。可从 Kafka 下载页面选择适合的版本。请勿使用过时文档中陈旧的 2.8.0 版本。自 3.x 起，其目录结构略有微调，并引入了不再依赖 ZooKeeper 的 KRaft 模式。

# 替换为选定的具体版本号
KAFKA_SCALA=2.13
KAFKA_VERSION=3.8.0

cd /usr/local
wget https://downloads.apache.org/kafka/${KAFKA_VERSION}/kafka_${KAFKA_SCALA}-${KAFKA_VERSION}.tgz
tar -xzf kafka_${KAFKA_SCALA}-${KAFKA_VERSION}.tgz
cd kafka_${KAFKA_SCALA}-${KAFKA_VERSION}

启动 ZooKeeper 服务与 Kafka 代理服务器（以下命令适合开发测试环境；生产环境建议使用进程守护管理器，如 systemd 或 supervisor）：

# 启动 ZooKeeper 注册中心（在前台新建终端或后台运行）
bin/zookeeper-server-start.sh config/zookeeper.properties &

# 启动 Kafka 代理服务 (Broker)
bin/kafka-server-start.sh config/server.properties &

若在生产环境运行，请记得配置合理的主题副本数（Replication Factor）与日志保留周期（Retention Policy），以确保高可用与磁盘空间的平衡。

3. 配置全节点

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

event.subscribe = {
  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-1.0.0.zip"   // 插件压缩包的绝对路径
  server = "127.0.0.1:9092"                  // Kafka 代理服务的连接地址
  dbconfig = ""                              // 仅用于 MongoDB 插件；Kafka 场景下保留为空
  contractParse = true
}

各配置项含义说明：

配置项键名核心含义
version指定事件订阅框架版本。1 代表 V2.0（支持历史区块回填）；0 代表 V1.0（默认值）。版本区别详见事件服务框架介绍。
startSyncBlockNum仅在 V2.0 框架下生效。设为 0 或负值时禁用历史回填；设为正数高度值时，节点会从该区块高度开始向前重播事件。在启用回填前请确保插件使用的是最新编译版本。
native.useNativeQueue必须配置为 false 以将数据分发给外部 Kafka 插件。（若设为 true 则意味着开启了内置的 ZeroMQ 发布通道。）
path指向本地 plugin-kafka-1.0.0.zip 文件的绝对路径。
serverKafka 代理服务（Broker）的连接地址，格式为 IP:Port。Kafka 的默认通信端口为 9092。
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 主题

在全节点启动前，必须在 Kafka 代理服务中手动创建 config.conf 的 topics 数组里配置的主题：

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

因为 Kafka 默认可能会拒绝写入不存在的主题，请为每个启用的事件主题重复执行该创建步骤。

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

事件订阅模块默认关闭。启动全节点时，必须在命令行附加 --es 参数以加载插件：

java -jar FullNode.jar -c config.conf --es

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

grep -i eventplugin logs/tron.log

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

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

如果加载失败，请检查全节点日志中的报错原因——这通常是由于 JAR 包路径配置有误、包文件无读写权限，或配置的 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 26 days ago