无消息代理的分布式Spring Batch：用共享数据库作为协调平面

关键要点

• Spring Batch 可以通过将本来就需要的关系型数据库用作协调平面，在无需消息代理的情况下横向扩展。这适合许多团队运行的中小规模集群。
• 该模式解决了原本由消息代理代为处理的问题：根据存活节点数来确定工作规模、为每次执行分配主节点职责而无需常驻协调器，并通过比较并设置写入来认领分区，确保任何分区都不会被静默丢失。
• 节点故障通过心跳和两阶段恢复来处理。它会将过期节点标记为不可达，然后在宽限期窗口之后移除该节点，并重新分配其未完成且可转移的分区。
• 正确性是一份显式契约。可转移工作至少运行一次，因此必须幂等；不可转移工作至多运行一次，失败时会被标记为失败而不是迁移。分区永远不会被静默丢失。
• 数据库协调最适合运维简单性和成本最为重要的场景。它的上限是数据库争用；超过这一点，在超大规模集群或亚秒级延迟要求下，消息代理仍是更好的选择。

引言

Spring Batch 十多年来一直是 JVM 上批处理的默认框架。作业由步骤构成，例如面向块的读取器、处理器和写入器，并通过元数据仓库使运行可重启。对于单机上的夜间抽取、转换、加载（ETL）、财务对账和数据校验工作负载，它很难被取代。

当一台机器不再够用时，压力就会出现。一次曾经能在窗口内完成的对账运行会随业务增长而变大，直到无法按时完成，而自然的应对方式就是把工作分散到多个节点。Spring Batch 既有的答案是远程分区。一个管理步骤将工作划分为多个分区，并通过 Spring Integration 经 Apache Kafka、RabbitMQ 或 JMS 提供者之类的消息代理将分区派发给工作节点。

远程分区是一种久经考验的方法。当消息代理已经是平台的一部分时，它就是正确的选择。但消息代理是额外的运维层级，而许多团队原本根本不会运行一个消息代理。他们已经在运维一个管理良好的关系型数据库（无论如何，Spring Batch 都需要一个关系型数据库作为作业仓库）。团队几乎没有意愿仅仅为了分发批处理分区而去搭建、保护和监控一个消息传递集群。

本文描述的是针对这种常见情况的一种替代方案：仅使用共享数据库，在一个中小规模集群中协调一个分布式、容错的 Spring Batch 作业。这并不是反对使用消息代理，而是说明从这个作业中移除消息代理需要什么、这种方法的边界在哪里，以及消息代理在哪些场景下仍是更好的工具。

单节点批处理在哪里结束，消息代理问题从哪里开始

如果把分布式批处理归结到本质，有四个问题需要通过你所选择的协调机制来回答。第一，工作如何划分，以及划分到多少个工作节点上？划分得太粗，节点会闲置；划分得太细，开销会无谓增加。正确的数量取决于作业启动时有多少可用容量。第二，谁负责某一次给定运行？必须有某个东西来拆分工作、跟踪其完成情况并最终完成作业。在基于消息代理的分区中，这就是运行管理步骤的节点。第三，一个工作节点如何取得某个分区的所有权，而不会让另一个工作节点也运行它？运行两次可能损坏数据。一次都不运行，作业就会静默地不完整。第四，当一个节点在运行中途死亡时会发生什么？它正在处理的工作必须被检测到，并在安全的情况下自动恢复，同时不能违反第三个要求。

消息代理会替你回答其中一些问题。它的队列把每条消息交给单一消费者，并且派发与执行解耦，因此某个工作节点变慢或死亡不会阻塞生产者。移除消息代理后，每个问题都必须用 SQL 显式回答，而且是针对一个为持久共享状态而非消息派发而构建的存储来回答。本文其余部分就是关于如何把这些答案做好。

构建数据库协调的分区

该设计把 Spring Batch 已经需要的数据库视为唯一的协调平面，不依赖消息代理或 Apache ZooKeeper 之类的协调服务，也不依赖部署平台。每一部分集群状态都存放在普通的 SQL 表中，因此整个集群都可以用一条 SELECT 语句检查。三个表扩充了 Spring Batch 自身的 schema。BATCH_NODES 表是节点注册表，每个节点一行，包含其状态（活动或不可达）、最后心跳时间戳以及当前负载。BATCH_PARTITIONS 表跟踪分区生命周期，每个分区一行，并以该分区自己的步骤执行 ID 为键（每个分区本身就是一个 Spring Batch StepExecution，与界定本次运行范围的管理步骤不同），包含其分配到的节点、状态（待处理、已认领、已完成或失败），以及失败时是否可以转移到另一个节点。BATCH_JOB_COORDINATION 表将每个作业执行绑定到其主节点和管理步骤，并带有自己的状态。这里重要的是这三张表的形态，而不是完整的列清单。

这个设计没有移除的一件事，是对共享、持久作业仓库的需求，但这一要求是分布式分区本身固有的，而不是这个模式额外添加的。分区的负载——描述其工作切片的执行上下文——由主节点写入，并由执行它的任意节点读回。然后主节点再读回结果进行聚合。这种交接是跨 JVM 的 Spring Batch 元数据交换，因此存储必须是共享且持久的。Spring Batch 自身的远程分区也施加同样的要求。在那里，消息代理只携带指向分区的指针，从不携带分区数据。这个设计移除的是消息代理，而不是作业仓库。同一个共享数据库同时充当元数据存储和协调平面。

协调在设计上是去中心化的。不存在为整个集群选举出来的常驻主节点。那种做法会重新引入领导选举机制和单点故障，而移除消息代理正是为了避免这些。相反，启动作业的节点仅在该次执行中成为该作业的主节点（后文会介绍它做什么）。在两个不同节点上启动的两个作业，会同时拥有两个不同的主节点。由于主节点身份仅限定于单次作业执行，丢失一个主节点只会影响那一个作业，而不会影响整个集群。没有协调器需要选举，也没有协调器会丢失。

复用 Spring Batch 的分区钩子：入口点

对于 Spring Batch 开发者来说，第一个实际问题是：我的代码在哪里接入，以及其中有多少需要改变？答案是几乎不需要改变，这正是该方法可用的原因。这种简单性来自 Spring Batch 一直拥有的一个扩展点。

Spring Batch 已经把分区步骤拆分为两个独立职责。Partitioner 接口决定工作如何拆分。它产生一组命名的 ExecutionContext 对象，每个分区一个，每个对象描述输入的一个切片。单独的 PartitionHandler 决定这些分区如何执行。内置的 TaskExecutorPartitionHandler 在本地线程中运行它们，而 Spring Batch 自己的远程分区则换入一个 MessageChannelPartitionHandler，通过消息代理发送它们。管理步骤由这两部分再加上一个负责合并结果的 StepExecutionAggregator 组装而成：

new StepBuilder("reconcileStep.manager", jobRepository)
    .partitioner("reconcileStep", clusterAwarePartitioner) // how work is split
    .partitionHandler(clusterAwarePartitionHandler) // how partitions are distributed
    .step(reconcileStep) // the ordinary worker step
    .aggregator(aggregator)
    .build();

那一行 partitionHandler(...) 就是整个接缝。因为 Spring Batch 将分区分发视为可插拔策略，所以该方法提供自己的 PartitionHandler：它把分区行写入数据库，并在原本由消息代理支持的分区处理器占据的位置上监控这些分区。分区器是 Spring Batch 自身 Partitioner 的子类。其他都不需要改动。工作步骤（上面的 reconcileStep）是普通的基于块或 tasklet 的步骤，结果通过标准的 StepExecutionAggregator 聚合，重启也通过标准的 JobRepository 工作。熟悉本地分区的开发者会编写同样的管理步骤，并且只改动两个 bean：Partitioner 和 PartitionHandler。

这涵盖了主节点。把同一个可部署单元变成集群成员是配置，而不是代码。单个标志即可启用该行为，自动配置会注册节点、启动其心跳并启动轮询器，无需消息代理客户端或额外 bean。几个间隔参数可调整该行为，包括心跳节奏、工作节点轮询节奏、每节点并发数，以及驱动故障转移的两个心跳年龄阈值，并带有适合生产的合理默认值（节点约在十秒时被标记为不可达，约在三十秒时被移除）。从那时起，每个节点，无论主节点还是工作节点，都运行一个循环。启动时，它将自己插入 BATCH_NODES 表，并按配置的间隔发送心跳，发布其当前负载，即它当前正在运行的分区数量。每次轮询时，它扫描 BATCH_PARTITIONS 表中分配给自己的行，用后文展示的比较并设置更新来认领它们，然后做一件保持编程模型不变的事：按名称查找工作步骤并运行它。这个名称不是随意的。它就是传给管理步骤上的 .partitioner("reconcileStep", ...) 方法的名称，记录在协调行中，因此工作节点会解析同一个 Spring Batch 步骤 bean，并用分区的 ExecutionContext 执行它。每个节点最多同时运行配置数量的分区；当某个步骤完成时，它会把该分区记录为 COMPLETED 或 FAILED。因此，协调逻辑位于 Spring Batch 本就要求你实现的扩展点之后，通过配置开启，而不是通过新代码开启。

在 Spring Batch 集群上实现它

[LOADING...]

图 1. 架构：每个作业的主节点与工作节点通过共享数据库进行协调。（来源：作者创建。）

图 1 展示了运行时形态。一组常驻的对等 JVM 节点各自在 BATCH_NODES 表中注册并发送心跳。当其中一个节点启动作业时，该节点就充当本次运行的主节点。它查询存活节点，根据这些节点确定拆分规模，写入分区行，并监控直到完成；与此同时，每个节点（包括主节点）都会轮询并执行分配给自己的分区。

管理步骤如何驱动运行

管理步骤并不特殊。它是一个普通的 Spring Batch 分区（管理）步骤，运行在启动作业的那个节点上。让该节点成为主节点的原因，在于 PartitionHandler 运行的位置。当 Spring Batch 执行管理步骤时，它会在该节点上调用处理器的 handle(...) 方法，而基于数据库的 PartitionHandler 会在这里阻塞，直到整个分布式运行完成。在这单次调用内部，主节点按顺序做四件事。首先，它记录本次运行，在 BATCH_JOB_COORDINATION 表中写入一行新记录，把该作业执行与管理步骤的执行 ID 关联起来，作为本次运行的标识，并将其标记为 STARTED。然后，它向 BATCH_PARTITIONS 表为每个分区写入一行 PENDING 记录，每行都标记其分配到的节点。接着，它运行监控循环，按短间隔唤醒，重新分配因节点被移除而成为孤儿的分区（其分配节点已不在 BATCH_NODES 表中），并检查是否仍有分区处于 PENDING 或 CLAIMED 状态。最后，当没有未完成事项时，如果有任何分区以 FAILED 结束，它就让管理步骤失败；否则，它将协调行标记为 COMPLETED，把工作节点的步骤执行交给 StepExecutionAggregator，并完成作业。

注意这里没有什么。工作节点从不调用主节点，主节点也从不调用工作节点。主节点做出的每一个决定，都是通过读写这些行来做出的。这就是为什么丢失一个主节点会被限制在局部，而不会导致集群中断。它的状态在数据库中，而不在协调器的内存中。

企业场景

考虑跨动态供应的云节点进行夜间财务对账。数据集足够大，如果单个节点错过批处理窗口，作业就会受影响；而一个管理良好的关系型数据库已经就位。运维团队没有意愿仅仅为了这个工作负载而引入消息代理。同样的形态也出现在中型 ETL 和数据校验作业中，其中每个分区都是输入的一个切片，如果节点死亡必须可恢复。这些就是该模式所针对的条件。

按存活集群规模确定工作规模

标准的远程分区在知道有多少可用容量之前就拆分工作。网格大小是盲目选择的，通常是一个固定常量，然后把各个部分派发到消息代理中，等容量释放后再被取走。没有消息代理时，你必须直接面对这个问题。在拆分之前，主节点向数据库询问当前哪些节点存活，并将该数量传给分区器。分区器暴露一个方法，接收这个存活节点数并返回分区，因此拆分大小会根据那一时刻实际存在的集群来确定。每个分区都表示为一个 Spring Batch ExecutionContext，这是框架用于承载分区输入的键值载体。定义一个工作切片的参数（行范围、文件名和 ID 窗口）被放入其中，工作步骤再读回这些参数，以知道要处理什么。由于分区器拿到了存活节点数，它可以根据手头的集群选择分区数量，例如每个存活节点一个或两个，或者按工作负载需要来定。然后由分配策略把这些分区映射到节点上：轮询（round-robin）实现均匀分布，固定节点数可将作业限制在 n 个节点上，或者采用最小负载策略，把每个分区放到报告当前负载最低的节点上，从而把工作导向最空闲的节点。由于分配既按容量确定大小，又感知负载，拆分结果匹配的是实际存在的集群，而不是预先固定的数字。

分配在此时就固定下来。它是放置，而不是运行时再平衡，因此提前完成的节点不会从较慢的同伴那里拉取分区，只有节点故障才会移动工作。运行时再平衡是消息代理的拉取语义免费提供的东西之一，而这里放弃了它。

认领工作而不重复执行

一旦分区被写入并分配，每个工作节点就会独立轮询其 PENDING 分区，并在执行前将它们转换为 CLAIMED。这个转换是正确性关键的时刻，是一条带条件保护的语句，驱动每个工作节点的状态转换，目标状态作为参数传入。

-- claim, then later complete or fail: one guarded UPDATE
UPDATE batch_partitions
SET status = ?, -- CLAIMED, then COMPLETED or FAILED
    last_updated = CURRENT_TIMESTAMP
WHERE step_execution_id = ?
  AND assigned_node = ?
  AND status IN ('PENDING', 'CLAIMED');

WHERE 子句就是全部要点。它解释了为什么 CLAIMED 在两边都出现。它只接纳仍为 PENDING 或 CLAIMED 的行，因此一旦某个分区到达终态（COMPLETED 或 FAILED）或被重新分配，任何迟到的写入都无法再移动它。对于认领而言，行一开始是 PENDING。同时，允许 CLAIMED 使重新认领具有幂等性，并允许同一条语句承载后续的 COMPLETED 和 FAILED 写入，这些写入从 CLAIMED 开始。随后有两个属性防止重复执行。第一，每个分区在分区时只分配给一个节点，所以在正常运行时，只有该节点会轮询自己的行。第二，工作节点的写入和主节点的重新分配都受到条件保护。工作节点的写入由 assigned_node 和 PENDING/CLAIMED 状态保护，重新分配则由该状态保护，因为它会在新节点下把行移回 PENDING。因此，如果一个停滞的节点在主节点重新分配它的同时完成了自己的分区，第二次写入将不匹配任何行。如果主节点先执行，完成操作会失败，因为 assigned_node 不再指向该节点。如果节点先执行，重新分配会失败，因为状态已经是终态。无论哪种方式，只有一次写入会生效。

这不是一个带有 SELECT ... FOR UPDATE SKIP LOCKED 语句的竞争性拉取队列。分区是预先分配的，因此受条件保护的更新支配的是生命周期，而不是派发。消息代理通过队列提供的一分区一节点保证，在这里来自行级状态转换。

从节点故障中恢复

[LOADING...]

图 2. 节点生命周期与两阶段故障转移。（来源：作者创建。）

节点会故障。这里的存活状态是直接跟踪的，而不是从消息代理确认中推断出来的。每个节点按间隔更新心跳时间戳，使用数据库时钟而不是自身时钟。过期检查与同一个时钟比较，因此存活状态不受机器自身时钟之间偏差的影响。检测和移除是去中心化的。每个节点都会运行清理扫描，因此没有哪个单一监控器死亡就会让集群失明。不过，重新分配被移除节点的分区，是由每个受影响作业自己的主节点在先前描述的监控循环中完成的。恢复是一个两阶段过程，如图 2 所示。在第一阶段，心跳早于不可达阈值的节点（默认十秒）会从活动翻转为不可达。在第二阶段，仍然过期超过更长清理阈值（三十秒）的节点会被移除。其未完成分区按契约解决。可转移分区变为可重新分配，不可转移分区被标记为 FAILED，因此作业会干净地失败，而不是冒险第二次运行。两个阈值之间的间隔是宽限期窗口，这是有意设置的。短暂的垃圾回收暂停或网络抖动会把节点标记为不可达，但不会拿走它的工作，它会在下一次心跳时恢复。只有持续缺席才会升级为移除。重新分配复用与认领相同的比较并设置纪律，因此在此期间已经到达 COMPLETED 的分区永远不会被复活。

最佳实践与陷阱：让数据库协调达到生产级

将可转移性视为契约，而不是提示

最具后果的设计决策是：可转移性是声明出来的，而不是推断出来的。每个分区携带一位元数据。如果运行该分区的节点死亡，这项工作是否可以安全地在别处重跑？分区器实现会把每个分区标记为可转移或不可转移，框架会严格遵循这一标记，而不是自行决定。（在参考实现中，这是分区器重写的一个方法。）可转移分区断言工作是幂等的并且可以安全重跑，即至少一次保证。节点丢失时，分区会被重新分配，并可能运行两次。不可转移分区断言相反，即至多一次。节点丢失时，分区会被标记为失败而不是迁移，并且永远不会在别处运行。对数值范围求和是可转移的，但向外部账本追加记录或调用一次性支付 API 则不是。这个选择是一份正确性契约，因此要按分区决定工作是否幂等。

依赖防护机制，但不要只信任它

一个丢失自身心跳的节点会停止认领工作，并在运行已认领分区之前中止，把它留给主节点重新分配。这种防护缩小了两个节点同时运行同一个可转移分区的窗口，但它是尽力而为的。一个被硬暂停的 JVM 可能醒来并完成已经被重新分配的工作，因此可转移工作必须真正幂等。系统永远不会放宽的一点是：没有分区会被静默丢失。分区要么完成，要么作业失败。

将重启设计为幂等

工作节点故障是自动处理的，但主节点故障更难处理。当某个存活节点发现一个仍在进行中的协调行，而它的主节点已经离开时，它会原子地认领这个搁浅的作业并将其标记为失败，因此作业可以干净地重启，而不是永远卡住。恢复方式是失败并重启，而不是原地恢复，因此即使主节点在每个分区都已完成但尚未最终完成作业时死亡，作业也可以重启。因此，重启必须是幂等的。这个要求还把至多一次保证绑定到单次执行上。丢失主节点会强制重启，而其工作节点可能仍在完成工作，因此必须经受住重启的不可转移工作，需要在重启时执行对账步骤。自动接管，即不重启就恢复作业，已在路线图中。当两个节点可能同时充当主节点时，如何让接管安全，是尚未解决的工作。

像对待关键依赖一样运行数据库

一旦数据库成为协调平面，它就是整个集群的硬依赖，因此要以适合该工作负载的方式进行复制、备份和故障转移。这个设计的回报是，所有协调状态都是普通 SQL。你可以用一条针对 BATCH_PARTITIONS 表的 SELECT 语句诊断卡住的运行，而不是去查消息代理控制台。为热点轮询和完成扫描查询建立索引，使它们永远不会退化为表扫描。

知道上限在哪里

该模式大致面向二到二十个节点，上限是数据库争用，而不是固定的节点数。心跳、轮询和状态更新都是数据库流量。在中小规模集群上，这种流量不大，但它会随节点数量和作业并发度增长，直到与批处理工作负载竞争同一个连接池和 I/O。要在自己的负载下验证这个范围。

权衡：何时使用数据库协调，何时选择消息代理

数据库协调并不是基于消息代理的分区的通用替代品。当数据库已经存在且管理良好、当运维简单性和成本比原始规模更重要、当添加消息中间件不切实际（受限环境、有限的运维能力、跨云可移植性）以及当集群处于中小规模范围时，选择它。当集群增长到远远超过该范围、达到几十个节点时，当需要亚秒级协调延迟时，当批处理平台已经与其他工作负载共享成熟流式骨干时，或者当工作天然映射到消息代理的投递和顺序语义时，选择消息代理。共享数据库也会成为你必须以高标准管理的依赖。如果这种做法不现实，消息代理的关注点分离可能更合适。

把数据库协调方法与相邻方案放在一起比较会有所帮助。Spring Batch 自身的远程分区通过消息代理派发，不内置对有多少工作节点存活的感知，也不会在工作节点收到分区后死亡时自动重新分配，而这些正是本模式弥补的缺口。基于 Spring Cloud Task 部署器的分区处理器同样不依赖消息代理，但它通过 Kubernetes 之类的平台为每个分区供应一个工作节点，而本模式协调的是一组常驻的对等 JVM 节点，由它们轮询工作。Quartz、db-scheduler 和 JobRunr 等数据库协调调度器共享同一种直觉：通过数据库协调，并在节点丢失时故障转移，但它们面向调度或通用后台作业，而不是 Spring Batch 分区步骤执行。

结论

从分布式 Spring Batch 中移除消息代理，并不是删除一项需求，而是把它移入 SQL。消息代理隐式回答的同样四个问题——确定规模、协调、单一所有权和节点故障——都得到了显式回答：用存活节点数输入分区器、每个作业一个主节点且没有常驻协调器、比较并设置的生命周期转换，以及两阶段心跳故障转移。回报是横向扩展和自动恢复，而且技术栈中除了 Spring Batch 原本就需要的数据库之外，没有任何额外组件。数据库已经在那里，并被用于这种常见场景：消息代理只会增加活动部件。数据库确实有真实的上限。知道这个上限在哪里最有用。要点是，对于许多规模从未要求使用消息代理的团队来说，把关系型数据库用作协调平面，足以运行正确、容错的分布式批处理。

作者说明

这里描述的设计来自 spring-batch-db-cluster-partitioning，这是作者构建并维护的开源框架，经过同行评审并发表在《Journal of Open Source Software》上。它被用作完整示例，因为它是作者完全了解的实现；无论读者采用它还是构建自己的实现，这些模式都可以迁移。完整的多种数据库 schema、代码以及可运行的多节点示例可在开源仓库中获取。

关于作者

[LOADING...]

Janardhan Reddy Chejarla

Janardhan Reddy Chejarla 是一名首席软件工程师，拥有十七年经验，专长于分布式系统、云原生平台和可扩展数据管道，并有为金融机构设计关键任务架构的背景。他是电子与电信工程师学会（IETE）的会士，也是开源 spring-batch-db-cluster-partitioning 框架的创建者，该框架经过同行评审并发表在《Journal of Open Source Software》上；他在其中撰写关于生产系统中容错、成本效率和运维简单性的内容。