非并发代码实现Ethereum智能合约查询的并发优化及相关问题咨询

一、最简并发实现：线程池（ThreadPoolExecutor） #

链上合约查询属于IO密集型操作，使用标准库的concurrent.futures.ThreadPoolExecutor是最适合的无侵入方案，无需修改原有库代码即可实现并发请求。

代码示例 #

import library
from concurrent.futures import ThreadPoolExecutor

items = [
    "address1",
    "address2",
    "address3",
    # 可添加更多地址
]

def fetch_info(address):
    try:
        return library.get_smartcontractinfo(address)
    except Exception as e:
        print(f"获取{address}数据失败: {str(e)}")
        return None

if __name__ == '__main__':
    # max_workers控制并发数，需根据链节点的速率限制调整
    with ThreadPoolExecutor(max_workers=5) as executor:
        # 批量提交查询任务
        results = executor.map(fetch_info, items)
        # 遍历输出结果
        for addr, data in zip(items, results):
            if data:
                print(f"{addr}: {data}")

executor.map会自动将每个地址分配给线程池中的线程执行，等待所有任务完成后返回结果列表，相比串行循环能大幅提升效率。

二、应对速率限制 #

链上节点普遍存在请求频率限制，直接高并发可能导致请求被拒绝，可通过以下方式处理：

• 控制并发数：调整ThreadPoolExecutor的max_workers参数，比如节点允许每秒5次请求，就将值设为5，避免超出限制。
• 添加请求间隔：在查询函数中加入短暂休眠，分散请求压力：

  import time
  def fetch_info(address):
      try:
          time.sleep(0.2)  # 每秒最多5次请求
          return library.get_smartcontractinfo(address)
      except Exception as e:
          print(f"获取{address}数据失败: {str(e)}")
          return None
  

• 信号量精确限流：使用threading.Semaphore限制同时执行的请求数：

  import threading
  semaphore = threading.Semaphore(5)  # 同时最多5个请求
  
  def fetch_info(address):
      with semaphore:
          try:
              return library.get_smartcontractinfo(address)
          except Exception as e:
              print(f"获取{address}数据失败: {str(e)}")
              return None
  

• 指数退避重试：遇到速率限制错误时，自动重试并逐步延长间隔：

  def fetch_info(address, retries=3):
      delay = 1
      for attempt in range(retries):
          try:
              return library.get_smartcontractinfo(address)
          except Exception as e:
              if attempt < retries - 1:
                  time.sleep(delay)
                  delay *= 2  # 指数退避
              else:
                  print(f"获取{address}数据失败，重试{retries}次后仍失败: {str(e)}")
                  return None
  

三、无需改写库的批量调用可能性 #

如果原库没有提供批量查询接口，无法实现真正意义上的批量调用（即单次请求包含多个查询），但可以通过并发单个请求模拟批量调用的效率。

若链上节点支持批量JSON-RPC请求，你可以绕过原库直接发送批量请求，但这需要自行处理签名、请求封装等逻辑，复杂度较高。对于大多数场景，用线程池并发单个请求是更简单的无侵入替代方案，能达到近似批量调用的效率。

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