事件监听 - 合约事件监听的方案汇总

合约事件监听的方案

合约事件监听是区块链开发中常见的需求，主要应用在 Dapp 开发、钱包开发、交易监控、交易所开发等场景。在这里，我们介绍了四种场景的合约事件监听方案，并通过使用 go 语言实操，搭建了一套监听合约事件的最小代码。

wbsocket（eth_subscribe）

这是一种实时推送的方式，采用 websocket 连接到以太坊 RPC 节点。然后通过调用 eth_subscribe 的方式建立双向流进行数据传输。具体流程如下：

1. 建立 wesocket 连接
2. 使用合约地址构建过滤参数
3. 调用 eth_subscribe 调用这个 JSON-RPC接口，传入参数，结果的 channel，事件结果会被推送到 channel 中。
4. 通过匹配 ABI 中的事件签名即可获取到事件

websocket 的方式适用于实时获取合约事件的场景下使用，若系统故障，有丢失事件的风险。

The Graph

Graph Node 自动从区块链读取合约事件（logs），并将其解析、处理为结构化数据（entities）的一种机制。 我们通过编写 subgraph 程序，部署到 Graph Node 即可为我们以指定逻辑解析、存储事件。 在 The Graph 中，我们可以 GraphQL 进行自定义查询。Graph Node 的底层是通过eth_getLogs 接口来进行时间获取的。 The Graph 在 Dapp 开发中较为常用，支持自动索引+存储，提供 GraphQL 查询接口，开箱即用，十分方便。

eth_getTransactionReceipt

在扫链监听事件中，可以采用 eth_getTransactionReceipt 的 RPC 接口来进行获取当前交易 hash 的合约事件。 在eth_getTransactionReceipt 这个接口中，会返回一个 logs 的字段，我们针对这个字段进行解析即可获取相应事件。 具体解析流程如下：

1. HTTP 连接到以太坊 RPC 节点
2. 根据 eth_getBlockNumber() 获取到最新区块或指定区块
3. 解析区块里面的交易，通过 from 地址和 to 地址，筛选出相关的交易 hash。
4. 根据交易 hash，调用 eth_getTransactionReceipt 接口，解析 log字段里面的数据(可通过合约 ABI 中的事件签名和事件 topic[0] 进行匹配)。

   eth_getLogs

   在扫链监听事件中，可以采用 eth_getLogs 接口来指定区块范围，传入合约地址来获取事件。eth_getLogs 是以太坊 JSON-RPC 中用于获取区块范围内日志（事件）的接口，是最常用的获取事件方式。

具体操作步骤如下：

1. HTTP 连接到以太坊 RPC 节点
2. 构造一个区块号范围，传入一个合约地址，并传入事件签名，调用 eth_getLogs 即可获取到该范围内、该合约地址的某个事件的日志（相当于提前帮我们筛选出来事件了）。

go 语言监听合约事件实操

github仓库：https://github.com/Shawn-Shaw-x/eth-event-watch.git

eth_getTransactionReceipt

func TestGetTransactionReceipt(t *testing.T) {
    // 加载 .env 文件
    err := godotenv.Load()
    if err != nil {
        log.Fatalf("Error loading .env file")
    }

    // 获取环境变量
    rpcURL := os.Getenv("ETHEREUM_RPC_URL")
    apiKey := os.Getenv("ETHEREUM_API_KEY")
    contractAddress := os.Getenv("CONTRACT_ADDRESS")
    eventSignature := os.Getenv("EVENT_SIGNATURE")

    if rpcURL == "" || apiKey == "" || contractAddress == "" {
        log.Fatal(" ❌ Missing required environment variables")
    }

    // 连接以太坊节点（使用你的 RPC）
    client, err := ethclient.Dial(rpcURL + apiKey)
    if err != nil {
        log.Fatalf("❌ 连接失败: %v", err)
    }
    defer client.Close()

    // 获取最新区块号
    latestBlockNum, err := client.BlockNumber(context.Background())
    if err != nil {
        log.Fatalf("❌ 获取最新区块号失败: %v", err)
    }
    // 根据区块号获取区块
    latestBlock, err := client.BlockByNumber(context.Background(), big.NewInt(int64(latestBlockNum)))
    if err != nil {
        log.Fatalf("❌ 获取最新区块内容失败: %v", err)
    }

    // todo 筛选交易
    // 替换为你想要监听的交易哈希
    txHash := latestBlock.Transactions()[10].Hash()

    // 获取交易回执（包含事件 logs）
    receipt, err := client.TransactionReceipt(context.Background(), txHash)
    if err != nil {
        log.Fatalf("❌ 获取回执失败: %v", err)
    }

    // 遍历日志
    for _, vLog := range receipt.Logs {
        if (vLog.Topics[0].Hex()) != eventSignature {
            // 跳过非 Transfer 事件
            continue
        }
        fmt.Println("📦 日志地址:", vLog.Address.Hex())
        fmt.Println("📝 topics:")
        for i, topic := range vLog.Topics {
            fmt.Printf("  - topic[%d]: %s\n", i, topic.Hex())
        }
        fmt.Println("📨 data:", hex.EncodeToString(vLog.Data))
    }
}

• 测试 在这个测试中，我们连接了 infura 的 rpc 节点，通过 eth_getTransactionReceipt 获取了 log 的日志，并筛选出符合 Transfer 事件签名的事件。

eth_getLogs

func TestGetLogs(t *testing.T) {
    // 加载 .env 文件
    err := godotenv.Load()
    if err != nil {
        log.Fatalf("Error loading .env file")
    }

    // 获取环境变量
    rpcURL := os.Getenv("ETHEREUM_RPC_URL")
    apiKey := os.Getenv("ETHEREUM_API_KEY")
    contractAddress := os.Getenv("CONTRACT_ADDRESS")
    eventSignature := os.Getenv("EVENT_SIGNATURE")

    if rpcURL == "" || apiKey == "" || contractAddress == "" {
        log.Fatal(" ❌ Missing required environment variables")
    }

    // 连接以太坊节点（使用你的 RPC）
    client, err := ethclient.Dial(rpcURL + apiKey)
    if err != nil {
        log.Fatalf("❌ 连接失败: %v", err)
    }
    defer client.Close()

    // 构造筛选参数
    var addresses []common.Address
    addresses = append(addresses, common.HexToAddress(contractAddress))
    var eventSignatures [][]common.Hash
    eventSignatures = append(eventSignatures, []common.Hash{common.HexToHash(eventSignature)})
    query := ethereum.FilterQuery{
        BlockHash: nil,
        FromBlock: big.NewInt(22341322),
        ToBlock:   big.NewInt(22341324),
        Addresses: addresses,
        Topics:    eventSignatures,
    }
    // 发起调用
    logs, err := client.FilterLogs(context.Background(), query)
    if err != nil {
        log.Fatalf("❌ 获取 log失败: %v", err)
    }
    // 遍历日志
    for _, vLog := range logs {
        fmt.Println("📦 日志地址:", vLog.Address.Hex())
        fmt.Println("📝 topics:")
        for i, topic := range vLog.Topics {
            fmt.Printf("  - topic[%d]: %s\n", i, topic.Hex())
        }
        fmt.Println("📨 data:", hex.EncodeToString(vLog.Data))
    }

}

• 测试

eth_getLogs 方法获取日志事件相对来讲更简单。只需传入查询参数：

1. fromBlock: 起始区块号
2. toBlock: 终止区块号
3. address: 合约地址（可以单个也可以传一个数组）
4. topics: 事件的 Topic （可以传二维数组）

   topics := [][]common.Hash{
   {transferSigHash},                // topic[0] = 事件签名
   {fromAHash, fromBHash},           // topic[1] = from 是 A 或 B
   {toCHash},                        // topic[2] = to 是 C
   }