<!--StartFragment-->

Viem 监听合约事件完整指南

目录

1. 基本概念
2. 监听方法概览
3. watchContractEvent 详解
4. watchEvent 详解
5. 过滤器配置
6. 错误处理
7. 性能优化
8. 实战示例

基本概念

什么是合约事件？

• Solidity 合约中通过 event关键字声明
• 用于记录链上发生的特定操作
• 比直接查询状态更高效的监听方式
• Gas 成本较低（日志存储费用）

Viem 事件监听优势

• 类型安全: 完整的 TypeScript 支持
• 实时监听: WebSocket 实时推送
• 过滤器: 强大的事件过滤能力
• 错误恢复: 自动重连机制

监听方法概览

Viem 提供两种主要的事件监听方式：

watchContractEvent 详解

基本用法

import { createPublicClient, http, parseAbi } from 'viem'
import { mainnet } from 'viem/chains'

const publicClient = createPublicClient({
  chain: mainnet,
  transport: http()
})

// 定义合约 ABI
const erc20Abi = parseAbi([
  'event Transfer(address indexed from, address indexed to, uint256 value)',
  'event Approval(address indexed owner, address indexed spender, uint256 value)'
])

// 监听 Transfer 事件
const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984', // UNI 合约
  abi: erc20Abi,
  eventName: 'Transfer',
  onLogs: (logs) => {
    logs.forEach((log) => {
      console.log('Transfer 事件:', {
        发送方: log.args.from,
        接收方: log.args.to,
        金额: log.args.value,
        交易哈希: log.transactionHash,
        区块号: log.blockNumber
      })
    })
  }
})

// 停止监听
// unwatch()

参数过滤

// 只监听特定地址的转账
const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  args: {
    from: '0x742d35Cc6634C0532925a3b8D5C9eA6A2A3b8D5C' // 只监听从这个地址转出的
  },
  onLogs: (logs) => {
    console.log('特定地址转出:', logs)
  }
})

多事件监听

// 监听多个事件类型
const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  onLogs: (logs) => {
    logs.forEach((log) => {
      if (log.eventName === 'Transfer') {
        console.log('转账事件:', log.args)
      } else if (log.eventName === 'Approval') {
        console.log('授权事件:', log.args)
      }
    })
  }
})

watchEvent 详解

原始事件监听

import { parseAbiItem } from 'viem'

// 使用 watchEvent 进行更底层的监听
const unwatch = publicClient.watchEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  event: parseAbiItem('event Transfer(address indexed, address indexed, uint256)'),
  onLogs: (logs) => {
    logs.forEach((log) => {
      console.log('原始事件数据:', {
        topics: log.topics,
        data: log.data,
        完整日志: log
      })
    })
  }
})

监听所有合约事件

// 监听特定主题的所有事件（不指定地址）
const unwatch = publicClient.watchEvent({
  event: parseAbiItem('event Transfer(address indexed, address indexed, uint256)'),
  onLogs: (logs) => {
    // 会收到所有合约的 Transfer 事件
    logs.forEach((log) => {
      console.log('全网 Transfer 事件:', log)
    })
  }
})

过滤器配置

区块范围过滤

const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  fromBlock: 18000000n, // 从特定区块开始
  toBlock: 'latest',    // 监听最新区块
  onLogs: (logs) => {
    console.log('区块范围内的事件:', logs)
  }
})

多参数过滤

// 复杂的参数过滤
const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  args: {
    from: ['0x742d35Cc6634C0532925a3b8D5C9eA6A2A3b8D5C', '0x另外一个地址'],
    to: '0x特定接收地址'
  },
  onLogs: (logs) => {
    console.log('复杂过滤结果:', logs)
  }
})

错误处理

基本错误处理

const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  onLogs: (logs) => {
    // 正常处理逻辑
  },
  onError: (error) => {
    console.error('监听事件时发生错误:', error)

    // 根据错误类型处理
    if (error.message.includes('connection')) {
      console.log('网络连接问题，尝试重连...')
      // 这里可以添加重连逻辑
    }
  },
  poll: true, // 启用轮询模式（WebSocket 失败时回退）
  pollingInterval: 4000 // 轮询间隔 4 秒
})

自动重试机制

class EventWatcher {
  private unwatch: (() => void) | null = null
  private retryCount = 0
  private maxRetries = 5

  startWatching() {
    this.unwatch = publicClient.watchContractEvent({
      address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
      abi: erc20Abi,
      eventName: 'Transfer',
      onLogs: this.handleLogs.bind(this),
      onError: this.handleError.bind(this)
    })
  }

  private handleLogs(logs: any[]) {
    // 重置重试计数
    this.retryCount = 0
    console.log('收到事件:', logs)
  }

  private handleError(error: Error) {
    console.error('监听错误:', error)
    this.retryCount++

    if (this.retryCount <= this.maxRetries) {
      console.log(`尝试重连... (${this.retryCount}/${this.maxRetries})`)
      setTimeout(() => this.startWatching(), 2000 * this.retryCount)
    } else {
      console.error('达到最大重试次数，停止监听')
    }
  }

  stopWatching() {
    if (this.unwatch) {
      this.unwatch()
      this.unwatch = null
    }
  }
}

性能优化

批量处理事件

let batchQueue: any[] = []
let batchTimer: NodeJS.Timeout | null = null

const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  onLogs: (logs) => {
    // 批量处理，减少频繁操作
    batchQueue.push(...logs)

    if (batchTimer) clearTimeout(batchTimer)

    batchTimer = setTimeout(() => {
      this.processBatch(batchQueue)
      batchQueue = []
    }, 1000) // 1秒批量处理一次
  }
})

private processBatch(logs: any[]) {
  console.log(`批量处理 ${logs.length} 个事件`)
  // 批量插入数据库或其他操作
}

使用 WebSocket 传输

import { createPublicClient, webSocket } from 'viem'
import { mainnet } from 'viem/chains'

// 使用 WebSocket 实现实时监听（推荐）
const publicClient = createPublicClient({
  chain: mainnet,
  transport: webSocket('wss://eth-mainnet.g.alchemy.com/v2/your-api-key')
})

// WebSocket 连接更高效，支持服务端推送
const unwatch = publicClient.watchContractEvent({
  address: '0x1f9840a85d5af5bf1d1762f925bdaddc4201f984',
  abi: erc20Abi,
  eventName: 'Transfer',
  onLogs: (logs) => {
    console.log('实时事件:', logs)
  }
})

实战示例

示例1: DEX 交易监控

import { parseAbi } from 'viem'

const uniswapV3PoolAbi = parseAbi([
  'event Swap(address indexed sender, address indexed recipient, int256 amount0, int256 amount1, uint160 sqrtPriceX96, uint128 liquidity, int24 tick)',
  'event Mint(address indexed sender, address indexed owner, int24 tickLower, int24 tickUpper, uint128 amount, uint256 amount0, uint256 amount1)',
  'event Burn(address indexed owner, int24 tickLower, int24 tickUpper, uint128 amount, uint256 amount0, uint256 amount1)'
])

class DexMonitor {
  private unwatchFunctions: (() => void)[] = []

  startMonitoring(poolAddress: string) {
    // 监听交易事件
    const swapUnwatch = publicClient.watchContractEvent({
      address: poolAddress,
      abi: uniswapV3PoolAbi,
      eventName: 'Swap',
      onLogs: (logs) => {
        logs.forEach(log => {
          console.log('交易发生:', {
            交易对: poolAddress,
            输入数量: log.args.amount0,
            输出数量: log.args.amount1,
            价格: log.args.sqrtPriceX96,
            交易哈希: log.transactionHash
          })
        })
      }
    })

    // 监听流动性事件
    const liquidityUnwatch = publicClient.watchContractEvent({
      address: poolAddress,
      abi: uniswapV3PoolAbi,
      eventName: ['Mint', 'Burn'],
      onLogs: (logs) => {
        logs.forEach(log => {
          if (log.eventName === 'Mint') {
            console.log('添加流动性:', log.args)
          } else {
            console.log('移除流动性:', log.args)
          }
        })
      }
    })

    this.unwatchFunctions.push(swapUnwatch, liquidityUnwatch)
  }

  stopMonitoring() {
    this.unwatchFunctions.forEach(unwatch => unwatch())
    this.unwatchFunctions = []
  }
}

示例2: NFT 市场监控

const seaportAbi = parseAbi([
  'event OrderFulfilled(bytes32 orderHash, address offerer, address zone, address recipient, tuple(uint8 itemType, address token, uint256 identifier, uint256 amount)[] offer, tuple(uint8 itemType, address token, uint256 identifier, uint256 amount, address recipient)[] consideration)'
])

class NFTMarketMonitor {
  constructor(private marketAddress: string) {}

  startWatching() {
    return publicClient.watchContractEvent({
      address: this.marketAddress,
      abi: seaportAbi,
      eventName: 'OrderFulfilled',
      onLogs: (logs) => {
        logs.forEach(log => {
          const { offerer, recipient, offer, consideration } = log.args

          console.log('NFT 交易完成:', {
            卖家: offerer,
            买家: recipient,
            出售的NFT: offer.filter(item => item.itemType === 2), // ERC721
            支付代币: consideration.filter(item => item.itemType === 0), // 原生代币
            交易哈希: log.transactionHash
          })
        })
      }
    })
  }
}

示例3: 跨链桥事件监控

const bridgeAbi = parseAbi([
  'event Deposit(uint256 chainId, address token, address from, address to, uint256 amount, uint256 nonce)',
  'event Withdraw(bytes32 depositHash, address token, address to, uint256 amount)'
])

class CrossChainBridgeMonitor {
  private unwatch: (() => void) | null = null

  startBridgeMonitoring(bridgeAddress: string) {
    this.unwatch = publicClient.watchContractEvent({
      address: bridgeAddress,
      abi: bridgeAbi,
      onLogs: (logs) => {
        logs.forEach(log => {
          if (log.eventName === 'Deposit') {
            this.handleDeposit(log.args)
          } else if (log.eventName === 'Withdraw') {
            this.handleWithdraw(log.args)
          }
        })
      }
    })
  }

  private handleDeposit(args: any) {
    console.log('跨链存款:', {
      目标链: args.chainId,
      代币: args.token,
      存款人: args.from,
      接收人: args.to,
      金额: args.amount
    })
  }

  private handleWithdraw(args: any) {
    console.log('跨链取款:', {
      存款哈希: args.depositHash,
      代币: args.token,
      接收人: args.to,
      金额: args.amount
    })
  }

  stop() {
    if (this.unwatch) {
      this.unwatch()
      this.unwatch = null
    }
  }
}

最佳实践

1. 使用 WebSocket 传输: 实时性更好，资源消耗更低
2. 合理设置过滤器: 避免监听不必要的事件
3. 实现错误恢复: 网络中断时自动重连
4. 批量处理事件: 高频事件场景下提升性能
5. 及时清理监听器: 避免内存泄漏
6. 类型安全: 充分利用 TypeScript 类型推断

<!--EndFragment-->