Spring Batch中如何避免CSV转MQ任务重跑时的消息重复

要解决大型CSV处理中断后重跑的重复数据问题，核心是实现断点续跑+保证消息处理的幂等性，下面是几个可落地的方案：

1. 断点续跑 + 消费端幂等校验 #

• 记录处理进度：每次成功处理完一个CSV块（或固定行数）后，把当前处理到的位置（比如文件偏移量、行号、块ID）写入持久化存储（本地文本文件、SQLite/MySQL数据库都可以）。注意进度记录必须保证原子性——比如写文件时先写临时文件，成功后再替换正式进度文件；用数据库的话就把进度更新放在事务里，避免写一半中断导致进度错乱。
• 重跑时恢复进度：作业启动时先读取进度记录，直接跳转到上次中断的位置继续处理，跳过已经处理过的块/行。
• 消费端幂等处理：给每条CSV数据生成唯一标识（比如用文件名+行号、数据自带的业务唯一ID，或者对整行数据做哈希），消费端收到消息后，先检查这个标识是否已经处理过（可以存在本地缓存、数据库的去重表），如果已处理就直接跳过，否则再执行业务逻辑。

2. 块级别的事务性提交 #

把「读取一块数据+发送MQ消息」做成原子操作，确保要么整个块的消息都发送成功，要么都不发送：

• 本地消息表方案：先将待发送的块数据写入本地数据库的消息表（状态标记为「待发送」），然后批量发送到MQ；发送成功后，更新消息表的状态为「已发送」。作业中断重跑时，先扫描本地消息表，把「待发送」的块重新发送，「已发送」的直接跳过。
• MQ事务消息：如果使用的MQ支持事务消息（比如RocketMQ），可以在发送块消息前开启事务，确认所有消息都成功发送到MQ后再提交事务；如果发送失败则回滚事务，不记录进度。

3. 利用MQ的精确一次语义 #

如果你的MQ支持精确一次（Exactly-Once）语义，可以直接依赖MQ的能力避免重复：

• 比如Kafka生产者开启enable.idempotence=true，同时配置事务ID，生产者会自动避免重复发送相同的消息；消费者配合事务提交，保证消息只被处理一次。
• 注意这种方案依赖MQ本身的特性，需要确认你的MQ是否支持，以及配置是否正确。

• 进度记录的存储要可靠：绝对不能用内存存储，一定要持久化到磁盘或数据库，否则重启后进度就丢失了。
• 唯一标识的生成要靠谱：如果CSV本身有业务唯一ID（比如用户ID、订单号），优先用这个；没有的话再用文件名+行号（前提是CSV文件不会被修改，否则行号会失效）。

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