go-ethereum订阅以太坊事件日志主题能否运行时动态扩展

已经创建成功的日志订阅不支持运行时动态修改过滤规则，要新增监听主题必须重建订阅。

调用SubscribeFilterLogs时，你传入的FilterQuery（包含合约地址、主题列表等所有过滤条件）会被序列化为RPC参数通过Websocket发送到连接的以太坊节点（比如你用的Infura节点），节点会根据你首次提交的规则创建独立的日志推送任务，后续只会返回匹配这套初始规则的日志。不管是go-ethereum客户端本身，还是以太坊的RPC接口规范，都没有提供修改已生效订阅规则的能力。

你可以在服务层维护一套订阅状态来实现动态增删主题的需求，核心逻辑是规则变化时重建订阅，具体做法：

• 用带读写锁的结构全局维护当前生效的订阅配置：包括监听的合约地址、主题列表、当前活动的订阅实例、日志channel、协程取消函数，保证REST接口和日志处理协程的并发安全
• 收到新增主题的REST请求时，先加写锁：
  • 校验新主题是否已经在监听列表里，避免重复操作
  • 调用旧订阅的Unsubscribe()方法取消旧订阅，执行协程取消函数，关闭旧的日志channel，避免连接和goroutine泄漏
  • 把新主题合并到本地维护的主题列表中，用更新后的完整过滤规则重新调用SubscribeFilterLogs创建新订阅
  • 启动新的goroutine从新的日志channel消费日志，执行业务逻辑
• 如果对数据完整性要求高，可以在新订阅创建成功后，补拉一次旧订阅取消到新订阅生效这个区块区间内的匹配日志，避免切换间隙的漏数问题。

补充：你也可以选择初始订阅时把主题列表传空，接收对应合约的所有事件日志，在本地代码层做主题过滤。这种方案不需要重建订阅，但如果合约事件量很大，会拉取大量你不需要的日志，浪费带宽，还容易触发节点侧的限流（比如Infura免费版的配额限制），如果后续还要动态新增监听的合约地址，这种方案依然需要重建订阅，优先选重建订阅的方案更灵活。

import (
    "context"
    "sync"
    "github.com/ethereum/go-ethereum"
    "github.com/ethereum/go-ethereum/common"
    "github.com/ethereum/go-ethereum/core/types"
    "github.com/ethereum/go-ethereum/ethclient"
)

type ContractSubManager struct {
    mu           sync.RWMutex
    client       *ethclient.Client
    contractAddr common.Address
    watchTopics  []common.Hash
    activeSub    ethereum.Subscription
    logChan      chan types.Log
    subCancel    context.CancelFunc
}

func NewContractSubManager(client *ethclient.Client, contractAddr common.Address, initTopics []common.Hash) *ContractSubManager {
    m := &ContractSubManager{
        client:       client,
        contractAddr: contractAddr,
        watchTopics:  initTopics,
    }
    // 初始化第一次订阅
    _ = m.refreshSubLocked()
    return m
}

// 供REST接口调用的新增主题方法
func (m *ContractSubManager) AddWatchTopic(newTopic common.Hash) error {
    m.mu.Lock()
    defer m.mu.Unlock()

    // 主题已存在无需操作
    for _, t := range m.watchTopics {
        if t == newTopic {
            return nil
        }
    }
    m.watchTopics = append(m.watchTopics, newTopic)
    // 重建订阅
    return m.refreshSubLocked()
}

// 内部方法：重建订阅，调用时必须持有写锁
func (m *ContractSubManager) refreshSubLocked() error {
    // 清理旧订阅
    if m.activeSub != nil {
        m.activeSub.Unsubscribe()
    }
    if m.subCancel != nil {
        m.subCancel()
    }
    if m.logChan != nil {
        close(m.logChan)
    }

    // 创建新订阅
    ctx, cancel := context.WithCancel(context.Background())
    logCh := make(chan types.Log)
    sub, err := m.client.SubscribeFilterLogs(ctx, ethereum.FilterQuery{
        Addresses: []common.Address{m.contractAddr},
        Topics:    [][]common.Hash{m.watchTopics},
    }, logCh)
    if err != nil {
        cancel()
        return err
    }

    // 更新状态
    m.activeSub = sub
    m.logChan = logCh
    m.subCancel = cancel
    // 启动日志处理协程
    go m.processLogs(logCh)
    return nil
}

// 实际业务日志处理逻辑
func (m *ContractSubManager) processLogs(ch chan types.Log) {
    for log := range ch {
        // 这里写你的事件解析、业务处理逻辑
    }
}

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