Storm中Bolt与Spout执行次数统计问题咨询（多机器场景）

嘿，这个问题我之前做Storm集群监控的时候也碰到过！单机用ConcurrentHashMap确实能搞定计数，但一到多机器部署就歇菜了——毕竟每个Worker都跑在独立的JVM里，内存里的数据根本没法跨节点共享。下面给你几个靠谱的解决方案，按需选就行：

Storm本身就提供了一套成熟的Metrics框架，专门用来收集分布式环境下的组件指标，完美适配多机器集群场景，而且不需要额外引入第三方依赖。步骤大概是这样：

• 自定义Metrics类：继承BaseMetric，用AtomicLong维护线程安全的计数器，实现累加方法。
• 在Spout/Bolt里注册Metrics：在open（Spout）或prepare（Bolt）方法中，通过TopologyContext获取MetricRegistry，把自定义Metrics注册进去。
• 配置Metrics Reporter：Storm支持把数据输出到Console、Graphite、Prometheus等平台，只需要在拓扑配置里添加对应的Reporter即可。

给你贴个简单的代码示例：

// 自定义执行次数计数器Metric
public class ExecutionCountMetric extends BaseMetric {
    private final AtomicLong executeCount = new AtomicLong(0);

    public void increment() {
        executeCount.incrementAndGet();
    }

    public long getTotalCount() {
        return executeCount.get();
    }
}

// 在Bolt中注册并使用
public class MyBusinessBolt extends BaseBasicBolt {
    private ExecutionCountMetric executionMetric;

    @Override
    public void prepare(Map<String, Object> topoConf, TopologyContext context) {
        executionMetric = new ExecutionCountMetric();
        // 注册Metric，指定名称+上报间隔（这里是10秒上报一次）
        context.registerMetric("my-bolt-execution-count", executionMetric, 10);
    }

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        // 你的业务逻辑处理
        executionMetric.increment(); // 每次执行就累加计数
    }
}

集群里所有Worker的Metrics数据会被统一收集到Reporter，你直接在监控平台就能看到全局的执行次数统计了。

如果不想用Storm自带的Metrics，也可以把计数存在分布式存储里，比如Redis、ZooKeeper或者MySQL：

• Redis：用INCR命令做原子性累加，性能拉满，完全适配多线程多机器场景。记得用连接池复用Redis连接，别每次执行都新建连接。
• ZooKeeper：可以用节点的数据字段或者版本号来累加，但性能不如Redis，适合计数频率不高的场景。
• MySQL：用UPDATE table SET count = count + 1 WHERE id = ?做原子更新，但性能是这几个里最低的，不推荐高吞吐场景。

比如Redis的代码片段：

// 假设已经初始化了Jedis连接池
private JedisPool jedisPool;

@Override
public void execute(Tuple tuple, BasicOutputCollector collector) {
    try (Jedis jedis = jedisPool.getResource()) {
        jedis.incr("my-spout-execution-count");
    }
    // 业务逻辑处理
}

如果需要按Tuple类型、来源节点等多维度统计，可以自己实现一个汇总逻辑：

• 每个Worker本地用ConcurrentHashMap或AtomicLong维护分维度的计数。
• 定期把本地计数发送到一个专门的汇总Bolt。
• 汇总Bolt把所有节点的计数累加后，再存储到分布式存储或输出到外部系统。
  这种方式灵活性最高，但需要自己处理数据传输和一致性，复杂度比前两个方案高，适合有特殊统计需求的场景。

最后提醒下：用分布式存储的时候一定要做好资源复用（比如连接池），不然频繁创建连接会拖垮拓扑性能；优先选Storm官方的Metrics API，集成度最高，后期维护也最省心。

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