Spring Integration对Redis的支持

Spring Integration 2.1引入了对Redis​的支持：“一个开源的高级键值存储”。 这种支持以基于 Redis 以及发布-订阅消息传递适配器的形式出现，Redis 通过其 PUBLISH、SUBSCRIBE和 UNSUBSCRIBE​命令支持这些适配器。​​MessageStore​​

您需要将此依赖项包含在项目中：

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-redis</artifactId>
    <version>6.0.0</version>
</dependency>

您还需要包含 Redis 客户端依赖项，例如 Lettuce。

要下载、安装和运行 Redis，请参阅 Redis 文档。

连接到 Redis

要开始与 Redis 交互，您首先需要连接到它。 Spring Integration 使用另一个 Spring 项目 Spring Data Redis 提供的支持，该项目提供了典型的 Spring 构造：和 。 这些抽象简化了与多个 Redis 客户端 Java API 的集成。 目前，Spring Data Redis 支持 Jedis 和 Lettuce。​​ConnectionFactory​​​​Template​​

用​​RedisConnectionFactory​​

要连接到 Redis，您可以使用接口的实现之一。 以下清单显示了接口定义：​​RedisConnectionFactory​​

public interface RedisConnectionFactory extends PersistenceExceptionTranslator {

    /**
     * Provides a suitable connection for interacting with Redis.
     * @return connection for interacting with Redis.
     */
    RedisConnection getConnection();
}

以下示例显示了如何在 Java 中创建：​​LettuceConnectionFactory​​

LettuceConnectionFactory cf = new LettuceConnectionFactory();
cf.afterPropertiesSet();

下面的示例展示了如何在 Spring 的 XML 配置中创建：​​LettuceConnectionFactory​​

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379" />
</bean>

的实现提供了一组属性，例如端口和主机，您可以根据需要设置这些属性。 拥有 的实例后，可以创建 的实例并将其注入 .​​RedisConnectionFactory​​​​RedisConnectionFactory​​​​RedisTemplate​​​​RedisConnectionFactory​​

用​​RedisTemplate​​

与 Spring 中的其他模板类（例如 和 ）一样，它是一个简化 Redis 数据访问代码的辅助类。 有关及其变体的更多信息（例如），请参阅 Spring Data Redis 文档。​​JdbcTemplate​​​​JmsTemplate​​​​RedisTemplate​​​​RedisTemplate​​​​StringRedisTemplate​​

以下示例演示如何在 Java 中创建 的实例：​​RedisTemplate​​

RedisTemplate rt = new RedisTemplate<String, Object>();
rt.setConnectionFactory(redisConnectionFactory);

下面的示例展示了如何在 Spring 的 XML 配置中创建 的实例：​​RedisTemplate​​

<bean id="redisTemplate"
         class="org.springframework.data.redis.core.RedisTemplate">
    <property name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

使用 Redis 发送消息

如简介中所述，Redis 通过其 、 和命令提供对发布-订阅消息传递的支持。 与JMS和AMQP一样，Spring Integration提供了消息通道和适配器，用于通过Redis发送和接收消息。​​PUBLISH​​​​SUBSCRIBE​​​​UNSUBSCRIBE​​

瑞迪斯发布/订阅频道

与 JMS 类似，在某些情况下，生产者和使用者都打算成为同一应用程序的一部分，在同一进程中运行。 可以使用一对入站和出站通道适配器来实现此目的。 但是，与Spring Integration的JMS支持一样，有一种更简单的方法来解决这个用例。 您可以创建发布-订阅通道，如以下示例所示：

<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"/>

A 的行为很像 Spring 集成主命名空间中的普通元素。 它可以由任何终结点的 和属性引用。 不同之处在于，此通道由 Redis 主题名称支持：属性指定的值。 但是，与 JMS 不同的是，本主题不必提前创建，甚至不必由 Redis 自动创建。 在 Redis 中，主题是扮演地址角色的简单值。 生成者和使用者可以使用与其主题名称相同的值进行通信。 对此通道的简单订阅意味着可以在生产和消费终结点之间实现异步发布-订阅消息传递。 但是，与通过在简单的 Spring 集成元素中添加元素创建的异步消息通道不同，消息不会存储在内存中队列中。 相反，这些消息通过 Redis 传递，这使您可以依赖它对持久性和集群的支持以及与其他非 Java 平台的互操作性。​​publish-subscribe-channel​​​​<publish-subscribe-channel/>​​​​input-channel​​​​output-channel​​​​String​​​​topic-name​​​​String​​​​String​​​​<queue/>​​​​<channel/>​​

Redis 入站通道适配器

Redis 入站通道适配器 （） 以与其他入站适配器相同的方式将传入的 Redis 消息调整为 Spring 消息。 它接收特定于平台的消息（在本例中为 Redis），并使用策略将它们转换为 Spring 消息。 以下示例演示如何配置 Redis 入站通道适配器：​​RedisInboundChannelAdapter​​​​MessageConverter​​

<int-redis:inbound-channel-adapter id="redisAdapter"
       topics="thing1, thing2"
       channel="receiveChannel"
       error-channel="testErrorChannel"
       message-converter="testConverter" />

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379" />
</bean>

<bean id="testConverter" class="things.something.SampleMessageConverter" />

前面的示例显示了 Redis 入站通道适配器的简单但完整的配置。 请注意，前面的配置依赖于熟悉的 Spring 范式，即自动发现某些 bean。 在这种情况下，将隐式注入适配器。 可以改用属性显式指定它。​​redisConnectionFactory​​​​connection-factory​​

另请注意，上述配置为适配器注入了自定义 . 该方法类似于 JMS，其中实例用于在 Redis 消息和 Spring 集成消息有效负载之间进行转换。 默认值为 .​​MessageConverter​​​​MessageConverter​​​​SimpleMessageConverter​​

入站适配器可以订阅多个主题名称，因此属性中的值集以逗号分隔。​​topics​​

从 V3.0 开始，除了现有属性之外，入站适配器现在还具有该属性。 此属性包含一组以逗号分隔的 Redis 主题模式。 有关 Redis 发布-订阅的更多信息，请参阅 Redis 发布/订阅。​​topics​​​​topic-patterns​​

入站适配器可以使用 对 Redis 消息正文进行反序列化。 的属性可以设置为空字符串，这将生成属性的值。 在这种情况下，Redis 消息的原始正文作为消息有效负载提供。​​RedisSerializer​​​​serializer​​​​<int-redis:inbound-channel-adapter>​​​​null​​​​RedisSerializer​​​​byte[]​​

从 V5.0 开始，您可以使用 的属性向入站适配器提供实例。 此外，收到的 Spring 集成消息现在具有指示已发布消息来源的标头：主题或模式。 您可以将此下游用于路由逻辑。​​Executor​​​​task-executor​​​​<int-redis:inbound-channel-adapter>​​​​RedisHeaders.MESSAGE_SOURCE​​

Redis 出站通道适配器

Redis 出站通道适配器将传出的 Spring 集成消息调整为 Redis 消息，方式与其他出站适配器相同。 它接收 Spring 集成消息，并使用策略将它们转换为特定于平台的消息（在本例中为 Redis）。 以下示例演示如何配置 Redis 出站通道适配器：​​MessageConverter​​

<int-redis:outbound-channel-adapter id="outboundAdapter"
    channel="sendChannel"
    topic="thing1"
    message-converter="testConverter"/>

<bean id="redisConnectionFactory"
    class="o.s.data.redis.connection.lettuce.LettuceConnectionFactory">
    <property name="port" value="7379"/>
</bean>

<bean id="testConverter" class="things.something.SampleMessageConverter" />

该配置与 Redis 入站通道适配器并行。 适配器隐式注入 ，该 定义为其 Bean 名称。 此示例还包括可选（和自定义）（bean）。​​RedisConnectionFactory​​​​redisConnectionFactory​​​​MessageConverter​​​​testConverter​​

从 Spring Integration 3.0 开始，提供了该属性的替代方法：您可以使用该属性在运行时确定消息的 Redis 主题。 这些属性是互斥的。​​<int-redis:outbound-channel-adapter>​​​​topic​​​​topic-expression​​

Redis 队列入站通道适配器

Spring Integration 3.0引入了一个队列入站通道适配器，用于从Redis列表中“弹出”消息。 默认情况下，它使用“右弹出”，但您可以将其配置为使用“左弹出”。 适配器是消息驱动的。 它使用内部侦听器线程，不使用轮询器。

以下清单显示了 的所有可用属性：​​queue-inbound-channel-adapter​​

<int-redis:queue-inbound-channel-adapter id=""  
                    channel=""  
                    auto-startup=""  
                    phase=""  
                    connection-factory=""  
                    queue=""  
                    error-channel=""  
                    serializer=""  
                    receive-timeout=""  
                    recovery-interval=""  
                    expect-message=""  
                    task-executor=""  
                    right-pop=""/>

Redis 队列出站通道适配器

Spring Integration 3.0引入了一个队列出站通道适配器，用于从Spring Integration消息“推送”到Redis列表。 默认情况下，它使用“左推”，但您可以将其配置为使用“右推”。 以下清单显示了 Redis 的所有可用属性：​​queue-outbound-channel-adapter​​

<int-redis:queue-outbound-channel-adapter id=""  
                    channel=""  
                    connection-factory=""  
                    queue=""  
                    queue-expression=""  
                    serializer=""  
                    extract-payload=""  
                    left-push=""/>

瑞迪斯应用程序事件

从 Spring Integration 3.0 开始，Redis 模块提供了 的实现，而 又是一个 . 封装了 Redis 操作的异常（端点是事件的“源”）。 例如，在捕获操作中的异常后发出这些事件。 例外可以是任何泛型或 . 使用 处理这些事件对于确定后台 Redis 任务的问题和采取管理操作非常有用。​​IntegrationEvent​​​​org.springframework.context.ApplicationEvent​​​​RedisExceptionEvent​​​​<int-redis:queue-inbound-channel-adapter/>​​​​BoundListOperations.rightPop​​​​org.springframework.data.redis.RedisSystemException​​​​org.springframework.data.redis.RedisConnectionFailureException​​​​<int-event:inbound-channel-adapter/>​​

红与红消息存储

如企业集成模式 （EIP） 一书中所述，邮件存储允许您保留消息。 当考虑可靠性时，这在处理具有缓冲消息功能的组件（聚合器、重新排序器等）时非常有用。 在 Spring 集成中，该策略还为声明检查模式提供了基础，EIP 中也有描述。​​MessageStore​​

Spring 集成的 Redis 模块提供了 . 以下示例演示如何将其与聚合器一起使用：​​RedisMessageStore​​

<bean id="redisMessageStore" class="o.s.i.redis.store.RedisMessageStore">
    <constructor-arg ref="redisConnectionFactory"/>
</bean>

<int:aggregator input-channel="inputChannel" output-channel="outputChannel"
         message-store="redisMessageStore"/>

前面的示例是一个 Bean 配置，它需要 a 作为构造函数参数。​​RedisConnectionFactory​​

缺省情况下，使用 Java 序列化来序列化消息。 但是，如果要使用其他序列化技术（如 JSON），可以通过设置 .​​RedisMessageStore​​​​valueSerializer​​​​RedisMessageStore​​

从版本 4.3.10 开始，框架分别为实例和实例提供了 Jackson 序列化程序和解序列化程序实现 — 和 。 必须使用的选项配置它们。 此外，还应设置为每个序列化的复杂对象添加类型信息（如果您信任源）。 然后在反序列化期间使用该类型信息。 该框架提供了一个名为 的实用程序方法，该方法已随前面提到的所有属性和序列化程序一起提供。 此实用程序方法附带一个参数，用于限制 Java 包进行反序列化以避免安全漏洞。 默认的受信任软件包：、、。 若要在 中管理 JSON 序列化，必须以类似于以下示例的方式对其进行配置：​​Message​​​​MessageHeaders​​​​MessageJacksonDeserializer​​​​MessageHeadersJacksonSerializer​​​​SimpleModule​​​​ObjectMapper​​​​enableDefaultTyping​​​​ObjectMapper​​​​JacksonJsonUtils.messagingAwareMapper()​​​​trustedPackages​​​​java.util​​​​java.lang​​​​org.springframework.messaging.support​​​​org.springframework.integration.support​​​​org.springframework.integration.message​​​​org.springframework.integration.store​​​​RedisMessageStore​​

RedisMessageStore store = new RedisMessageStore(redisConnectionFactory);
ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper();
RedisSerializer<Object> serializer = new GenericJackson2JsonRedisSerializer(mapper);
store.setValueSerializer(serializer);

从版本 4.3.12 开始，支持允许区分同一 Redis 服务器上的存储实例的选项。​​RedisMessageStore​​​​prefix​​

Redis 通道消息存储库

前面显示的内容将每个组维护为单个键（组 ID）下的值。 虽然您可以使用它来支持持久性，但为此目的提供了一个专用的（从版本 4.0 开始）。 此存储在每个通道、发送消息和接收消息时使用 。 默认情况下，此存储也使用 JDK 序列化，但您可以修改值序列化程序，如前所述。​​RedisMessageStore​​​​QueueChannel​​​​RedisChannelMessageStore​​​​LIST​​​​LPUSH​​​​RPOP​​

我们建议使用此存储支持通道，而不是使用常规 . 以下示例定义了一个 Redis 消息存储库，并在具有队列的通道中使用它：​​RedisMessageStore​​

<bean id="redisMessageStore" class="o.s.i.redis.store.RedisChannelMessageStore">
    <constructor-arg ref="redisConnectionFactory"/>
</bean>

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

用于存储数据的键的格式为：（在前面的示例中，）。​​<storeBeanName>:<channelId>​​​​redisMessageStore:somePersistentQueueChannel​​

此外，还提供了子类。 当您将其与 一起使用时，将按 （FIFO） 优先级顺序接收消息。 它使用标准标头并支持优先级值 （）。 具有其他优先级的消息（以及没有优先级的消息）将按FIFO顺序在任何具有优先级的消息之后检索。​​RedisChannelPriorityMessageStore​​​​QueueChannel​​​​IntegrationMessageHeaderAccessor.PRIORITY​​​​0 - 9​​

红地元数据存储

Spring Integration 3.0引入了一个新的基于Redis的MetadataStore（参见Metadata Store）实现。 可以使用 来维护跨应用程序重新启动的状态。 您可以将此新实现与适配器一起使用，例如：​​RedisMetadataStore​​​​MetadataStore​​​​MetadataStore​​

• 饲料
• 文件
• 邮票
• 自来水龙

要指示这些适配器使用 new ，请声明一个名为 的 Spring Bean。 源入站通道适配器和源入站通道适配器都自动拾取并使用声明的 . 下面的示例演示如何声明这样的 Bean：​​RedisMetadataStore​​​​metadataStore​​​​RedisMetadataStore​​

<bean name="metadataStore" class="o.s.i.redis.store.metadata.RedisMetadataStore">
    <constructor-arg name="connectionFactory" ref="redisConnectionFactory"/>
</bean>

由RedisProperties支持。 与它的交互使用 BoundHashOperations，这反过来又需要整个存储区。 在 的情况下，这扮演了一个区域的角色，这在分布式环境中很有用，当多个应用程序使用相同的 Redis 服务器时。 缺省情况下，此值为 。​​RedisMetadataStore​​​​key​​​​Properties​​​​MetadataStore​​​​key​​​​key​​​​MetaData​​

从版本 4.0 开始，此存储实现了 ，使其在多个应用程序实例之间可靠地共享，其中只允许一个实例存储或修改键的值。​​ConcurrentMetadataStore​​

Redis 存储入站通道适配器

Redis 存储入站通道适配器是一个轮询使用者，它从 Redis 集合读取数据并将其作为有效负载发送。 以下示例演示如何配置 Redis 存储入站通道适配器：​​Message​​

<int-redis:store-inbound-channel-adapter id="listAdapter"
    connection-factory="redisConnectionFactory"
    key="myCollection"
    channel="redisChannel"
    collection-type="LIST" >
    <int:poller fixed-rate="2000" max-messages-per-poll="10"/>
</int-redis:store-inbound-channel-adapter>

前面的示例演示如何使用元素配置 Redis 存储入站通道适配器，为各种属性提供值，例如：​​store-inbound-channel-adapter​​

• ​​key​​或：正在使用的集合的密钥的名称。key-expression
• ​​collection-type​​：此适配器支持的集合类型的枚举。 支持的集合包括 、、 、 和 。LISTSETZSETPROPERTIESMAP
• ​​connection-factory​​：对 实例的引用。o.s.data.redis.connection.RedisConnectionFactory
• ​​redis-template​​：对 实例的引用。o.s.data.redis.core.RedisTemplate
• 所有其他入站适配器通用的其他属性（例如“通道”）。

<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate">
    <property name="connectionFactory" ref="redisConnectionFactory"/>
    <property name="keySerializer">
        <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
    </property>
    <property name="hashKeySerializer">
        <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/>
    </property>
</bean>

因为它具有 的文本值，所以前面的示例相对简单且静态。 有时，您可能需要根据某些条件在运行时更改密钥的值。 为此，请改用，其中提供的表达式可以是任何有效的 SpEL 表达式。​​key​​​​key-expression​​

此外，您可能希望对从 Redis 集合读取的成功处理数据执行一些后处理。 例如，您可能希望在处理完值后移动或删除该值。 您可以使用 Spring Integration 2.2 中添加的事务同步功能来执行此操作。 以下示例使用和事务同步：​​key-expression​​

<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScoreAndSynchronization"
        connection-factory="redisConnectionFactory"
        key-expression="'presidents'"
        channel="otherRedisChannel"
        auto-startup="false"
        collection-type="ZSET">
            <int:poller fixed-rate="1000" max-messages-per-poll="2">
                <int:transactional synchronization-factory="syncFactory"/>
            </int:poller>
</int-redis:store-inbound-channel-adapter>

<int:transaction-synchronization-factory id="syncFactory">
  <int:after-commit expression="payload.removeByScore(18, 18)"/>
</int:transaction-synchronization-factory>

<bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>

可以使用元素将轮询器声明为事务性轮询器。 此元素可以引用真正的事务管理器（例如，如果流的其他部分调用 JDBC）。 如果您没有“真实”事务，则可以使用 ，这是 Spring 的实现，可以在没有实际事务时使用 Redis 适配器的事务同步功能。​​transactional​​​​o.s.i.transaction.PseudoTransactionManager​​​​PlatformTransactionManager​​

轮询器是事务性的后，您可以设置 on 元素的实例。 创建 的实例。 为方便起见，我们公开了一个默认的基于 SpEL 的 ，它允许您配置 SpEL 表达式，其执行与事务协调（同步）。 支持提交前、提交后和回滚后的表达式，以及发送评估结果（如果有）的通道（每种事件一个）。 对于每个子元素，可以指定 和 属性。 如果仅存在该属性，则接收到的消息将作为特定同步方案的一部分发送到那里。 如果仅存在该属性，并且表达式的结果为非 null 值，则会生成一条消息，并将结果作为有效负载发送到默认通道 （），并显示在日志中（在级别）。 如果您希望评估结果转到特定渠道，请添加属性。 如果表达式的结果为 null 或 void，则不会生成任何消息。​​o.s.i.transaction.TransactionSynchronizationFactory​​​​transactional​​​​TransactionSynchronizationFactory​​​​TransactionSynchronization​​​​TransactionSynchronizationFactory​​​​expression​​​​channel​​​​channel​​​​expression​​​​NullChannel​​​​DEBUG​​​​channel​​

有关事务同步的详细信息，请参阅事务同步。

RedisStore 出站通道适配器

RedisStore 出站通道适配器允许您将消息负载写入 Redis 集合，如以下示例所示：

<int-redis:store-outbound-channel-adapter id="redisListAdapter"
          collection-type="LIST"
          channel="requestChannel"
          key="myCollection" />

前面的配置 Redis 存储出站通道适配器使用该元素。 它为各种属性提供值，例如：​​store-inbound-channel-adapter​​

• ​​key​​或：正在使用的集合的密钥的名称。key-expression
• ​​extract-payload-elements​​：如果设置为 （默认值） 并且有效负载是“多值”对象（即 a 或 a）的实例，则使用 “addAll” 和 “putAll” 语义存储该对象。 否则，如果设置为 ，则有效负载将存储为单个条目，而不管其类型如何。 如果有效负载不是“多值”对象的实例，则忽略此属性的值，并且有效负载始终存储为单个条目。trueCollectionMapfalse
• ​​collection-type​​：此适配器支持的类型的枚举。 支持的集合包括 、、 、 和 。CollectionLISTSETZSETPROPERTIESMAP
• ​​map-key-expression​​：返回所存储条目的密钥名称的 SpEL 表达式。 仅当 is or 和 'extract-payload-elements' 为 false 时，它才适用。collection-typeMAPPROPERTIES
• ​​connection-factory​​：对 实例的引用。o.s.data.redis.connection.RedisConnectionFactory
• ​​redis-template​​：对 实例的引用。o.s.data.redis.core.RedisTemplate
• 所有其他入站适配器通用的其他属性（例如“通道”）。

由于它具有 和其他属性的文本值，因此前面的示例相对简单且静态。 有时，您可能需要根据某些条件在运行时动态更改值。 为此，请使用它们的等效项（、 等），其中提供的表达式可以是任何有效的 SpEL 表达式。​​key​​​​-expression​​​​key-expression​​​​map-key-expression​​

Redis 出站命令网关

Spring Integration 4.0 引入了 Redis 命令网关，允许您使用通用方法执行任何标准的 Redis 命令。 以下清单显示了 Redis 出站网关的可用属性：​​RedisConnection#execute​​

<int-redis:outbound-gateway
        request-channel=""  
        reply-channel=""  
        requires-reply=""  
        reply-timeout=""  
        connection-factory=""  
        redis-template=""  
        arguments-serializer=""  
        command-expression=""  
        argument-expressions=""  
        use-command-variable=""  
        arguments-strategy="" />

您可以将 用作通用组件来执行任何所需的 Redis 操作。 以下示例演示如何从 Redis 原子序数获取递增值：​​<int-redis:outbound-gateway>​​

<int-redis:outbound-gateway request-channel="requestChannel"
    reply-channel="replyChannel"
    command-expression="'INCR'"/>

有效负载应具有 的名称 ，该名称可能由 Bean 定义提供。​​Message​​​​redisCounter​​​​org.springframework.data.redis.support.atomic.RedisAtomicInteger​​

该方法具有泛型作为其返回类型。 实际结果取决于命令类型。 例如，返回一个 . 有关命令及其参数和结果类型的更多信息，请参阅 Redis 规范。​​RedisConnection#execute​​​​Object​​​​MGET​​​​List<byte[]>​​

Redis 队列出站网关

Spring 集成引入了 Redis 队列出站网关来执行请求和回复场景。 它将对话推送到提供的 ，将该值作为其键推送到 Redis 列表，并等待来自键为 . 每个交互使用不同的 UUID。 以下清单显示了 Redis 出站网关的可用属性：​​UUID​​​​queue​​​​UUID​​​​UUID' plus '.reply​​

<int-redis:queue-outbound-gateway
        request-channel=""  
        reply-channel=""  
        requires-reply=""  
        reply-timeout=""  
        connection-factory=""  
        queue=""  
        order=""  
        serializer=""  
        extract-payload=""/>

Redis 队列入站网关

Spring Integration 4.1 引入了 Redis 队列入站网关来执行请求和回复场景。 它从提供的 中弹出一个对话，从 Redis 列表中弹出值作为其键，并使用加号键将回复推送到 Redis 列表。 以下清单显示了 Redis 队列入站网关的可用属性：​​UUID​​​​queue​​​​UUID​​​​UUID​​​​.reply​​

<int-redis:queue-inbound-gateway
        request-channel=""  
        reply-channel=""  
        executor=""  
        reply-timeout=""  
        connection-factory=""  
        queue=""  
        order=""  
        serializer=""  
        receive-timeout=""  
        expect-message=""  
        recovery-interval=""/>

Redis 流出站通道适配器

Spring Integration 5.4 引入了反应式 Redis 流出站通道适配器，用于将消息负载写入 Redis 流。 出站通道适配器用于向流中添加 。 以下示例演示如何将 Java 配置和服务类用于 Redis 流出站通道适配器。​​ReactiveStreamOperations.add(…)​​​​Record​​

@Bean
@ServiceActivator(inputChannel = "messageChannel")
public ReactiveRedisStreamMessageHandler reactiveValidatorMessageHandler(
        ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
    ReactiveRedisStreamMessageHandler reactiveStreamMessageHandler =
        new ReactiveRedisStreamMessageHandler(reactiveRedisConnectionFactory, "myStreamKey"); 
    reactiveStreamMessageHandler.setSerializationContext(serializationContext); 
    reactiveStreamMessageHandler.setHashMapper(hashMapper); 
    reactiveStreamMessageHandler.setExtractPayload(true); 
    return reactiveStreamMessageHandler;
}

Redis 流入站通道适配器

Spring Integration 5.4 引入了 Reactive Stream 入站通道适配器，用于从 Redis 流读取消息。 入站通道适配器使用 或基于自动确认标志从 Redis 流读取记录。 以下示例演示如何将 Java 配置用于 Redis 流入站通道适配器。​​StreamReceiver.receive(…)​​​​StreamReceiver.receiveAutoAck()​​

@Bean
public ReactiveRedisStreamMessageProducer reactiveRedisStreamProducer(
       ReactiveRedisConnectionFactory reactiveRedisConnectionFactory) {
ReactiveRedisStreamMessageProducer messageProducer =
            new ReactiveRedisStreamMessageProducer(reactiveRedisConnectionFactory, "myStreamKey"); 
    messageProducer.setStreamReceiverOptions( 
                StreamReceiver.StreamReceiverOptions.builder()
                      .pollTimeout(Duration.ofMillis(100))
                      .build());
    messageProducer.setAutoStartup(true); 
    messageProducer.setAutoAck(false); 
    messageProducer.setCreateConsumerGroup(true); 
    messageProducer.setConsumerGroup("my-group"); 
    messageProducer.setConsumerName("my-consumer"); 
    messageProducer.setOutputChannel(fromRedisStreamChannel); 
    messageProducer.setReadOffset(ReadOffset.latest()); 
    messageProducer.extractPayload(true); 
    return messageProducer;
}

如果设置为 ，则 Redis 驱动程序不会自动确认 Redis 流中的标头，而是将标头添加到消息中，以实例作为值生成。 目标集成流的责任是在基于此类记录为消息完成业务逻辑时调用其回调。 即使在反序列化期间发生异常并进行了配置，也需要类似的逻辑。 因此，目标错误处理程序必须决定确认或纳克此类失败的消息。 与 一起，还将这些标头填充到消息中以生成：、 和 。​​autoAck​​​​false​​​​Record​​​​IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK​​​​SimpleAcknowledgment​​​​acknowledge()​​​​errorChannel​​​​IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK​​​​ReactiveRedisStreamMessageProducer​​​​RedisHeaders.STREAM_KEY​​​​RedisHeaders.STREAM_MESSAGE_ID​​​​RedisHeaders.CONSUMER_GROUP​​​​RedisHeaders.CONSUMER​​

从版本 5.5 开始，您可以在 上显式配置选项，包括新引入的函数，如果 Redis Stream 使用者在发生反序列化错误时应继续轮询，则需要该函数。 默认函数将消息发送到错误通道（如果提供），并可能确认失败的消息，如上所述。 所有这些都与外部提供的.​​StreamReceiver.StreamReceiverOptionsBuilder​​​​ReactiveRedisStreamMessageProducer​​​​onErrorResume​​​​StreamReceiver.StreamReceiverOptionsBuilder​​​​StreamReceiver.StreamReceiverOptions​​

瑞迪斯锁注册表

Spring Integration 4.0 引入了 . 某些组件（例如，聚合器和重新排序器）使用从实例获取的锁来确保一次只有一个线程操作一个组。 在单个组件中执行此功能。 您现在可以在这些组件上配置外部锁定注册表。 当您将它与共享 一起使用时，您可以使用 跨多个应用程序实例提供此功能，以便一次只有一个实例可以操作该组。​​RedisLockRegistry​​​​LockRegistry​​​​DefaultLockRegistry​​​​MessageGroupStore​​​​RedisLockRegistry​​

当一个本地线程释放一个锁时，另一个本地线程通常可以立即获取该锁。 如果线程使用不同的注册表实例释放锁，则最多可能需要 100 毫秒才能获取锁。

若要避免“挂起”锁（当服务器发生故障时），此注册表中的锁将在默认的 60 秒后过期，但您可以在注册表上配置此值。 锁的持有时间通常要短得多。

从版本 5.0 开始，实现 ，这将删除上次获取的锁，这些锁当前未锁定。​​RedisLockRegistry​​​​ExpirableLockRegistry​​​​age​​

5.5.6 版本的字符串，支持通过 . 有关更多信息，请参阅其 JavaDocs。​​RedisLockRegistry​​​​RedisLockRegistry.locks​​​​RedisLockRegistry.setCacheCapacity()​​

字符串 对于版本 5.5.13，公开了一个选项，用于确定应在哪种模式下进行 Redis 锁获取：​​RedisLockRegistry​​​​setRedisLockType(RedisLockType)​​

• ​​RedisLockType.SPIN_LOCK​​- 锁通过周期循环（100ms）获取，检查是否可以获取锁。 违约。
• ​​RedisLockType.PUB_SUB_LOCK​​- 锁是通过 redis 发布订阅获取的。

发布-订阅是首选模式 - 客户端 Redis 服务器之间的网络抖动更少，性能更高 - 当订阅收到有关在其他进程中解锁的通知时，会立即获取锁定。 但是，Redis 不支持主/副本连接中的发布-订阅（例如在 AWS ElastiCache 环境中），因此选择繁忙旋转模式作为默认值，以使注册表在任何环境中工作。