JDBC 消息存储 (JDBC Message Store) _ Spring Integration6.4.0-SNAPSHOT中文文档(官方文档中文翻译)|Spring 教程 —

此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 Spring Integration 6.3.4！spring-doc.cn

Spring 集成提供了两种特定于 JDBC 的消息存储实现。适合与聚合商和索赔检查模式一起使用。该实现专门为消息通道提供了更具针对性和可扩展性的实现。JdbcMessageStoreJdbcChannelMessageStorespring-doc.cn

请注意，您可以使用 a 来支持消息通道，它是为此目的而优化的。JdbcMessageStoreJdbcChannelMessageStorespring-doc.cn

从版本 5.0.11、5.1.2 开始，这些的索引已得到优化。如果您在此类存储中有大型消息组，则可能需要更改索引。此外，索引 for 被注释掉，因为除非您使用 JDBC 支持的此类通道，否则不需要它。JdbcChannelMessageStorePriorityChannel

使用时，必须添加优先级通道索引，因为它包含在查询的 hint 中。OracleChannelMessageStoreQueryProvider

初始化数据库

在开始使用 JDBC 消息存储组件之前，您应该为目标数据库配置适当的对象。spring-doc.cn

Spring 集成附带了一些可用于初始化数据库的示例脚本。在 JAR 文件中，您可以在包中找到脚本。它为一系列常见数据库平台提供了示例 create 和 example drop 脚本。使用这些脚本的常见方法是在 Spring JDBC 数据源初始化器中引用它们。请注意，这些脚本作为示例以及所需表和列名称的规范提供。您可能会发现需要增强它们以供生产环境使用（例如，通过添加索引声明）。spring-integration-jdbcorg.springframework.integration.jdbcspring-doc.cn

从版本 6.2 开始，，，，和在方法中对它们各自的表实施并执行 'SELECT COUNT' 查询，以确保目标数据库中存在所需的表（根据提供的前缀）。如果所需的表不存在，则应用程序上下文无法启动。可以通过以下方式禁用该检查。JdbcMessageStoreJdbcChannelMessageStoreJdbcMetadataStoreDefaultLockRepositorySmartLifecyclestart()setCheckDatabaseOnStart(false)spring-doc.cn

通用 JDBC 消息存储

JDBC 模块提供了 Spring 集成的实现（在声明检查模式中很重要）和（在有状态模式（如聚合器）中很重要）由数据库支持。这两个接口都是由实现的，并且支持在 XML 中配置存储实例，如下例所示：MessageStoreMessageGroupStoreJdbcMessageStorespring-doc.cn

<int-jdbc:message-store id="messageStore" data-source="dataSource"/>

您可以指定 a 而不是 .JdbcTemplateDataSourcespring-doc.cn

以下示例显示了一些其他可选属性：spring-doc.cn

<int-jdbc:message-store id="messageStore" data-source="dataSource" table-prefix="MY_INT_"/>

在前面的示例中，我们为存储生成的查询中的表名称指定了前缀。表名前缀默认为 .INT_spring-doc.cn

后备消息通道

如果您打算使用 JDBC 备份消息通道，我们建议使用实现。它只能与 Message Channel 结合使用。JdbcChannelMessageStorespring-doc.cn

支持的数据库

使用特定于数据库的 SQL 查询从数据库中检索消息。因此，您必须在 . 这将为您指定的特定数据库提供 SQL 查询。 Spring 集成为以下关系数据库提供支持：JdbcChannelMessageStoreChannelMessageStoreQueryProviderJdbcChannelMessageStorechannelMessageStoreQueryProviderspring-doc.cn

PostgreSQL 数据库spring-doc.cn
HSQLDB 数据库spring-doc.cn
MySQL （MySQL的spring-doc.cn
神谕spring-doc.cn
德比spring-doc.cn
H2 系列spring-doc.cn
SqlServer 服务器spring-doc.cn
Sybasespring-doc.cn
DB2spring-doc.cn

如果您的数据库未列出，您可以实现该接口并提供您自己的自定义查询。ChannelMessageStoreQueryProviderspring-doc.cn

版本 4.0 在表中添加了该列，以确保先进先出（FIFO）排队，即使消息以相同的毫秒数存储也是如此。MESSAGE_SEQUENCEspring-doc.cn

从版本 6.2 开始，公开了一个标志，其中 poll 的返回值及其查询现在基于单个语句。如果仅支持单个 poll 语句，则会使用选项进行咨询并跳过单独的语句。ChannelMessageStoreQueryProviderisSingleStatementForPollPostgresChannelMessageStoreQueryProvidertrueDELETE…RETURNINGJdbcChannelMessageStoreisSingleStatementForPollDELETEspring-doc.cn

自定义消息插入

从版本 5.0 开始，通过重载该类，您可以在 . 您可以使用它来设置不同的列或更改表结构或序列化策略。例如，您可以将其结构存储为 JSON 字符串，而不是默认序列化。ChannelMessageStorePreparedStatementSetterJdbcChannelMessageStorebyte[]spring-doc.cn

以下示例使用默认实现来存储公共列，并覆盖将消息有效负载存储为的行为：setValuesvarcharspring-doc.cn

public class JsonPreparedStatementSetter extends ChannelMessageStorePreparedStatementSetter {

    @Override
    public void setValues(PreparedStatement preparedStatement, Message<?> requestMessage,
        Object groupId, String region, 	boolean priorityEnabled) throws SQLException {
        // Populate common columns
        super.setValues(preparedStatement, requestMessage, groupId, region, priorityEnabled);
        // Store message payload as varchar
        preparedStatement.setString(6, requestMessage.getPayload().toString());
    }
}

通常，我们不建议使用关系数据库进行排队。相反，如果可能，请考虑改用 JMS 或 AMQP 支持的通道。有关进一步参考，请参阅以下资源：spring-doc.cn

数据库作为队列反模式。spring-doc.cn

如果您仍计划将数据库用作队列，请考虑使用 PostgreSQL 及其通知机制，这将在后续部分中介绍。spring-doc.cn

并发轮询

轮询消息通道时，您可以选择配置与引用关联的。PollerTaskExecutorspring-doc.cn

但请记住，如果您使用 JDBC 支持的消息通道，并且您计划轮询该通道，从而使用多个线程轮询消息存储事务性消息存储，则应确保使用支持 Multiversion Concurrency Control （MVCC）的关系数据库。否则，锁定可能是一个问题，并且在使用多个线程时，性能可能无法按预期实现。例如，Apache Derby 在这方面是有问题的。spring-doc.cn

为了实现更好的 JDBC 队列吞吐量并避免不同线程可能从队列中轮询相同的线程时出现问题，在使用不支持 MVCC 的数据库时，将 的属性设置为 to 非常重要。以下示例显示了如何执行此操作：MessageusingIdCacheJdbcChannelMessageStoretruespring-doc.cn

<bean id="queryProvider"
    class="o.s.i.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider"/>

<int:transaction-synchronization-factory id="syncFactory">
    <int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
    <int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/>
</int:transaction-synchronization-factory>

<task:executor id="pool" pool-size="10"
    queue-capacity="10" rejection-policy="CALLER_RUNS" />

<bean id="store" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
    <property name="region" value="TX_TIMEOUT"/>
    <property name="usingIdCache" value="true"/>
</bean>

<int:channel id="inputChannel">
    <int:queue message-store="store"/>
</int:channel>

<int:bridge input-channel="inputChannel" output-channel="outputChannel">
    <int:poller fixed-delay="500" receive-timeout="500"
        max-messages-per-poll="1" task-executor="pool">
        <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
        isolation="READ_COMMITTED" transaction-manager="transactionManager" />
    </int:poller>
</int:bridge>

<int:channel id="outputChannel" />

优先通道

从版本 4.0 开始，实现并提供选项，使其用作实例的参考。为此，该表有一个列来存储消息标头的值。此外，新列让我们实现了强大的先进先出（FIFO）轮询机制，即使在同一毫秒内以相同的优先级存储了多条消息。使用从数据库中轮询（选择）消息。JdbcChannelMessageStorePriorityCapableChannelMessageStorepriorityEnabledmessage-storepriority-queueINT_CHANNEL_MESSAGEMESSAGE_PRIORITYPRIORITYMESSAGE_SEQUENCEorder by MESSAGE_PRIORITY DESC NULLS LAST, CREATED_DATE, MESSAGE_SEQUENCEspring-doc.cn

我们不建议对优先级和非优先级队列通道使用相同的 bean，因为该选项适用于整个存储，并且不会为队列通道保留适当的 FIFO 队列语义。但是，同一表（甚至）可用于这两种类型。要配置该场景，可以从一个消息存储 Bean 扩展另一个消息存储 Bean，如下例所示：JdbcChannelMessageStorepriorityEnabledINT_CHANNEL_MESSAGEregionJdbcChannelMessageStore

<bean id="channelStore" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
</bean>

<int:channel id="queueChannel">
    <int:queue message-store="channelStore"/>
</int:channel>

<bean id="priorityStore" parent="channelStore">
    <property name="priorityEnabled" value="true"/>
</bean>

<int:channel id="priorityChannel">
    <int:priority-queue message-store="priorityStore"/>
</int:channel>

通常，我们不建议使用关系数据库进行排队。相反，如果可能，请考虑改用 JMS 或 AMQP 支持的通道。有关进一步参考，请参阅以下资源：spring-doc.cn

数据库作为队列反模式。spring-doc.cn

如果您仍计划将数据库用作队列，请考虑使用 PostgreSQL 及其通知机制，这将在后续部分中介绍。spring-doc.cn

<bean id="queryProvider"
    class="o.s.i.jdbc.store.channel.PostgresChannelMessageStoreQueryProvider"/>

<int:transaction-synchronization-factory id="syncFactory">
    <int:after-commit expression="@store.removeFromIdCache(headers.id.toString())" />
    <int:after-rollback expression="@store.removeFromIdCache(headers.id.toString())"/>
</int:transaction-synchronization-factory>

<task:executor id="pool" pool-size="10"
    queue-capacity="10" rejection-policy="CALLER_RUNS" />

<bean id="store" class="o.s.i.jdbc.store.JdbcChannelMessageStore">
    <property name="dataSource" ref="dataSource"/>
    <property name="channelMessageStoreQueryProvider" ref="queryProvider"/>
    <property name="region" value="TX_TIMEOUT"/>
    <property name="usingIdCache" value="true"/>
</bean>

<int:channel id="inputChannel">
    <int:queue message-store="store"/>
</int:channel>

<int:bridge input-channel="inputChannel" output-channel="outputChannel">
    <int:poller fixed-delay="500" receive-timeout="500"
        max-messages-per-poll="1" task-executor="pool">
        <int:transactional propagation="REQUIRED" synchronization-factory="syncFactory"
        isolation="READ_COMMITTED" transaction-manager="transactionManager" />
    </int:poller>
</int:bridge>

<int:channel id="outputChannel" />

对消息存储进行分区

通常将用作同一应用程序中的一组应用程序或节点的全局存储。为了提供一些防止名称冲突的保护并控制数据库元数据配置，消息存储允许以两种方式对表进行分区。一种方法是通过更改前缀（如前所述）来使用单独的表名。另一种方法是指定一个名称，用于在单个表中对数据进行分区。第二种方法的一个重要用例是当 Management 支持 Spring Integration Message Channel 的持久队列时。持久通道的消息数据在存储区中的通道名称上键入。因此，如果通道名称不是全局唯一的，则通道可以选取并非适用于它们的数据。为避免这种危险，您可以使用消息存储将具有相同逻辑名称的不同物理通道的数据分开。JdbcMessageStoreregionMessageStoreregionspring-doc.cn

PostgreSQL：接收推送通知

PostgreSQL 提供了一个侦听和通知框架，用于在数据库表操作时接收推送通知。 Spring 集成利用这种机制（从版本 6.0 开始）来允许在将新消息添加到 . 使用此功能时，必须定义一个数据库触发器，该触发器可以作为 Spring Integration 的 JDBC 模块中包含的文件的注释的一部分找到。JdbcChannelMessageStoreschema-postgresql.sqlspring-doc.cn

推送通知是通过类接收的，该类允许其订阅者在到达任何给定和的新消息时接收回调。即使消息附加到不同的 JVM 上，但附加到同一个数据库，也会收到这些通知。该实现使用协定从 store 中提取消息，作为对来自上述通知的通知的反应。PostgresChannelMessageTableSubscriberregiongroupIdPostgresSubscribableChannelPostgresChannelMessageTableSubscriber.SubscriptionPostgresChannelMessageTableSubscriberspring-doc.cn

例如，可以按如下方式接收的推送通知：some groupspring-doc.cn

@Bean
public JdbcChannelMessageStore messageStore(DataSource dataSource) {
    JdbcChannelMessageStore messageStore = new JdbcChannelMessageStore(dataSource);
    messageStore.setChannelMessageStoreQueryProvider(new PostgresChannelMessageStoreQueryProvider());
    return messageStore;
}

@Bean
public PostgresChannelMessageTableSubscriber subscriber(
      @Value("${spring.datasource.url}") String url,
      @Value("${spring.datasource.username}") String username,
      @Value("${spring.datasource.password}") String password) {
    return new PostgresChannelMessageTableSubscriber(() ->
        DriverManager.getConnection(url, username, password).unwrap(PgConnection.class));
}

@Bean
public PostgresSubscribableChannel channel(
    PostgresChannelMessageTableSubscriber subscriber,
    JdbcChannelMessageStore messageStore) {
  return new PostgresSubscribableChannel(messageStore, "some group", subscriber);
}

事务支持spring-doc.cn

从版本 6.0.5 开始，在 a 上指定 a 将通知事务中的订阅者。订户中的异常将导致事务回滚，并将消息放回消息存储中。默认情况下，事务支持未激活。PlatformTransactionManagerPostgresSubscribableChannelspring-doc.cn

重试spring-doc.cn

从版本 6.0.5 开始，可以通过向 . 默认情况下，不执行重试。RetryTemplatePostgresSubscribableChannelspring-doc.cn

任何活动在其活动生命周期内都占用一个独占 JDBC。因此，此连接不源自 pooling 非常重要。此类连接池通常希望发出的连接在预定义的超时窗口内关闭。PostgresChannelMessageTableSubscriberConnectionDataSourcespring-doc.cn

对于独占连接的这种需求，还建议 JVM 只运行一个可用于注册任意数量的订阅的单个连接。PostgresChannelMessageTableSubscriberspring-doc.cn