一、先说说真实世界里的一次“翻车”

很多团队用Storm的时候，最头疼的不是逻辑怎么算，而是任务跑着跑着突然重启，原本稳定的数据流就开始乱套。我见过一个实时订单统计任务，本来一切正常，一次集群抖动后，第二天早上出来的报表里多了一堆重复订单，同时又有一些订单消失不见。大家排查了很久，最后把锅扣到了Kafka的偏移量上。说白了，Storm从Kafka拿数据，就像快递员往仓库送包裹，仓库人员每收到一个包裹都得记一笔“收到第几号”。如果记早了，包裹还没拆封就当成已处理，后面出了问题没处找；如果记晚了，仓库明明已经处理完，重新启动后又会把同一个包裹再处理一遍。这就是无界数据流里最折磨人的偏移量同步问题。

二、先搞清楚Kafka里的偏移量是个什么玩意儿

Kafka把一个主题分成多个分区，每个分区内部是有顺序的，每一条新消息都会拿到一个递增编号，这个编号就是偏移量。消费者需要告诉Kafka“我已经读到第几号了”，这个动作叫提交偏移量。提交得靠前，说明你确认读到后面的位置了，前面的消息还没处理完也就不再管；提交得靠后，说明你只确认读到前面的位置，后面的消息下次还会再读一次。

2.1 谁在帮我们记这个账

Kafka的消费组里有一个协调者，负责给消费者分配分区。每个分区在同一时刻只能被同一个消费组里的一个消费者读取，消费者每隔一段时间就会汇报“我读到哪了”。Storm从Kafka拉取数据的时候，本质上自己也是一个消费者，所以也必须遵守Kafka这套记账规则。如果Storm完全不理会这套规则，那Kafka和Storm各记各的账，最后肯定对不上。

2.2 自动记账看着省心，其实容易出大事

Kafka客户端默认会打开自动提交开关，每隔几秒自动提交一次当前拉取到的偏移量。问题就出在这个“自动”上。Storm从Kafka里拉出一条消息，不等于这条消息已经被计算完毕。打个比方，自动提交逻辑觉得你已经把蛋糕吃了，可实际上蛋糕还在烤箱里。如果这时候系统崩了，Storm重启后会从已经提交的位置继续读，那些“已经拉出来但还没算完”的数据就永远找不回来了。反过来，如果处理完了但迟迟没提交，重启后会重复处理一大批数据。

三、Storm和Kafka集成时的三个坑

3.1 坑一：提交偏移量不是跟着消息处理走

很多项目图省事，直接用默认的自动提交配置，每5秒提交一次。表面上看一切正常，等业务量突然涨上来，某个Bolt处理速度稍微慢一点，自动提交的位置已经远远跑到前面去了。这就像看在线视频，进度条已经缓冲到结尾，但你才看到一半，这时直接关掉播放器，下次再打开只能从结尾继续看，中间那一大段剧情就丢掉了。

3.2 坑二：消息失败了却不知道找谁算账

Storm有一个很聪明的acker机制，一条数据从Spout发出去，经过所有Bolt处理成功，才会收到确认。但Kafka提交偏移量的时候，只关心“这个分区我已经读到哪个编号”，它完全不知道这条数据在Storm里有没有被后面的逻辑成功处理。这两个机制各干各的，结果就是：Kafka说已经消费了，Storm说还没算完。两边责任不清晰，出问题只能靠人肉去对数据。

3.3 坑三：并行度一改，偏移量就串位

实时计算里，调整并行度是家常便饭。你给Spout增加了两个并发线程，原来由一个线程负责的分区就会被拆分到多个线程上。这时候如果偏移量的归属还是按老状态来，新线程可能不知道该从哪儿开始读，甚至会出现几个线程同时消费同一个分区的情况，轻则重复计算，重则直接报错。

四、破解思路：把偏移量的“确认权”交还给业务逻辑

既然自动提交不靠谱，那就不让Kafka自己乱提交。正确做法是把自动提交关掉，让KafkaSpout等tuple真正处理完，再提交偏移量。这就是我们常说的“消息处理完再确认”。下面从代码层面把这个方案拆开讲。

先统一说明一下，所有示例用的技术栈是Java 8、Storm 1.2以上、Kafka 2.x。

4.1 第一步：关掉自动提交，开启at-least-once

使用Storm官方推荐的storm-kafka-client组件，可以很干净地配置偏移量提交策略。看下面的示例。

import org.apache.storm.kafka.spout.KafkaSpout;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.KafkaSpoutConfig.FirstPollOffsetStrategy;
import org.apache.storm.kafka.spout.KafkaSpoutConfig.ProcessingGuarantee;
import org.apache.kafka.clients.consumer.ConsumerConfig;

public class RobustKafkaSpout {

    public static KafkaSpout<String, String> create() {
        // 配置Kafka服务地址，多个节点用逗号分隔，这里写的是示例地址
        KafkaSpoutConfig<String, String> config = KafkaSpoutConfig
                .builder("node1:9092,node2:9092", "order_topic")
                // 给消费组起一个清晰的名字，重启后能找回上次消费位置
                .setGroupId("storm-offset-demo")
                // 核心动作：关闭Kafka客户端的自动提交
                .setProp(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false")
                // 使用“至少一次”语义，保证消息不丢，但可能重复
                .setProcessingGuarantee(ProcessingGuarantee.AT_LEAST_ONCE)
                // 没有历史偏移量时，从最早的数据开始读，避免丢数据
                .setFirstPollOffsetStrategy(FirstPollOffsetStrategy.UNCOMMITTED_EARLIEST)
                // 每隔10秒提交一次偏移量，具体时间看业务对重复的容忍度
                .setOffsetCommitPeriodMs(10_000)
                // 允许最多10万条未确认的记录，超过这个数Spout会限流
                .setMaxUncommittedOffsets(100_000)
                .build();
        return new KafkaSpout<>(config);
    }
}

这个配置的关键在于，提交偏移量的时机从“拉取后”改成了“确认后”。一个tuple只有被acker标记为成功，KafkaSpout才会向前移动偏移量；如果tuple处理失败，偏移量就停在原地，等待下次重试。

4.2 第二步：写一个会“负责到底”的Bolt

业务处理Bolt最好使用BaseBasicBolt，因为它的execute方法方法执行完后会自动帮你确认tuple，不用手动调用ack，代码更干净。下面这个示例负责解析订单数据，并把解析结果发到下游。

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

public class OrderParserBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        // 从KafkaSpout里取出订单号和原始数据
        String orderId = tuple.getStringByField("orderId");
        String rawData = tuple.getStringByField("value");

        // 这里可以做真正的业务处理，比如拆分字段、清洗垃圾数据
        // 注意：如果这里抛出异常，tuple会被标记为失败，偏移量不会往前走
        if (rawData == null || rawData.length() == 0) {
            // 遇到脏数据时先打印日志，跳过去继续处理，避免任务卡死
            System.out.println("订单" + orderId + "是空的，跳过");
            return;
        }

        // 模拟业务处理成功
        System.out.println("订单" + orderId + "解析完成，数据长度" + rawData.length());

        // 把结果发往下一个Bolt
        collector.emit(new Values(orderId, rawData));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 告诉Storm本Bolt会输出orderId和value两个字段
        declarer.declare(new Fields("orderId", "value"));
    }
}

4.3 第三步：把Spout和Bolt串成一个完整拓扑

下面这段代码创建了一个完整拓扑，把Spout、解析Bolt、入库Bolt串起来。为了演示完整，我把入库Bolt也写了出来。

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;

public class OffsetSyncTopology {

    public static void main(String[] args) throws Exception {
        TopologyBuilder builder = new TopologyBuilder();

        // 并行度设为4，表示4个线程同时从Kafka拉数据
        builder.setSpout("kafka-spout", RobustKafkaSpout.create(), 4);

        // 第一个Bolt：解析订单，并行度也是4，随机分发即可
        builder.setBolt("order-parser", new OrderParserBolt(), 4)
               .shuffleGrouping("kafka-spout");

        // 第二个Bolt：订单入库，并行度2，相同订单号必须进同一个Bolt
        builder.setBolt("order-writer", new OrderWriterBolt(), 2)
               .fieldsGrouping("order-parser", new Fields("orderId"));

        Config config = new Config();
        // 本地模式跑起来看效果，生产环境用StormSubmitter提交到集群
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("offset-sync-demo", config, builder.createTopology());

        // 模拟运行30秒后停止，只是为了演示
        Thread.sleep(30_000);
        cluster.shutdown();
    }
}

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Tuple;

public class OrderWriterBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple tuple, BasicOutputCollector collector) {
        String orderId = tuple.getStringByField("orderId");
        String value = tuple.getStringByField("value");
        // 这里假装把数据写入数据库，只打印保存日志
        // 真实项目里可以换成JDBC、Redis或者其它存储
        System.out.println("保存订单" + orderId + "，内容=" + value);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 这个Bolt是最后一个环节，不需要对外输出字段
    }
}

4.4 为什么这样就能解决偏移量同步

因为消息是否提交，完全取决于Storm内部的处理结果。一个tuple从Spout出来，经过解析Bolt、入库Bolt，全部成功，acker才会通知KafkaSpout“这一批可以往前确认了”。如果任何一个环节失败，tuple进入重试流程，KafkaSpout就必须保持原来的偏移量不动。这样一来，Kafka的读数进度和Storm的真实处理进度就绑在了一条绳上，再也不会各跑各的。

五、这套方案的应用场景、优缺点

5.1 优点

最大的优点是数据不容易丢。只要业务Bolt不抛异常，偏移量一定会稳步推进；即使某个tuple失败，也会重试或重新发射。第二个优点是实现简单，完全依靠storm-kafka-client的配置，不用自己写分布式锁或者状态存储。第三个优点是对无界数据流友好，消息源源不断，偏移量按周期批量提交，不会因为每条消息都提交而拖垮Kafka。

5.2 缺点

最明显的缺点是“可能重复”。由于偏移量是每10秒提交一次，在这10秒内消息已经处理完，但还没到提交时间，这时任务崩了，重启后这些消息就会被重新消费。所以下游业务必须支持幂等。另一个缺点是如果某个Bolt处理特别慢，未确认的tuple会越攒越多，对内存造成压力，必须合理设置maxUncommittedOffsets。

5.3 适合用在哪里

日志采集与分析：每天海量日志，偶尔重复一两条对指标影响不大，但不能丢关键错误。订单类业务：要求不丢单，但允许存在重复，数据库写入时用唯一订单号做去重。实时风控：消息不能丢，重复了也能接受，反而可以补充计算。实时大屏展示：对精确度要求不是极高，重复一次也不影响趋势。

六、相关技术延伸：手动提交偏移量的精细控制

如果希望对提交时机有更精细的控制，可以使用KafkaConsumer原生的手动提交接口。在批处理完成后调用commitSync，确保这批消息处理完再提交。下面这段代码展示了基本逻辑，同样基于Java。

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ManualOffsetCommitDemo {

    public void consume() {
        Properties props = new Properties();
        // 指定Kafka服务地址，这里写本机示例
        props.put("bootstrap.servers", "127.0.0.1:9092");
        // 指定消费组名，重启后按组里的提交位置继续消费
        props.put("group.id", "manual-commit-group");
        // 关键一步：关闭自动提交
        props.put("enable.auto.commit", "false");
        // 设置消息反序列化器
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        // 订阅一个主题
        consumer.subscribe(Arrays.asList("order_topic"));

        try {
            while (true) {
                // 每500毫秒拉取一次消息
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
                for (ConsumerRecord<String, String> record : records) {
                    // 这里处理每条消息，比如写入数据库
                    // 如果处理失败抛异常，下面的commitSync就不会执行
                    System.out.println("处理消息，偏移量=" + record.offset());
                }
                // 这一批全部处理完，再提交偏移量
                consumer.commitSync();
            }
        } catch (Exception e) {
            // 遇到异常直接退出，未提交的偏移量会在下次启动时重新消费
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
}

这种手动提交方式的好处是精确，坏处是如果消息处理时间抖动很大，批处理可能会迟迟不结束，导致提交延迟。所以在Storm的拓扑里，我更推荐交给KafkaSpout统一管理，既保证了可靠性，又不需要自己在Bolt里偷偷攥着一个consumer。

七、注意事项梳理

第一，不要再使用默认的自动提交配置，那是让Kafka和Storm互相蒙在鼓里。第二，at-least-once会重复，所有下游都要考虑幂等，比如数据库表加唯一索引。第三，offset提交周期不是越短越好，太短会给Kafka增加压力，太长会扩大重复消费的范围，建议从5秒到30秒之间根据业务测试。第四，如果业务对重复零容忍，不要贸然追求exactly-once，那代价很高，你可以在应用层用事务或去重表实现。第五，调整Spout并行度前，尽量暂停消费，或者选择低峰期操作，同时监控新平行度是否重新分配到了正确分区。第六，监控Spout的uncommitted offset数量，如果持续高位，说明某个Bolt卡了，要马上定位。第七，优先使用storm-kafka-client，而不是老的storm-kafka组件，老库维护不积极，文档也少。

八、总结

Storm消费Kafka最大的痛点，就是偏移量提交时机和处理时机不一致。解决它不需要魔法，只需要把自动提交关掉，让Storm的ack机制来决定偏移量该不该动。消息处理成功才提交，处理失败就不提交，这样数据不会丢，最多在极端情况下重复几次。对于无界数据流来说，重复并不可怕，可怕的是丢了数据却毫无察觉。配合幂等设计、合理监控、谨慎调参，你完全可以放心地让Storm和Kafka一起扛住海量数据流。