使用ethers监听多合约时事件丢失的原因及解决方案咨询

• 节点限流与资源限制：免费/共享RPC节点通常有并发连接、请求频率的限制。同时监听100个合约会创建大量独立的事件订阅请求，触发节点限流机制，导致部分订阅被拒绝或事件推送中断。
• 事件回调阻塞：如果事件处理逻辑（如异步写数据库、调用外部API）耗时较长，会阻塞事件处理队列。当大量事件同时到来时，队列积压会直接导致后续事件被丢弃。
• 低效的订阅方式：每个合约单独调用contract.on()会创建独立的WebSocket订阅，过多订阅会占用客户端和节点的大量资源，引发连接不稳定、丢包问题。
• 未处理区块回滚：以太坊网络偶尔会出现区块回滚，若监听逻辑未处理这种情况，会丢失回滚区块内的事件，或重复处理新区块的事件。

1. 优化节点选择与订阅方式 #

• 使用批量事件过滤：放弃单个合约单独订阅，通过provider.on()直接监听多个合约的目标事件，仅创建一个订阅，大幅降低资源消耗：

const ethers = require('ethers');
const ERC20_ABI = [/* 你的ERC20 ABI */];

// 100个目标合约地址数组
const targetContracts = ["0xContract1", "0xContract2", "..."];
// Transfer事件的签名哈希
const transferEventSig = ethers.id("Transfer(address,address,uint256)");

const provider = new ethers.WebSocketProvider("wss://your-rpc-node-url");

// 批量监听所有合约的Transfer事件
provider.on({
  topics: [transferEventSig],
  address: targetContracts // 支持传入地址数组
}, async (log) => {
  // 解析事件数据
  const iface = new ethers.Interface(ERC20_ABI);
  const event = iface.parseLog(log);
  const { from, to, value } = event.args;
  
  // 快速将事件数据传入处理队列，避免阻塞监听
  await pushToProcessingQueue({
    contract: log.address,
    from,
    to,
    value: value.toString(),
    blockNumber: log.blockNumber
  });
});

• 选择专业RPC节点：付费节点通常有更高的并发限制和更稳定的连接，能支撑大量事件订阅需求。

2. 优化事件回调逻辑 #

• 用异步队列解耦监听与处理：让事件回调仅负责接收事件并放入队列，耗时操作交给单独的处理器执行，避免阻塞监听：

const Queue = require('bullmq');
// 创建事件处理队列
const eventQueue = new Queue('transfer-events');

// 监听事件并入队
provider.on(/* 批量过滤条件 */, async (log) => {
  const iface = new ethers.Interface(ERC20_ABI);
  const event = iface.parseLog(log);
  await eventQueue.add('process-transfer', {
    contractAddress: log.address,
    from: event.args.from,
    to: event.args.to,
    value: event.args.value.toString(),
    txHash: log.transactionHash,
    logIndex: log.logIndex
  });
});

// 单独的处理器执行耗时操作
eventQueue.process(async (job) => {
  const { data } = job;
  // 这里执行写数据库、调用API等耗时操作
  await saveTransferRecord(data);
});

• 捕获回调错误：未捕获的错误会导致监听中断，务必在回调中包裹try/catch：

provider.on(/* 过滤条件 */, async (log) => {
  try {
    // 事件解析与入队逻辑
  } catch (err) {
    console.error(`处理日志 ${log.transactionHash} 出错:`, err);
    // 可选：记录错误日志，避免监听终止
  }
});

3. 处理区块回滚与同步问题 #

• 监听区块高度，处理回滚：维护本地区块高度记录，当出现回滚时重新拉取对应区间的事件：

let latestProcessedBlock = 0;

provider.on('block', async (currentBlock) => {
  if (currentBlock < latestProcessedBlock) {
    // 发生回滚，重新处理回滚区间的事件
    const logs = await provider.getLogs({
      topics: [transferEventSig],
      address: targetContracts,
      fromBlock: currentBlock + 1,
      toBlock: latestProcessedBlock
    });
    // 重新将这些日志加入处理队列
    for (const log of logs) {
      await pushToProcessingQueue(parseLog(log));
    }
  }
  latestProcessedBlock = currentBlock;
});

• 启动时同步历史事件：避免漏抓服务启动前的事件，先同步历史数据再监听实时事件：

// 先同步最近1000块的历史事件
const historicalLogs = await provider.getLogs({
  topics: [transferEventSig],
  address: targetContracts,
  fromBlock: 'latest' - 1000,
  toBlock: 'latest'
});
// 处理历史日志
for (const log of historicalLogs) {
  await pushToProcessingQueue(parseLog(log));
}

// 再启动实时监听
provider.on(/* 过滤条件 */, (log) => {
  pushToProcessingQueue(parseLog(log));
});

4. 关于单文件监听单个合约的问题 #

完全不需要这种繁琐的方式，推荐使用批量订阅+异步队列的模式，在一个进程内处理所有合约的事件监听，通过队列解耦监听与处理逻辑，既高效又易维护。

• 校验订阅状态：可通过provider._subscriptions（ethers.js）查看当前订阅是否正常，异常时重新创建订阅。
• 避免重复处理：用log.transactionHash + log.logIndex作为唯一标识，存入数据库时做去重校验。

内容的提问来源于stack exchange，提问作者byggy