StackSaga PostgreSQL 响应式支持 (Reactive Support)

概述 (Overview)

stacksaga-pg-reactive-support 是 数据库支持模块 的 PostgreSQL 响应式（非阻塞 R2DBC） 实现。 它为 StackSaga 引擎提供了将 PostgreSQL 数据库作为事件存储所需的一切底层设施，同时也暴露了内置端点，供 StackSaga TraceWindow 链路追踪控制台 (Dashboard) 控制台查询追踪明细。

该 Maven 模块与工件名称统一采用 pg 简写形式（stacksaga-pg-reactive-support）而非 postgresql，尽管在本文档和数据库描述中均统称为 PostgreSQL。
请先阅读 StackSaga 事件存储 (Event Store)，以便更好地理解数据库支持模块的职责与核心架构。

添加 stacksaga-pg-reactive-support 依赖

请按以下方式将该依赖添加到您现有的编排器 (Orchestrator) 应用程序中：

在 pom.xml 中添加 stacksaga-pg-reactive-support 依赖
<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.stacksaga</groupId>
            <artifactId>stacksaga-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependency>
    <groupId>org.stacksaga</groupId>
    <artifactId>stacksaga-pg-reactive-support</artifactId>
</dependency>
推荐使用 StackSaga Initializer (初始化器) 获取项目所需的 StackSaga 依赖配置代码段，以确保拥有匹配正确的依赖版本和配置。
PostgreSQL R2DBC 驱动 (org.postgresql:r2dbc-postgresql) 由 stacksaga-pg-reactive-support 自动传递引入，支持 PostgreSQL 12 及以上版本，并在 PostgreSQL 16 上完成了完整的集成测试验证。

配置属性 (Configuration Properties)

下方的数据库连接、连接池、事务恢复以及恢复节点数据流配置属性，由通用的方言无关配置类进行统一绑定。这些配置在所有 stacksaga-*-reactive-support SQL 模块之间完全共享——包括 MySQL、Oracle、PostgreSQL，以及未来加入的任何其他 SQL 方言：

  • stacksaga.datasource.* 用于为事件存储配置专用的响应式 R2DBC 连接工厂与连接池。

  • stacksaga.sql.transaction.recovery.* 用于按 Saga 领域实体（具备全局默认值）配置事务重试延迟、崩溃恢复延迟以及事务生命周期。

  • stacksaga.transaction.recovery-node.* 用于在编排器实例作为 重试节点 (Retry-Node) 运行时，调优响应式流式处理流水线。该流水线仅在通过 stacksaga.coordinator.connector.enabled=true（默认值为 false）启用时才会激活。

属性名称 (Property Name) 默认值 (Default Value) 类型 (Type) 描述说明 (Description)

stacksaga.datasource.enabled

true

boolean

全局启用或禁用该 starter。默认为 true，引入依赖后自动激活。

以下 stacksaga.datasource.r2dbc. 属性用于配置响应式事件存储所使用的 R2DBC 数据库连接（针对事件存储的读写操作）。*

stacksaga.datasource.r2dbc.url

-

String

事件存储使用的 R2DBC 连接 URL，例如 r2dbc:mysql://host:3306/database 或 r2dbc:oracle://host:1521/service_name。如果配置了此项，其优先级高于独立的 host/port/database 配置。

stacksaga.datasource.r2dbc.host

localhost

String

响应式事件存储连接的目标数据库服务器主机地址。

stacksaga.datasource.r2dbc.port

3306

int

响应式事件存储连接的目标数据库端口。非 MySQL 方言时需覆盖（例如 Oracle 监听端口为 1521），或直接提供 stacksaga.datasource.r2dbc.url。

stacksaga.datasource.r2dbc.database

-

String

事件存储通过 R2DBC 访问的目标数据库名称或服务名。

stacksaga.datasource.r2dbc.username

-

String

事件存储 R2DBC 连接的用户名。

stacksaga.datasource.r2dbc.password

-

String

事件存储 R2DBC 连接的密码。

stacksaga.datasource.r2dbc.connect-timeout

10s

Duration

与数据库服务器建立事件存储连接的最大等待超时时间。

stacksaga.datasource.r2dbc.options

-

Map<String,String>

事件存储专用的 R2DBC 驱动附加连接选项（键值对形式）。

stacksaga.datasource.r2dbc.pool.initial-size

5

int

事件存储连接池启动时初始创建的连接数。

stacksaga.datasource.r2dbc.pool.max-size

20

int

事件存储连接池允许的最大连接数。

stacksaga.datasource.r2dbc.pool.max-idle-time

30m

Duration

连接在池中保持空闲的最长时间，超时将被自动关闭回收。

stacksaga.datasource.r2dbc.pool.max-life-time

1h

Duration

连接在池中允许存活的最长时间，超出此寿命的连接将被关闭并用新连接替代。

stacksaga.datasource.r2dbc.pool.max-acquire-time

10s

Duration

从连接池中获取可用连接的最大超时等待时间。

stacksaga.datasource.r2dbc.pool.validation-query

SELECT 1

String

从池中借出连接前用于探活校验的 SQL 查询语句。

以下 stacksaga.sql.transaction.recovery. 属性用于管控事件存储的事务重试、崩溃恢复与生命周期，支持全局默认设置及基于 Saga 领域实体的局部覆盖。*

stacksaga.sql.transaction.recovery.default.retry.delay

1m

Duration

事务等待下次被暴露用于重试的默认延迟时间，避免短时间内过于频繁的无意义重试。

stacksaga.sql.transaction.recovery.default.restore.delay

5h

Duration

事务在无状态更新的情况下被认定为已崩溃（例如 Pod 异常中断崩溃）并被恢复用于重试的默认延迟时间。

stacksaga.sql.transaction.recovery.default.lifetime

24h

Duration

事务在事件存储中允许存活的最大时间。超过此期限后事务将被隔离归档，即便仍处于未完成状态也绝不会再被暴露进行恢复。

stacksaga.sql.transaction.recovery.<domain-name>.retry.delay

-

Duration

针对特定领域实体的重试延迟覆盖配置（例如 order-domain）。未配置时自动回退使用 default.retry.delay。

stacksaga.sql.transaction.recovery.<domain-name>.restore.delay

-

Duration

针对特定领域实体的崩溃恢复延迟覆盖配置（例如 order-domain）。未配置时自动回退使用 default.restore.delay。

stacksaga.sql.transaction.recovery.<domain-name>.lifetime

-

Duration

针对特定领域实体的事务生命周期覆盖配置（例如 order-domain）。未配置时自动回退使用 default.lifetime。

以下 stacksaga.transaction.recovery-node. 属性用于调优响应式批量拉取与流式处理流水线。仅当编排器实例通过 stacksaga.coordinator.connector.enabled=true 声明为 重试节点 (Retry-Node) 时生效。*

stacksaga.transaction.recovery-node.batch-size

100

int

控制单次从数据库批量拉取到内存缓冲区的事务数量 (Reactor limitRate 高水位)。较高值可减少数据库网络往返，但会占用更多内存。

stacksaga.transaction.recovery-node.refill-threshold

75

int

控制框架何时触发下一次数据库批量拉取以补充缓冲区 (Reactor limitRate 低水位)。当缓冲数量降至此阈值时触发后台拉取，确保流水线永远不会饥饿停滞。必须小于 batch-size（推荐约为 75%）。

stacksaga.transaction.recovery-node.concurrency

8

int

控制在恢复流水线中并发处理的事务数量 (Reactor flatMap 并发度)。建议保持在连接池大小的 50–70% 之间，以避免连接池耗尽。

stacksaga.transaction.recovery-node.pre-fetch

4

int

控制在处理开始前预先拉取并缓存在内存中的事务数量 (Reactor flatMap 预取度)，提高吞吐量并降低批次间的调度延迟。

事务恢复配置示例 (Recovery Configuration Examples)

StackSaga 在 stacksaga.sql.transaction.recovery.default.* 下提供了开箱即用的默认值，并支持基于领域实体名称（如 order-domain）的局部覆盖；当特定领域未显式配置某项属性时，会自动回退使用默认值：

示例 application.properties
# 内置全局默认配置（适用于所有 Saga 领域）
stacksaga.sql.transaction.recovery.default.retry.delay=1m
stacksaga.sql.transaction.recovery.default.restore.delay=5h
stacksaga.sql.transaction.recovery.default.lifetime=24h

# 针对特定领域的局部覆盖（以 "order-domain" 为例）
stacksaga.sql.transaction.recovery.order-domain.retry.delay=10s
stacksaga.sql.transaction.recovery.order-domain.restore.delay=1h
stacksaga.sql.transaction.recovery.order-domain.lifetime=12h
示例 application.yml
stacksaga:
  sql:
    transaction:
      recovery:
        default:
          retry:
            delay: 1m
          restore:
            delay: 5h
          lifetime: 24h
        order-domain:
          retry:
            delay: 10s
          restore:
            delay: 1h
          lifetime: 12h

重试节点流水线配置示例 (Recovery-Node Pipeline Configuration Examples)

默认情况下，所有编排器实例均作为标准节点 (Standard-Nodes) 运行（stacksaga.coordinator.connector.enabled=false），不承担恢复任务。要将特定实例转变为专属的 重试节点 (Retry-Node)，请引入 stacksaga-ring-coordinator-connector 依赖并将 stacksaga.coordinator.connector.enabled=true。随后即可通过 stacksaga.transaction.recovery-node.* 调优响应式批处理与并发流水线：

示例 application.properties (重试节点实例)
# 在重试环架构中将此实例激活为重试节点 (Retry-Node)
stacksaga.coordinator.connector.enabled=true

# 重试节点流式处理与并发性能调优
stacksaga.transaction.recovery-node.batch-size=100
stacksaga.transaction.recovery-node.refill-threshold=75
stacksaga.transaction.recovery-node.concurrency=8
stacksaga.transaction.recovery-node.pre-fetch=4
示例 application.yml (重试节点实例)
stacksaga:
  coordinator:
    connector:
      enabled: true # 开启重试节点模式
  transaction:
    recovery-node:
      batch-size: 100
      refill-threshold: 75
      concurrency: 8
      pre-fetch: 4

Schema 结构管理与版本校验 (Schema Management & Version Verification)

与许多在应用启动时内置数据迁移框架（如 Liquibase 或 Flyway）直接变更数据库表结构的传统应用不同，StackSaga 严格推行数据库管理权限与应用运行时执行权限的清晰分离原则：

应用运行时权限边界 (Application Runtime Privilege Boundary)： 通过 R2DBC 连接数据库的应用运行时账号，仅需对事件存储表具备基础数据访问权限 (SELECT, INSERT, UPDATE, DELETE)。应用程序在启动期间绝不执行任何 DDL，也绝不擅自修改数据库元数据对象。

在应用发布部署之前，应由 DBA 或 CI/CD 自动化流水线使用具备 DDL 权限的高权限账号预先执行框架提供的 schema.sql 脚本。

启动前预检 Schema 版本自动校验 (Pre-Flight Schema Version Verification)

为了杜绝因数据库未初始化或版本不匹配而导致的字段缺失异常或静默数据不一致，StackSaga Starter 在应用启动时会执行自动化的、非阻塞预检 Schema 版本校验：

  1. 当 Spring 单例 Bean 完成实例化且连接池预热完成后，StackSaga 会查询 stacksaga_schema_version 元数据表。

  2. 校验表中是否已准确记录当前框架所需的 Schema 目标版本号（例如 1.0.0）。

  3. 如果该版本表不存在，或者缺少对应的版本记录行，应用程序将拒绝启动并立即快速失败 (Fail-Fast)，抛出包含明确修复指引的 SchemaVersionMismatchException。

获取 Schema 初始化 DDL 脚本

开发人员与 DBA 可以通过多种方式轻松获取 DDL 脚本：

  • 从依赖 JAR 包中提取：

    标准的官方 DDL 脚本打包在依赖 JAR 的以下路径中：

    db/stacksaga/pg/schema.sql

  • 通过 Java 代码编程式获取（适用于本地开发、单元测试与 Testcontainers）：

    在集成测试（如 Testcontainers 容器化测试启动）、自动化迁移工具或自定义初始化脚本中，您可以直接将原始 DDL 脚本以 String 形式提取出来：

    String ddl = StackSagaPgAutoConfiguration.getSchemaScript();

  • IDE 一键运行打印：

    如需快速查看或复制脚本，只需在 IDE（IntelliJ IDEA、Eclipse、VS Code）中定位到 StackSagaPgAutoConfiguration 类并运行其 main() 方法。该类会将完整的 DDL 脚本直接打印到控制台，您可以轻松复制并粘贴到数据库客户端（如 DBeaver、MySQL Workbench、pgAdmin 或终端 CLI）中执行。

累积性与幂等性升级机制 (Cumulative and Idempotent Upgrades)

schema.sql 的每次发布均严格遵循累积性 (Cumulative) 与幂等性 (Idempotent) 设计：

  • 全新环境初始化：执行 schema.sql 将自动创建所有必需的表结构、索引与约束，并在 stacksaga_schema_version 表中盖上版本戳。

  • 存量数据库原地升级：重新执行 schema.sql 是完全安全且非破坏性的——现有业务数据和表结构完整保留，仅安全增补新的 Schema 对象，并写入最新的版本记录行。

数据表分区维护 (Table Partition Maintenance)

PostgreSQL 事件存储表（es_transaction 与 es_transaction_execution_tryout）支持基于 created_at 的声明式范围分区 (Declarative Range Partitioning)。

要自动创建未来日期的新分区，而无需将 DDL 执行权限耦合到各业务微服务实例中，请参阅专门的 StackSaga SQL 数据库分区支持 模块。您可以将其独立部署一次，集中式统一管理所有 PostgreSQL 和 MySQL 数据库的分区生命周期。